diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index 14f8be56be..6d4eec9af8 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -9,7 +9,6 @@ from __future__ import annotations import json import logging -import random import re import time from typing import Any, Dict, List, Optional @@ -18,13 +17,11 @@ from agent.codex_responses_adapter import _summarize_user_message_for_log from agent.conversation_compression import ( conversation_history_after_compression, # noqa: F401 — resolved lazily by turn_overflow/turn_preflight/turn_recovery (tests patch it here) ) -from agent.display import KawaiiSpinner from agent.fast_mode import begin_turn as begin_fast_mode_turn from agent.message_metadata import append_message from agent.turn_context import ( PreflightCompressionTimedOut, build_turn_context, - reanchor_current_turn_user_idx, ) from agent.turn_retry_state import TurnRetryState from agent.runtime_cwd import resolve_agent_cwd @@ -61,11 +58,16 @@ from agent.turn_recovery import ( # noqa: F401 — resolved lazily by agent.tur # Bind before the turn starts so a source-tree swap cannot load a skewed # finalizer at turn end. from agent.turn_finalizer import finalize_turn -from agent.turn_iteration_prep import prepare_iteration +from agent.turn_iteration_prep import ( + announce_api_call, + apply_retry_restarts, + begin_iteration, + prepare_iteration, +) from agent.turn_preflight_gate import run_preflight_gate from agent.turn_request_assembly import assemble_api_request from agent.turn_api_request import build_api_request -from agent.turn_api_call import perform_api_call +from agent.turn_api_call import handle_api_interrupt, nous_rate_limit_guard, perform_api_call from agent.turn_response_check import check_api_response from agent.turn_api_error import handle_api_error from agent.turn_final_response import finish_text_response @@ -1542,6 +1544,55 @@ def _notify_context_engine_turn_complete( ) +def _decode_inline_moa_turn(user_message, persist_user_message): + """Decode a MoA preset encoded into ``user_message`` (``hermes_cli.moa_config``). + + Returns ``(user_message, moa_config, persist_user_message)``; unchanged with + ``moa_config=None`` when nothing is encoded or decoding fails.""" + try: + from hermes_cli.moa_config import decode_moa_turn + + _decoded_message, _decoded_moa_config = decode_moa_turn(user_message) + if _decoded_moa_config is not None: + if persist_user_message is None: + persist_user_message = _decoded_message + return _decoded_message, _decoded_moa_config, persist_user_message + except Exception: + pass + return user_message, None, persist_user_message + + +def _preflight_timeout_result(agent, exc, conversation_history) -> Dict[str, Any]: + """Typed recovery result when turn-start preflight compression timed out (#98424): + no provider call was sent. Surfaces hide raw exception text, which would bury the + actionable guidance and skip the compression_exhausted recovery contract.""" + logger.warning( + "Turn-start preflight compression timed out — ending turn with " + "typed recovery result: %s", + exc, + ) + # Clear the tripwire slot note_turn_start registered; the early return skips the + # persist funnel that clears it. The user row is deliberately NOT persisted: + # the gateway skips persistence for compression_exhausted results (#7100). + from agent.agent_runtime_helpers import note_turn_persisted + + note_turn_persisted(agent) + # Not _COMPRESSION_TIMEOUT_FINAL_RESPONSE — that describes a different state + # (compression ran, could not reduce); the exception text carries the guidance. + _final_response = str(exc) + return { + "final_response": _final_response, + "messages": list(conversation_history or []), + "completed": False, + "api_calls": 0, + "error": _final_response, + "partial": True, + "failed": True, + "compression_exhausted": True, + "turn_exit_reason": "context_compression_timeout", + } + + def run_conversation( agent, user_message: Any, @@ -1568,17 +1619,9 @@ def run_conversation( Returns: dict with the final response and message history.""" if moa_config is None: - try: - from hermes_cli.moa_config import decode_moa_turn - - _decoded_message, _decoded_moa_config = decode_moa_turn(user_message) - if _decoded_moa_config is not None: - user_message = _decoded_message - moa_config = _decoded_moa_config - if persist_user_message is None: - persist_user_message = _decoded_message - except Exception: - pass + user_message, moa_config, persist_user_message = _decode_inline_moa_turn( + user_message, persist_user_message + ) # The gateway caches agents across turns; compression state is per-turn, or a stale # in-place boundary would make a later uncompressed result look compacted. @@ -1622,34 +1665,7 @@ def run_conversation( moa_active=bool(moa_config), ) except PreflightCompressionTimedOut as _preflight_timeout_exc: - # Preflight compression timed out; no provider call sent (#98424). Return the - # typed recovery result: surfaces hide raw exception text, which would bury the - # actionable guidance and skip the compression_exhausted recovery contract. - logger.warning( - "Turn-start preflight compression timed out — ending turn with " - "typed recovery result: %s", - _preflight_timeout_exc, - ) - # Clear the tripwire slot note_turn_start registered; the early return skips the - # persist funnel that clears it. The user row is deliberately NOT persisted: - # the gateway skips persistence for compression_exhausted results (#7100). - from agent.agent_runtime_helpers import note_turn_persisted - - note_turn_persisted(agent) - # Not _COMPRESSION_TIMEOUT_FINAL_RESPONSE — that describes a different state - # (compression ran, could not reduce); the exception text carries the guidance. - _final_response = str(_preflight_timeout_exc) - return { - "final_response": _final_response, - "messages": list(conversation_history or []), - "completed": False, - "api_calls": 0, - "error": _final_response, - "partial": True, - "failed": True, - "compression_exhausted": True, - "turn_exit_reason": "context_compression_timeout", - } + return _preflight_timeout_result(agent, _preflight_timeout_exc, conversation_history) user_message = _ctx.user_message original_user_message = _ctx.original_user_message messages = _ctx.messages @@ -1733,52 +1749,20 @@ def run_conversation( ) while (api_call_count < agent.max_iterations and agent.iteration_budget.remaining > 0) or agent._budget_grace_call: - _redirect_text = agent._drain_pending_redirect() - if _redirect_text: - _apply_active_turn_redirect(agent, messages, _redirect_text) - if isinstance(original_user_message, str): - original_user_message = ( - f"{original_user_message}\n\n" - f"User correction during the turn: {_redirect_text}" - ) - agent._persist_session(messages, conversation_history) - - # Reset per-turn checkpoint dedup so each iteration can take one snapshot - agent._checkpoint_mgr.new_turn() - - # Check for interrupt request (e.g., user sent new message) - if agent._interrupt_requested: - interrupted = True - _turn_exit_reason = "interrupted_by_user" - if not agent.quiet_mode: - agent._safe_print("\n⚡ Breaking out of tool loop due to interrupt...") - break - - # Aggregate input budget for detached auxiliary forks: bounds the whole review, - # not each request. Checked between iterations so the crossing request's writes - # have landed, mirroring the iteration-budget exit (#93057). - if _review_input_budget_exhausted(agent): - _turn_exit_reason = "review_input_budget_exhausted" - if not agent.quiet_mode: - agent._safe_print( - f"\n⏹️ Review input budget exhausted " - f"({int(agent.session_input_tokens):,} tokens) — stopping " - f"the review tool loop before the next provider call." - ) - break - - api_call_count += 1 - agent._api_call_count = api_call_count - agent._touch_activity(f"starting API call #{api_call_count}") - - # Grace call: budget exhausted but the model gets one more call. Consume the - # flag so the loop exits after this iteration regardless of outcome. - if agent._budget_grace_call: - agent._budget_grace_call = False - elif not agent.iteration_budget.consume(): - _turn_exit_reason = "budget_exhausted" - if not agent.quiet_mode: - agent._safe_print(f"\n⚠️ Iteration budget exhausted ({agent.iteration_budget.used}/{agent.iteration_budget.max_total} iterations used)") + _it = begin_iteration( + agent, + messages=messages, + conversation_history=conversation_history, + original_user_message=original_user_message, + api_call_count=api_call_count, + interrupted=interrupted, + _turn_exit_reason=_turn_exit_reason, + ) + original_user_message = _it.original_user_message + api_call_count = _it.api_call_count + interrupted = _it.interrupted + _turn_exit_reason = _it._turn_exit_reason + if _it.action == "break": break _ip = prepare_iteration( @@ -1851,33 +1835,15 @@ def run_conversation( if _pg.action == "continue": continue - # Thinking spinner for quiet mode (animated during API call) - thinking_spinner = None - - if not agent.quiet_mode: - agent._vprint(f"\n{agent.log_prefix}🔄 Making API call #{api_call_count}/{agent.max_iterations}...") - agent._vprint(f"{agent.log_prefix} 📊 Request size: {len(api_messages)} messages, ~{approx_tokens:,} tokens (~{total_chars:,} chars)") - agent._vprint(f"{agent.log_prefix} 🔧 Available tools: {len(agent.tools) if agent.tools else 0}") - else: - # Animated thinking spinner in quiet mode - face = random.choice(KawaiiSpinner.get_thinking_faces()) - verb = random.choice(KawaiiSpinner.get_thinking_verbs()) - if agent.thinking_callback: - # CLI TUI mode: use prompt_toolkit widget instead of raw spinner - # (works in both streaming and non-streaming modes) - agent.thinking_callback(f"{face} {verb}...") - elif not agent._has_stream_consumers() and agent._should_start_quiet_spinner(): - # Raw KawaiiSpinner only when no streaming consumers and the - # spinner output has a safe sink. - spinner_type = random.choice(['brain', 'sparkle', 'pulse', 'moon', 'star']) - thinking_spinner = KawaiiSpinner(f"{face} {verb}...", spinner_type=spinner_type, print_fn=agent._print_fn) - thinking_spinner.start() - - # Log request details if verbose - if agent.verbose_logging: - logging.debug(f"API Request - Model: {agent.model}, Messages: {len(messages)}, Tools: {len(agent.tools) if agent.tools else 0}") - logging.debug(f"Last message role: {messages[-1]['role'] if messages else 'none'}") - logging.debug(f"Total message size: ~{approx_tokens:,} tokens") + _an = announce_api_call( + agent, + messages=messages, + api_messages=api_messages, + api_call_count=api_call_count, + approx_tokens=approx_tokens, + total_chars=total_chars, + ) + thinking_spinner = _an.thinking_spinner api_start_time = time.time() retry_count = 0 @@ -1891,52 +1857,24 @@ def run_conversation( agent._current_api_request_id = api_request_id while retry_count < max_retries: - # ── Nous Portal rate limit guard ────────────────────── - # Skip the call if another session recorded a rate limit: every attempt - # (incl. SDK retries) counts against RPH. - if agent.provider == "nous": - try: - from agent.nous_rate_guard import ( - nous_rate_limit_remaining, - format_remaining as _fmt_nous_remaining, - ) - _nous_remaining = nous_rate_limit_remaining() - if _nous_remaining is not None and _nous_remaining > 0: - _nous_msg = ( - f"Nous Portal rate limit active — " - f"resets in {_fmt_nous_remaining(_nous_remaining)}." - ) - agent._buffer_vprint( - f"⏳ {_nous_msg} Trying fallback..." - ) - agent._buffer_status(f"⏳ {_nous_msg}") - if agent._try_activate_fallback(): - active_system_prompt = _arm_fallback_restart( - agent, api_messages, active_system_prompt, _retry) - retry_count = 0 - compression_attempts = 0 - break - # No fallback available — surface buffered context - # so user sees the rate-limit message that led here. - agent._flush_status_buffer() - agent._persist_session(messages, conversation_history) - return { - "final_response": ( - f"⏳ {_nous_msg}\n\n" - "No fallback provider available. " - "Try again after the reset, or add a " - "fallback provider in config.yaml." - ), - "messages": messages, - "api_calls": api_call_count, - "completed": False, - "failed": True, - "error": _nous_msg, - } - except ImportError: - pass - except Exception: - pass # Never let rate guard break the agent loop + _ng = nous_rate_limit_guard( + agent, + _retry=_retry, + api_messages=api_messages, + messages=messages, + conversation_history=conversation_history, + active_system_prompt=active_system_prompt, + retry_count=retry_count, + compression_attempts=compression_attempts, + api_call_count=api_call_count, + ) + active_system_prompt = _ng.active_system_prompt + retry_count = _ng.retry_count + compression_attempts = _ng.compression_attempts + if _ng.action == "return": + return _ng.result + if _ng.action == "break": + break try: _rq = build_api_request( @@ -2031,33 +1969,21 @@ def run_conversation( continue except InterruptedError: - if thinking_spinner: - thinking_spinner.stop("") - thinking_spinner = None - if agent.thinking_callback: - agent.thinking_callback("") - if agent._has_pending_redirect(): - # redirect() cancelled only this request: keep the correction - # queued, clear the cancellation bit, let the outer loop rebuild. - # Never materialize incomplete signed/encrypted reasoning items. - if agent.clear_interrupt(preserve_redirect=True): - _retry.restart_with_redirected_messages = True - break - api_elapsed = time.time() - api_start_time - agent._vprint(f"{agent.log_prefix}⚡ Interrupted during API call.", force=True) - interrupted = True - # Keep assistant text already streamed before the stop, else the next - # turn has no record of the half-finished reply. - _partial = agent._strip_think_blocks( - getattr(agent, "_current_streamed_assistant_text", "") or "" - ).strip() - if _partial: - append_message(messages, {"role": "assistant", "content": _partial}) - final_response = _partial - else: - final_response = f"{INTERRUPT_WAITING_FOR_MODEL_PREFIX}{api_elapsed:.1f}s elapsed)." - agent._persist_session(messages, conversation_history) - break + _ai = handle_api_interrupt( + agent, + _retry=_retry, + thinking_spinner=thinking_spinner, + messages=messages, + conversation_history=conversation_history, + api_start_time=api_start_time, + interrupted=interrupted, + final_response=final_response, + ) + thinking_spinner = _ai.thinking_spinner + interrupted = _ai.interrupted + final_response = _ai.final_response + if _ai.action == "break": + break except Exception as api_error: _ae = handle_api_error( @@ -2099,78 +2025,34 @@ def run_conversation( if _ae.action == "continue": continue - if _retry.restart_with_redirected_messages: - # Cancelled request produced no valid assistant item: reuse the same logical - # iteration after the outer loop appends partial context + correction. - api_call_count -= 1 - agent.iteration_budget.refund() - _retry.restart_with_redirected_messages = False - continue - - # If the API call was interrupted, skip response processing - if interrupted: - _turn_exit_reason = "interrupted_during_api_call" + _rs = apply_retry_restarts( + agent, + _retry=_retry, + response=response, + interrupted=interrupted, + messages=messages, + conversation_history=conversation_history, + user_message=user_message, + api_kwargs=api_kwargs, + current_turn_user_idx=current_turn_user_idx, + final_response=final_response, + retry_count=retry_count, + api_call_count=api_call_count, + length_continue_retries=length_continue_retries, + _preflight_compression_blocked=_preflight_compression_blocked, + _turn_exit_reason=_turn_exit_reason, + ) + current_turn_user_idx = _rs.current_turn_user_idx + final_response = _rs.final_response + retry_count = _rs.retry_count + api_call_count = _rs.api_call_count + _preflight_compression_blocked = _rs._preflight_compression_blocked + _turn_exit_reason = _rs._turn_exit_reason + if _rs.action == "break": break - - if _retry.restart_with_compressed_messages: - api_call_count -= 1 - agent.iteration_budget.refund() - # Compression restarts count toward the retry limit so a compression that - # shrinks messages but not enough can't loop forever. - retry_count += 1 - _retry.restart_with_compressed_messages = False - if _should_skip_model_call_for_reference_handoff( - messages, user_message - ): - logger.info( - "Skipping compressed-restart model call: reference-only " - "handoff would be the sole active user turn (#80622)" - ) - if not final_response: - final_response = _HANDOFF_SKIP_FINAL_RESPONSE - _turn_exit_reason = "compaction_handoff_not_actionable" - break - # In-loop compression rebuilt `messages`; re-anchor the current-turn index - # like the prologue, AFTER the handoff guard (it may re-append this turn's - # ask). A stale anchor injects prefetch into a historical row. - current_turn_user_idx = reanchor_current_turn_user_idx( - messages, user_message - ) - agent._persist_user_message_idx = current_turn_user_idx + if _rs.action == "continue": continue - if _retry.restart_with_rebuilt_messages: - # A stall/failure escalated to the fallback chain: re-issue against the - # active fallback provider, refunding budget/count for the stalled attempt. - api_call_count -= 1 - agent.iteration_budget.refund() - _retry.restart_with_rebuilt_messages = False - # Failover shrank the compressor window: clear the preflight block so - # preflight re-runs before the first fallback call. Hoisted to the single - # consumer. (#84733) - _preflight_compression_blocked = False - continue - - if _retry.restart_with_length_continuation: - # Boost output budget per retry: 2×, 4×, 8×, 16× base, capped at 32 768, via - # _ephemeral_max_output_tokens. Keep a larger original provider/model - # default as the floor so retries never downshift. - _boost_base = agent.max_tokens if agent.max_tokens else 4096 - _boost = _boost_base * (2 ** length_continue_retries) - _requested_cap = agent._requested_output_cap_from_api_kwargs(api_kwargs) - if _requested_cap is not None: - _boost = max(_boost, _requested_cap) - _boost_cap = max(32768, _requested_cap or 0) - agent._ephemeral_max_output_tokens = min(_boost, _boost_cap) - continue - - # All retries may exhaust with `response` still None; break out cleanly. - if response is None: - _turn_exit_reason = "all_retries_exhausted_no_response" - print(f"{agent.log_prefix}❌ All API retries exhausted with no successful response.") - agent._persist_session(messages, conversation_history) - break - try: _ri = normalize_model_response( agent, diff --git a/agent/turn_api_call.py b/agent/turn_api_call.py index 4ea2c85de1..58c0af546d 100644 --- a/agent/turn_api_call.py +++ b/agent/turn_api_call.py @@ -6,10 +6,12 @@ from ``run_conversation``; nothing here imports ``agent.conversation_loop`` at m from __future__ import annotations -import logging from dataclasses import dataclass +import logging +import time from typing import Any, Dict, Optional +from agent.message_metadata import append_message logger = logging.getLogger("agent.conversation_loop") @@ -179,3 +181,159 @@ def perform_api_call( interrupted = True return _verdict("break") return _verdict("fallthrough") + + +@dataclass +class ApiInterruptVerdict: + """Always ``action == "break"`` (leave the retry loop): either a redirect restart was + armed on ``_retry`` or the turn is ``interrupted`` with ``final_response`` set.""" + + action: str + thinking_spinner: Any + interrupted: Any + final_response: Any + + +def handle_api_interrupt( + agent: Any, + *, + _retry: Any, + thinking_spinner: Any, + messages: Any, + conversation_history: Any, + api_start_time: Any, + interrupted: Any, + final_response: Any, +) -> ApiInterruptVerdict: + """``InterruptedError`` during the provider call: a pending redirect keeps its correction + queued for the outer-loop rebuild; otherwise keep any streamed partial text so the next + turn has a record of the half-finished reply.""" + from agent.conversation_loop import ( + INTERRUPT_WAITING_FOR_MODEL_PREFIX, + ) + + def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> ApiInterruptVerdict: + return ApiInterruptVerdict( + action=action, + thinking_spinner=thinking_spinner, + interrupted=interrupted, + final_response=final_response, + + ) + + if thinking_spinner: + thinking_spinner.stop("") + thinking_spinner = None + if agent.thinking_callback: + agent.thinking_callback("") + if agent._has_pending_redirect(): + # redirect() cancelled only this request: keep the correction + # queued, clear the cancellation bit, let the outer loop rebuild. + # Never materialize incomplete signed/encrypted reasoning items. + if agent.clear_interrupt(preserve_redirect=True): + _retry.restart_with_redirected_messages = True + return _verdict("break") + api_elapsed = time.time() - api_start_time + agent._vprint(f"{agent.log_prefix}⚡ Interrupted during API call.", force=True) + interrupted = True + # Keep assistant text already streamed before the stop, else the next + # turn has no record of the half-finished reply. + _partial = agent._strip_think_blocks( + getattr(agent, "_current_streamed_assistant_text", "") or "" + ).strip() + if _partial: + append_message(messages, {"role": "assistant", "content": _partial}) + final_response = _partial + else: + final_response = f"{INTERRUPT_WAITING_FOR_MODEL_PREFIX}{api_elapsed:.1f}s elapsed)." + agent._persist_session(messages, conversation_history) + return _verdict("break") + return _verdict("fallthrough") + + +@dataclass +class NousRateGuardVerdict: + """``action``: ``"fallthrough"`` (no active limit — make the call), ``"break"`` + (fallback armed on ``_retry``) or ``"return"`` (``result``: no fallback available).""" + + action: str + active_system_prompt: Any + retry_count: Any + compression_attempts: Any + result: Optional[Dict[str, Any]] = None + + +def nous_rate_limit_guard( + agent: Any, + *, + _retry: Any, + api_messages: Any, + messages: Any, + conversation_history: Any, + active_system_prompt: Any, + retry_count: Any, + compression_attempts: Any, + api_call_count: Any, +) -> NousRateGuardVerdict: + """Skip the call if another session recorded a Nous Portal rate limit: every attempt (incl. + SDK retries) counts against RPH. Never lets the guard itself break the agent loop.""" + from agent.conversation_loop import ( + _arm_fallback_restart, + ) + + def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> NousRateGuardVerdict: + return NousRateGuardVerdict( + action=action, + active_system_prompt=active_system_prompt, + retry_count=retry_count, + compression_attempts=compression_attempts, + result=result, + ) + + # ── Nous Portal rate limit guard ────────────────────── + # Skip the call if another session recorded a rate limit: every attempt + # (incl. SDK retries) counts against RPH. + if agent.provider == "nous": + try: + from agent.nous_rate_guard import ( + nous_rate_limit_remaining, + format_remaining as _fmt_nous_remaining, + ) + _nous_remaining = nous_rate_limit_remaining() + if _nous_remaining is not None and _nous_remaining > 0: + _nous_msg = ( + f"Nous Portal rate limit active — " + f"resets in {_fmt_nous_remaining(_nous_remaining)}." + ) + agent._buffer_vprint( + f"⏳ {_nous_msg} Trying fallback..." + ) + agent._buffer_status(f"⏳ {_nous_msg}") + if agent._try_activate_fallback(): + active_system_prompt = _arm_fallback_restart( + agent, api_messages, active_system_prompt, _retry) + retry_count = 0 + compression_attempts = 0 + return _verdict("break") + # No fallback available — surface buffered context + # so user sees the rate-limit message that led here. + agent._flush_status_buffer() + agent._persist_session(messages, conversation_history) + return _verdict("return", { + "final_response": ( + f"⏳ {_nous_msg}\n\n" + "No fallback provider available. " + "Try again after the reset, or add a " + "fallback provider in config.yaml." + ), + "messages": messages, + "api_calls": api_call_count, + "completed": False, + "failed": True, + "error": _nous_msg, + }) + except ImportError: + pass + except Exception: + pass # Never let rate guard break the agent loop + return _verdict("fallthrough") diff --git a/agent/turn_api_error.py b/agent/turn_api_error.py index e3c41fc073..61a5ca287d 100644 --- a/agent/turn_api_error.py +++ b/agent/turn_api_error.py @@ -9,12 +9,12 @@ existing ``patch("agent.conversation_loop.X")`` sites keep intercepting. from __future__ import annotations -import logging from dataclasses import dataclass -from typing import Any, Dict, Optional import json +import logging import ssl import time +from typing import Any, Dict, Optional from agent.error_classifier import FailoverReason, classify_api_error from agent.message_sanitization import close_interrupted_tool_sequence diff --git a/agent/turn_api_request.py b/agent/turn_api_request.py index 84cbe9ab64..7cb6a6f20b 100644 --- a/agent/turn_api_request.py +++ b/agent/turn_api_request.py @@ -8,9 +8,10 @@ nothing here imports ``agent.conversation_loop`` at module level. from __future__ import annotations -import logging from dataclasses import dataclass +import logging from typing import Any, Dict, Optional + from agent.message_sanitization import ( _sanitize_structure_non_ascii, _sanitize_structure_surrogates, diff --git a/agent/turn_final_response.py b/agent/turn_final_response.py index 7a5ffdbec1..bf1baeb13d 100644 --- a/agent/turn_final_response.py +++ b/agent/turn_final_response.py @@ -7,9 +7,10 @@ re-prompt, scaffolding pop, stop gates, then the durable final flush. Extracted from __future__ import annotations -import logging from dataclasses import dataclass +import logging from typing import Any, Dict, Optional + from agent.message_metadata import append_message from agent.turn_empty_response import recover_empty_response from agent.turn_stop_gates import apply_stop_gates diff --git a/agent/turn_iteration_prep.py b/agent/turn_iteration_prep.py index e880077493..eb2f3130c3 100644 --- a/agent/turn_iteration_prep.py +++ b/agent/turn_iteration_prep.py @@ -8,11 +8,14 @@ repair. Extracted from ``run_conversation``; nothing here imports from __future__ import annotations -import logging from dataclasses import dataclass +import logging +import random from typing import Any, Dict, Optional +from agent.display import KawaiiSpinner +from agent.turn_context import reanchor_current_turn_user_idx logger = logging.getLogger("agent.conversation_loop") @@ -187,3 +190,280 @@ def prepare_iteration( agent.session_id or "-", ) return _verdict("fallthrough") + + +@dataclass +class ApiCallAnnouncement: + """Always ``action == "fallthrough"``; ``thinking_spinner`` is the started raw spinner or + None (TUI widget / streaming consumers / verbose mode).""" + + action: str + thinking_spinner: Any + + +def announce_api_call( + agent: Any, + *, + messages: Any, + api_messages: Any, + api_call_count: Any, + approx_tokens: Any, + total_chars: Any, +) -> ApiCallAnnouncement: + """Print the request summary (verbose) or start the quiet-mode thinking indicator.""" + + + def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> ApiCallAnnouncement: + return ApiCallAnnouncement( + action=action, + thinking_spinner=thinking_spinner, + + ) + + # Thinking spinner for quiet mode (animated during API call) + thinking_spinner = None + + if not agent.quiet_mode: + agent._vprint(f"\n{agent.log_prefix}🔄 Making API call #{api_call_count}/{agent.max_iterations}...") + agent._vprint(f"{agent.log_prefix} 📊 Request size: {len(api_messages)} messages, ~{approx_tokens:,} tokens (~{total_chars:,} chars)") + agent._vprint(f"{agent.log_prefix} 🔧 Available tools: {len(agent.tools) if agent.tools else 0}") + else: + # Animated thinking spinner in quiet mode + face = random.choice(KawaiiSpinner.get_thinking_faces()) + verb = random.choice(KawaiiSpinner.get_thinking_verbs()) + if agent.thinking_callback: + # CLI TUI mode: use prompt_toolkit widget instead of raw spinner + # (works in both streaming and non-streaming modes) + agent.thinking_callback(f"{face} {verb}...") + elif not agent._has_stream_consumers() and agent._should_start_quiet_spinner(): + # Raw KawaiiSpinner only when no streaming consumers and the + # spinner output has a safe sink. + spinner_type = random.choice(['brain', 'sparkle', 'pulse', 'moon', 'star']) + thinking_spinner = KawaiiSpinner(f"{face} {verb}...", spinner_type=spinner_type, print_fn=agent._print_fn) + thinking_spinner.start() + + # Log request details if verbose + if agent.verbose_logging: + logging.debug(f"API Request - Model: {agent.model}, Messages: {len(messages)}, Tools: {len(agent.tools) if agent.tools else 0}") + logging.debug(f"Last message role: {messages[-1]['role'] if messages else 'none'}") + logging.debug(f"Total message size: ~{approx_tokens:,} tokens") + return _verdict("fallthrough") + + +@dataclass +class IterationStart: + """``action``: ``"fallthrough"`` (run the iteration) or ``"break"`` (turn ends: interrupt, + review input budget or iteration budget exhausted — ``_turn_exit_reason`` set).""" + + action: str + original_user_message: Any + api_call_count: Any + interrupted: Any + _turn_exit_reason: Any + + +def begin_iteration( + agent: Any, + *, + messages: Any, + conversation_history: Any, + original_user_message: Any, + api_call_count: Any, + interrupted: Any, + _turn_exit_reason: Any, +) -> IterationStart: + """Iteration entry in the original order: apply a pending redirect, reset the checkpoint + dedup, then the interrupt / review-budget / iteration-budget exits. ``api_call_count`` is + incremented here (the grace call consumes its flag instead of the budget).""" + from agent.conversation_loop import ( + _apply_active_turn_redirect, + _review_input_budget_exhausted, + ) + + def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> IterationStart: + return IterationStart( + action=action, + original_user_message=original_user_message, + api_call_count=api_call_count, + interrupted=interrupted, + _turn_exit_reason=_turn_exit_reason, + + ) + + _redirect_text = agent._drain_pending_redirect() + if _redirect_text: + _apply_active_turn_redirect(agent, messages, _redirect_text) + if isinstance(original_user_message, str): + original_user_message = ( + f"{original_user_message}\n\n" + f"User correction during the turn: {_redirect_text}" + ) + agent._persist_session(messages, conversation_history) + + # Reset per-turn checkpoint dedup so each iteration can take one snapshot + agent._checkpoint_mgr.new_turn() + + # Check for interrupt request (e.g., user sent new message) + if agent._interrupt_requested: + interrupted = True + _turn_exit_reason = "interrupted_by_user" + if not agent.quiet_mode: + agent._safe_print("\n⚡ Breaking out of tool loop due to interrupt...") + return _verdict("break") + + # Aggregate input budget for detached auxiliary forks: bounds the whole review, + # not each request. Checked between iterations so the crossing request's writes + # have landed, mirroring the iteration-budget exit (#93057). + if _review_input_budget_exhausted(agent): + _turn_exit_reason = "review_input_budget_exhausted" + if not agent.quiet_mode: + agent._safe_print( + f"\n⏹️ Review input budget exhausted " + f"({int(agent.session_input_tokens):,} tokens) — stopping " + f"the review tool loop before the next provider call." + ) + return _verdict("break") + + api_call_count += 1 + agent._api_call_count = api_call_count + agent._touch_activity(f"starting API call #{api_call_count}") + + # Grace call: budget exhausted but the model gets one more call. Consume the + # flag so the loop exits after this iteration regardless of outcome. + if agent._budget_grace_call: + agent._budget_grace_call = False + elif not agent.iteration_budget.consume(): + _turn_exit_reason = "budget_exhausted" + if not agent.quiet_mode: + agent._safe_print(f"\n⚠️ Iteration budget exhausted ({agent.iteration_budget.used}/{agent.iteration_budget.max_total} iterations used)") + return _verdict("break") + return _verdict("fallthrough") + + +@dataclass +class RetryRestartVerdict: + """``action``: ``"fallthrough"`` (a response is ready — process it), ``"continue"`` + (a restart flag re-issues the iteration: redirect / compressed / rebuilt-for-fallback / + length continuation) or ``"break"`` (turn ends: interrupted, non-actionable compaction + handoff, or every retry exhausted without a response).""" + + action: str + current_turn_user_idx: Any + final_response: Any + retry_count: Any + api_call_count: Any + _preflight_compression_blocked: Any + _turn_exit_reason: Any + + +def apply_retry_restarts( + agent: Any, + *, + _retry: Any, + response: Any, + interrupted: Any, + messages: Any, + conversation_history: Any, + user_message: Any, + api_kwargs: Any, + current_turn_user_idx: Any, + final_response: Any, + retry_count: Any, + api_call_count: Any, + length_continue_retries: Any, + _preflight_compression_blocked: Any, + _turn_exit_reason: Any, +) -> RetryRestartVerdict: + """Consume the ``TurnRetryState`` restart flags after the retry loop, in the original + priority order. Refunds the iteration budget/count for restarts that produced no valid + assistant item; ``restart_with_rebuilt_messages`` is the single consumer that clears + ``_preflight_compression_blocked`` so the fallback gets a fresh preflight (#84733).""" + from agent.conversation_loop import ( + _HANDOFF_SKIP_FINAL_RESPONSE, + _should_skip_model_call_for_reference_handoff, + ) + + def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> RetryRestartVerdict: + return RetryRestartVerdict( + action=action, + current_turn_user_idx=current_turn_user_idx, + final_response=final_response, + retry_count=retry_count, + api_call_count=api_call_count, + _preflight_compression_blocked=_preflight_compression_blocked, + _turn_exit_reason=_turn_exit_reason, + + ) + + if _retry.restart_with_redirected_messages: + # Cancelled request produced no valid assistant item: reuse the same logical + # iteration after the outer loop appends partial context + correction. + api_call_count -= 1 + agent.iteration_budget.refund() + _retry.restart_with_redirected_messages = False + return _verdict("continue") + + # If the API call was interrupted, skip response processing + if interrupted: + _turn_exit_reason = "interrupted_during_api_call" + return _verdict("break") + + if _retry.restart_with_compressed_messages: + api_call_count -= 1 + agent.iteration_budget.refund() + # Compression restarts count toward the retry limit so a compression that + # shrinks messages but not enough can't loop forever. + retry_count += 1 + _retry.restart_with_compressed_messages = False + if _should_skip_model_call_for_reference_handoff( + messages, user_message + ): + logger.info( + "Skipping compressed-restart model call: reference-only " + "handoff would be the sole active user turn (#80622)" + ) + if not final_response: + final_response = _HANDOFF_SKIP_FINAL_RESPONSE + _turn_exit_reason = "compaction_handoff_not_actionable" + return _verdict("break") + # In-loop compression rebuilt `messages`; re-anchor the current-turn index + # like the prologue, AFTER the handoff guard (it may re-append this turn's + # ask). A stale anchor injects prefetch into a historical row. + current_turn_user_idx = reanchor_current_turn_user_idx( + messages, user_message + ) + agent._persist_user_message_idx = current_turn_user_idx + return _verdict("continue") + + if _retry.restart_with_rebuilt_messages: + # A stall/failure escalated to the fallback chain: re-issue against the + # active fallback provider, refunding budget/count for the stalled attempt. + api_call_count -= 1 + agent.iteration_budget.refund() + _retry.restart_with_rebuilt_messages = False + # Failover shrank the compressor window: clear the preflight block so + # preflight re-runs before the first fallback call. Hoisted to the single + # consumer. (#84733) + _preflight_compression_blocked = False + return _verdict("continue") + + if _retry.restart_with_length_continuation: + # Boost output budget per retry: 2×, 4×, 8×, 16× base, capped at 32 768, via + # _ephemeral_max_output_tokens. Keep a larger original provider/model + # default as the floor so retries never downshift. + _boost_base = agent.max_tokens if agent.max_tokens else 4096 + _boost = _boost_base * (2 ** length_continue_retries) + _requested_cap = agent._requested_output_cap_from_api_kwargs(api_kwargs) + if _requested_cap is not None: + _boost = max(_boost, _requested_cap) + _boost_cap = max(32768, _requested_cap or 0) + agent._ephemeral_max_output_tokens = min(_boost, _boost_cap) + return _verdict("continue") + + # All retries may exhaust with `response` still None; break out cleanly. + if response is None: + _turn_exit_reason = "all_retries_exhausted_no_response" + print(f"{agent.log_prefix}❌ All API retries exhausted with no successful response.") + agent._persist_session(messages, conversation_history) + return _verdict("break") + return _verdict("fallthrough") diff --git a/agent/turn_loop_errors.py b/agent/turn_loop_errors.py index d25a930bb9..92da98cb24 100644 --- a/agent/turn_loop_errors.py +++ b/agent/turn_loop_errors.py @@ -9,11 +9,11 @@ loop from spinning until the budget is gone (#92450). Nothing here imports from __future__ import annotations -import logging from dataclasses import dataclass -from typing import Any, Dict, Optional +import logging import os import sys +from typing import Any, Dict, Optional from agent.message_metadata import append_message diff --git a/agent/turn_preflight_gate.py b/agent/turn_preflight_gate.py index 738965719d..777c303f41 100644 --- a/agent/turn_preflight_gate.py +++ b/agent/turn_preflight_gate.py @@ -7,8 +7,8 @@ here imports ``agent.conversation_loop`` at module level (cycle). from __future__ import annotations -import logging from dataclasses import dataclass +import logging from typing import Any, Dict, Optional from agent.message_metadata import append_message diff --git a/agent/turn_request_assembly.py b/agent/turn_request_assembly.py index adb4f9dc13..ef85a26676 100644 --- a/agent/turn_request_assembly.py +++ b/agent/turn_request_assembly.py @@ -10,12 +10,12 @@ sites keep intercepting. from __future__ import annotations -import logging from dataclasses import dataclass +import logging from typing import Any, Dict, Optional -from agent.model_metadata import anchored_context_tokens from agent.message_sanitization import _sanitize_messages_surrogates +from agent.model_metadata import anchored_context_tokens from agent.prompt_caching import build_prompt_cache_plan, effective_cache_ttl from agent.turn_context import build_api_messages diff --git a/agent/turn_response_check.py b/agent/turn_response_check.py index d40fb7c5e5..a65c63956d 100644 --- a/agent/turn_response_check.py +++ b/agent/turn_response_check.py @@ -9,10 +9,10 @@ existing ``patch("agent.conversation_loop.X")`` sites keep intercepting. from __future__ import annotations -import logging from dataclasses import dataclass -from typing import Any, Dict, Optional +import logging import time +from typing import Any, Dict, Optional from agent.turn_truncation import handle_content_policy_refusal, recover_from_truncation from agent.turn_usage import record_response_usage diff --git a/agent/turn_response_intake.py b/agent/turn_response_intake.py index bd3fe722b7..3bb092e8c1 100644 --- a/agent/turn_response_intake.py +++ b/agent/turn_response_intake.py @@ -7,11 +7,11 @@ continuation guards. Extracted from ``run_conversation``; nothing here imports from __future__ import annotations -import logging from dataclasses import dataclass -from typing import Any, Dict, Optional import json +import logging import re +from typing import Any, Dict, Optional from agent.provider_projection import splice_provider_projection from agent.trajectory import has_incomplete_scratchpad diff --git a/agent/turn_tool_round.py b/agent/turn_tool_round.py index 02d532ff7f..5d13fb5309 100644 --- a/agent/turn_tool_round.py +++ b/agent/turn_tool_round.py @@ -7,9 +7,10 @@ guardrail halts / persistence failures, then compress after tool results. Extrac from __future__ import annotations -import logging from dataclasses import dataclass +import logging from typing import Any, Dict, Optional + from agent.message_metadata import append_message from agent.message_sanitization import coalesce_tool_call_id from agent.turn_preflight import compress_after_tool_results diff --git a/tests/agent/test_prompt_cache_ttl_propagation.py b/tests/agent/test_prompt_cache_ttl_propagation.py index f2ab8d97f3..4a8fd7ce20 100644 --- a/tests/agent/test_prompt_cache_ttl_propagation.py +++ b/tests/agent/test_prompt_cache_ttl_propagation.py @@ -198,89 +198,85 @@ class TestFailoverRestartsPreflight: """ def test_every_fallback_activation_restarts_preflight(self): - from agent import conversation_loop + """The retry-loop body now lives in verdict helpers (``turn_api_call``, + ``turn_api_error``, ``turn_response_check``): a fallback activation there must + end in ``return _verdict("break")`` — the loop's ``break`` — so the + ``restart_with_rebuilt_messages`` handler after the retry loop refunds the + budget and re-runs the preflight. Every ``_try_activate_fallback`` reference + must be a direct ``if agent._try_activate_fallback():`` site so this guard can + bind its restart discipline (#84733).""" + from agent import turn_api_call, turn_api_error, turn_response_check - tree = ast.parse(inspect.getsource(conversation_loop.run_conversation)) + def _verdict_kind(stmt): + if ( + isinstance(stmt, ast.Return) + and isinstance(stmt.value, ast.Call) + and isinstance(stmt.value.func, ast.Name) + and stmt.value.func.id == "_verdict" + and stmt.value.args + and isinstance(stmt.value.args[0], ast.Constant) + ): + return stmt.value.args[0].value + return None - # Parent map so each site can be bound to its nearest enclosing loop. - parents = {} - for node in ast.walk(tree): - for child in ast.iter_child_nodes(node): - parents[child] = node + seen = 0 + for mod in (turn_api_call, turn_api_error, turn_response_check): + tree = ast.parse(inspect.getsource(mod)) + fallback_ifs = [ + node + for node in ast.walk(tree) + if isinstance(node, ast.If) + and isinstance(node.test, ast.Call) + and isinstance(node.test.func, ast.Attribute) + and node.test.func.attr == "_try_activate_fallback" + ] + all_refs = [ + node + for node in ast.walk(tree) + if isinstance(node, ast.Attribute) + and node.attr == "_try_activate_fallback" + ] + assert len(all_refs) == len(fallback_ifs), ( + f"{mod.__name__}: every _try_activate_fallback reference must be a " + "direct `if agent._try_activate_fallback(...):` site (#84733)" + ) + for node in fallback_ifs: + kinds = [_verdict_kind(stmt) for stmt in node.body] + assert "break" in kinds, ( + f"{mod.__name__}: retry-loop fallback activation must return " + "_verdict(\"break\") so the restart-with-rebuilt-messages handler " + "re-runs the pre-API preflight against the fallback's context " + "window (#84733)" + ) + assert "continue" not in kinds, ( + f"{mod.__name__}: a `continue` verdict would only re-fire the retry " + "loop and skip the preflight (#84733)" + ) + seen += len(fallback_ifs) + assert seen >= 5, "expected the retry-loop _try_activate_fallback sites" - retry_loops = [ - node - for node in ast.walk(tree) - if isinstance(node, ast.While) - and isinstance(node.test, ast.Compare) - and isinstance(node.test.left, ast.Name) - and node.test.left.id == "retry_count" - ] - assert retry_loops, "expected the retry loop in run_conversation" - retry_loop_ids = {id(loop) for loop in retry_loops} + # The restart consumer must re-issue the OUTER iteration (`continue` verdict). + from agent import turn_iteration_prep - def _inside_retry_loop(node): - cur = parents.get(node) - while cur is not None: - if id(cur) in retry_loop_ids: - return True - cur = parents.get(cur) - return False - - fallback_ifs = [ + tree = ast.parse(inspect.getsource(turn_iteration_prep.apply_retry_restarts)) + handlers = [ node for node in ast.walk(tree) if isinstance(node, ast.If) - and isinstance(node.test, ast.Call) - and isinstance(node.test.func, ast.Attribute) - and node.test.func.attr == "_try_activate_fallback" + and isinstance(node.test, ast.Attribute) + and node.test.attr == "restart_with_rebuilt_messages" ] - assert fallback_ifs, "expected _try_activate_fallback sites in run_conversation" - # Every reference to _try_activate_fallback must be one of the matched - # `if agent._try_activate_fallback(...):` sites — a site written as - # `activated = agent._try_activate_fallback()` would silently escape - # this guard. - all_refs = [ - node - for node in ast.walk(tree) - if isinstance(node, ast.Attribute) - and node.attr == "_try_activate_fallback" - ] - assert len(all_refs) == len(fallback_ifs), ( - "every _try_activate_fallback reference must be a direct " - "`if agent._try_activate_fallback(...):` site so this guard " - "can bind its restart discipline (#84733)" - ) - for node in fallback_ifs: - if _inside_retry_loop(node): - assert any(isinstance(stmt, ast.Break) for stmt in node.body), ( - "retry-loop fallback activation must break to the " - "restart-with-rebuilt-messages handler so the pre-API " - "preflight re-runs against the fallback's context " - "window (#84733)" - ) - else: - assert any( - isinstance(stmt, ast.Continue) for stmt in node.body - ), ( - "outer-loop fallback activation must continue the outer " - "iteration (which re-runs the preflight); a break here " - "would end the turn without calling the fallback (#84733)" - ) - assert not any( - isinstance(stmt, ast.Break) for stmt in node.body - ), ( - "outer-loop fallback activation must not break — that " - "exits the conversation loop and ends the turn (#84733)" - ) + assert handlers and all( + any(_verdict_kind(stmt) == "continue" for stmt in node.body) for node in handlers + ), "the restart handler must `continue` the outer iteration (#84733)" def test_restart_handler_clears_preflight_block(self): """The single consumer of restart_with_rebuilt_messages must clear _preflight_compression_blocked, so every retry-loop failover gets a fresh preflight against the fallback's context window (#84733).""" - from agent import conversation_loop + from agent import turn_iteration_prep - tree = ast.parse(inspect.getsource(conversation_loop.run_conversation)) + tree = ast.parse(inspect.getsource(turn_iteration_prep.apply_retry_restarts)) handlers = [ node for node in ast.walk(tree)