diff --git a/agent/turn_api_call.py b/agent/turn_api_call.py index 6c7aa3744d..0aa3519289 100644 --- a/agent/turn_api_call.py +++ b/agent/turn_api_call.py @@ -2,12 +2,13 @@ the attempt while another session's Nous Portal rate limit is active), ``perform_api_call`` (streaming decision, MoA prepared-request handshake, LLM execution middleware wrapper, the redirect ``_model_request_active`` bracket and the response-vs-redirect crossing check) and -``handle_api_interrupt`` (``InterruptedError`` mid-call). Extracted from -``run_conversation``; nothing here imports ``agent.conversation_loop`` at module level. +``handle_api_interrupt`` (``InterruptedError`` mid-call). Nothing here imports +``agent.conversation_loop`` at module level (cycle). """ from __future__ import annotations +from contextlib import nullcontext from dataclasses import dataclass import logging import time @@ -18,6 +19,16 @@ from agent.message_metadata import append_message logger = logging.getLogger("agent.conversation_loop") +def stop_thinking_spinner(agent: Any, thinking_spinner: Any) -> None: + """Stop the spinner silently and clear the thinking callback; returns ``None`` so + callers can rebind ``thinking_spinner = stop_thinking_spinner(agent, thinking_spinner)``.""" + if thinking_spinner: + thinking_spinner.stop("") + if agent.thinking_callback: + agent.thinking_callback("") + return None + + @dataclass class ApiCallVerdict: """``action``: ``"fallthrough"`` (``response`` is ready for verification) or ``"break"`` @@ -29,58 +40,44 @@ class ApiCallVerdict: interrupted: Any +def _should_stream(agent: Any) -> bool: + """Streaming is preferred even without consumers (stale-stream / read-timeout health + checks); disabled on provider signal, ACP schemes, MoA without a display consumer, or + Mock clients in tests (SimpleNamespace, not stream iterators).""" + if getattr(agent, "_disable_streaming", False): + return False + _base = str(agent.base_url or "").lower() + if agent.provider in {"copilot-acp"} or _base.startswith(("acp://", "acp+tcp://")): + return False + if not agent._has_stream_consumers(): + if agent.provider == "moa": + return False + from unittest.mock import Mock + if isinstance(getattr(agent, "client", None), Mock): + return False + return True + + def perform_api_call( agent: Any, *, api_kwargs: Any, _original_api_kwargs: Any, _llm_middleware_trace: Any, _moa_prepared_request: Any, _retry: Any, thinking_spinner: Any, retry_count: Any, api_call_count: Any, api_request_id: Any, effective_task_id: Any, turn_id: Any, interrupted: Any, ) -> ApiCallVerdict: - """Issue the request. Streaming is preferred even without consumers (stale-stream / - read-timeout health checks) and disabled per provider signal, ACP schemes, MoA without a - display consumer, or Mock clients in tests.""" + """Issue the request (see ``_should_stream`` for the streaming decision).""" response = None - def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> ApiCallVerdict: + def _verdict(action: str) -> ApiCallVerdict: return ApiCallVerdict( action=action, response=response, thinking_spinner=thinking_spinner, interrupted=interrupted, ) - # Always prefer streaming even without consumers: it gives stale- - # stream/read-timeout health checks that quiet callers otherwise lack. - # Falls back if unsupported. def _stop_spinner(): nonlocal thinking_spinner - if thinking_spinner: - thinking_spinner.stop("") - thinking_spinner = None - if agent.thinking_callback: - agent.thinking_callback("") + thinking_spinner = stop_thinking_spinner(agent, thinking_spinner) - _use_streaming = True - # Provider signaled "stream not supported": stay non-streaming for the - # session. - if getattr(agent, "_disable_streaming", False): - _use_streaming = False - # ACP clients (`acp://` scheme, any vendor) return a plain - # SimpleNamespace, not a stream; mirrors the Responses API exclusion. - elif ( - agent.provider in {"copilot-acp"} - or str(agent.base_url or "").lower().startswith("acp://") - or str(agent.base_url or "").lower().startswith("acp+tcp://") - ): - _use_streaming = False - # MoA streams only with a display/TTS consumer - # (MoAChatCompletions.create() honors stream=True); else complete- - # response path. - elif agent.provider == "moa" and not agent._has_stream_consumers(): - _use_streaming = False - elif not agent._has_stream_consumers(): - # No consumer: still stream for health checking, except Mock clients - # in tests (SimpleNamespace, not stream iterators). - from unittest.mock import Mock - if isinstance(getattr(agent, "client", None), Mock): - _use_streaming = False + _use_streaming = _should_stream(agent) def _perform_api_call(next_api_kwargs): if agent.api_mode == "codex_responses": @@ -117,15 +114,14 @@ def perform_api_call( from hermes_cli.middleware import run_llm_execution_middleware + # The ``_model_request_active`` bracket is taken under the redirect lock when one exists, + # so redirect() can't observe a half-toggled flag. _model_request_active = getattr(agent, "_model_request_active", None) _redirect_lock = getattr(agent, "_pending_redirect_lock", None) - if _redirect_lock is not None: - with _redirect_lock: - if _model_request_active is not None: - _model_request_active.set() - elif _model_request_active is not None: - _model_request_active.set() - _redirect_crossed_response = False + _bracket = nullcontext() if _redirect_lock is None else _redirect_lock + with _bracket: + if _model_request_active is not None: + _model_request_active.set() try: response = run_llm_execution_middleware( api_kwargs, _perform_api_call, original_request=_original_api_kwargs, @@ -135,25 +131,17 @@ def perform_api_call( api_call_count=api_call_count, middleware_trace=list(_llm_middleware_trace), ) finally: - if _redirect_lock is not None: - with _redirect_lock: - if _model_request_active is not None: - _model_request_active.clear() - _redirect_crossed_response = bool( - agent._pending_redirect - ) - else: + with _bracket: if _model_request_active is not None: _model_request_active.clear() - _redirect_crossed_response = agent._has_pending_redirect() + _redirect_crossed_response = ( + bool(agent._pending_redirect) if _redirect_lock is not None + else agent._has_pending_redirect() + ) if _redirect_crossed_response: # Response and redirect can cross threads: discard the now-stale # response and rebuild from the correction rather than lose it. - if thinking_spinner: - thinking_spinner.stop("") - thinking_spinner = None - if agent.thinking_callback: - agent.thinking_callback("") + thinking_spinner = stop_thinking_spinner(agent, thinking_spinner) if agent.clear_interrupt(preserve_redirect=True): _retry.restart_with_redirected_messages = True else: @@ -180,36 +168,18 @@ def handle_api_interrupt( """``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, - ) + 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") + thinking_spinner = stop_thinking_spinner(agent, thinking_spinner) + # 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._has_pending_redirect() and agent.clear_interrupt(preserve_redirect=True): + _retry.restart_with_redirected_messages = True + return ApiInterruptVerdict("break", thinking_spinner, interrupted, final_response) 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() @@ -219,8 +189,7 @@ def handle_api_interrupt( 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") + return ApiInterruptVerdict("break", thinking_spinner, interrupted, final_response) @dataclass @@ -241,9 +210,7 @@ def nous_rate_limit_guard( ) -> 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, - ) + from agent.conversation_loop import _arm_fallback_restart def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> NousRateGuardVerdict: return NousRateGuardVerdict( @@ -251,9 +218,6 @@ def nous_rate_limit_guard( 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 ( @@ -262,12 +226,9 @@ def nous_rate_limit_guard( _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..." + f"Nous Portal rate limit active — 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( @@ -275,15 +236,13 @@ def nous_rate_limit_guard( 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. + # No fallback — surface the buffered rate-limit context 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 " + "No fallback provider available. Try again after the reset, or add a " "fallback provider in config.yaml." ), "messages": messages, @@ -292,8 +251,6 @@ def nous_rate_limit_guard( "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_request.py b/agent/turn_api_request.py index 6aca189616..fbbb321a85 100644 --- a/agent/turn_api_request.py +++ b/agent/turn_api_request.py @@ -2,15 +2,15 @@ reasoning echo pad and prompt-cache decoration for the CURRENT provider (a fallback may differ from the primary), build ``api_kwargs``, run the surrogate/ASCII chokepoints, Codex preflight, OpenRouter cache bypass, Copilot ``x-initiator``, the LLM request middleware, -the ``pre_api_request`` hook and the debug dump. Extracted from ``run_conversation``; -nothing here imports ``agent.conversation_loop`` at module level. +the ``pre_api_request`` hook and the debug dump. Nothing here imports +``agent.conversation_loop`` at module level. """ from __future__ import annotations from dataclasses import dataclass import logging -from typing import Any, Dict, Optional +from typing import Any from agent.message_sanitization import ( _sanitize_structure_non_ascii, _sanitize_structure_surrogates @@ -34,6 +34,63 @@ class ApiRequestBuild: _llm_middleware_trace: Any +def _set_extra_header(api_kwargs: Any, key: str, value: str) -> None: + """Copy-on-write header set (the dict may be shared with the transport's defaults).""" + _xh = dict(api_kwargs.get("extra_headers") or {}) + _xh[key] = value + api_kwargs["extra_headers"] = _xh + + +def _fire_pre_api_request_hook( + agent: Any, api_kwargs: Any, api_messages: Any, _llm_middleware_trace: Any, *, messages: Any, + original_user_message: Any, approx_tokens: Any, total_chars: Any, retry_count: Any, + api_call_count: Any, api_request_id: Any, api_start_time: Any, effective_task_id: Any, + turn_id: Any, +) -> None: + from agent.conversation_loop import _system_prompt_for_hooks + + try: + from hermes_cli.lifecycle import has_hook, invoke_hook as _invoke_hook + if has_hook("pre_api_request"): + request_messages = api_kwargs.get("messages") + if not isinstance(request_messages, list): + request_messages = api_kwargs.get("input") + if not isinstance(request_messages, list): + request_messages = api_messages + # Shallow copies: plugins may retain the lists; deepcopy is costly. + # ``request_messages``/``conversation_history`` are raw langfuse passthroughs. + # Anthropic (``system``) and Responses/Codex (``instructions``) move the system + # prompt out of messages; pass it for observability. + _invoke_hook( + "pre_api_request", + task_id=effective_task_id, + turn_id=turn_id, + api_request_id=api_request_id, + session_id=agent.session_id or "", + user_message=original_user_message, + conversation_history=list(messages), + platform=agent.platform or "", + model=agent.model, + provider=agent.provider, + base_url=agent.base_url, + api_mode=agent.api_mode, + api_call_count=api_call_count, + retry_count=retry_count, + request_messages=list(request_messages) if isinstance(request_messages, list) else [], + system_prompt=_system_prompt_for_hooks(api_kwargs, request_messages), + message_count=len(api_messages), + tool_count=len(agent.tools or []), + approx_input_tokens=approx_tokens, + request_char_count=total_chars, + max_tokens=agent.max_tokens, + started_at=api_start_time, + middleware_trace=list(_llm_middleware_trace), + request=agent._api_request_payload_for_hook(api_kwargs), + ) + except Exception: + pass + + def build_api_request( agent: Any, *, api_messages: Any, _moa_prepared_request: Any, tools_for_api: Any, system_message: Any, messages: Any, original_user_message: Any, approx_tokens: Any, @@ -44,29 +101,15 @@ def build_api_request( middleware/hooks/debug dumps observe the payload).""" from agent.conversation_loop import ( _moa_client_consumes_prepared_request, _redecorate_prompt_cache_for_provider, - _system_prompt_for_hooks, ) - api_kwargs = None - _original_api_kwargs = None - _llm_middleware_trace = [] - - def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> ApiRequestBuild: - return ApiRequestBuild( - action=action, api_messages=api_messages, _moa_prepared_request=_moa_prepared_request, - tools_for_api=tools_for_api, api_kwargs=api_kwargs, - _original_api_kwargs=_original_api_kwargs, _llm_middleware_trace=_llm_middleware_trace, - ) agent._reset_stream_delivery_tracking() - # Per-attempt first-chunk timestamp so a stale value never leaks into - # post_api_request. + # Per-attempt first-chunk timestamp so a stale value never leaks into post_api_request. agent._last_api_first_chunk_at = None - # api_messages was built for the primary; a fallback (DeepSeek / Kimi / - # MiMo) may require reasoning_content. Re-apply the echo-back pad - # (idempotent). + # api_messages was built for the primary; a fallback (DeepSeek / Kimi / MiMo) may + # require reasoning_content — re-apply the echo-back pad (idempotent) and re-render + # the prompt-cache decoration for the current provider. agent._reapply_reasoning_echo_for_provider(api_messages) - # Same for prompt-cache decoration (#72626): strip the primary's - # breakpoints and re-render for the current provider. api_messages, _moa_prepared_request, tools_for_api = ( _redecorate_prompt_cache_for_provider( agent, api_messages, system_message=system_message, moa_prepared=_moa_prepared_request, @@ -76,12 +119,9 @@ def build_api_request( if tools_for_api == agent.tools: api_kwargs = agent._build_api_kwargs(api_messages) else: - api_kwargs = agent._build_api_kwargs( - api_messages, tools_for_api=tools_for_api - ) - # Surrogate chokepoint (#50959): tool descriptions, extra_body and - # kwargs strings can carry invalid code points (HTTP 400). One walk - # makes the payload json.dumps()-safe. + api_kwargs = agent._build_api_kwargs(api_messages, tools_for_api=tools_for_api) + # Surrogate chokepoint: tool descriptions, extra_body and kwargs strings can carry + # invalid code points (HTTP 400). One walk makes the payload json.dumps()-safe. _sanitize_structure_surrogates(api_kwargs) if agent._force_ascii_payload: _sanitize_structure_non_ascii(api_kwargs) @@ -90,18 +130,14 @@ def build_api_request( api_kwargs, allow_stream=False, is_github_responses=agent._is_copilot_url(), sanitize_harmony_tokens=agent._is_codex_backend(), ) - # OpenRouter caching replays identical responses, even empty ones; an - # empty-response retry must bypass the cache. + # OpenRouter caching replays identical responses, even empty ones; an empty-response + # retry must bypass the cache. if agent._empty_content_retries > 0 and agent._is_openrouter_url(): - _xh = dict(api_kwargs.get("extra_headers") or {}) - _xh["X-OpenRouter-Cache"] = "false" - api_kwargs["extra_headers"] = _xh - # Copilot x-initiator: first call of a user turn is "user" (billed - # premium); tool-loop follow-ups keep the default "agent" (#3040). + _set_extra_header(api_kwargs, "X-OpenRouter-Cache", "false") + # Copilot x-initiator: first call of a user turn is "user" (billed premium); + # tool-loop follow-ups keep the default "agent". if getattr(agent, "_is_user_initiated_turn", False) and agent._is_copilot_url(): - _xh = dict(api_kwargs.get("extra_headers") or {}) - _xh["x-initiator"] = "user" - api_kwargs["extra_headers"] = _xh + _set_extra_header(api_kwargs, "x-initiator", "user") agent._is_user_initiated_turn = False try: from hermes_cli.middleware import apply_llm_request_middleware @@ -119,66 +155,22 @@ def build_api_request( _original_api_kwargs = dict(api_kwargs) _llm_middleware_trace = [] - try: - from hermes_cli.lifecycle import ( - has_hook, invoke_hook as _invoke_hook - ) - if has_hook("pre_api_request"): - request_messages = api_kwargs.get("messages") - if not isinstance(request_messages, list): - request_messages = api_kwargs.get("input") - if not isinstance(request_messages, list): - request_messages = api_messages - # Shallow copy: plugins may retain the list; deepcopy is costly. - # ``request_messages``/``conversation_history`` are raw langfuse - # passthroughs. - _request_payload = agent._api_request_payload_for_hook(api_kwargs) - # Anthropic (``system``) and Responses/Codex (``instructions``) - # move the system prompt out of messages; pass it for - # observability. - system_prompt_for_hooks = _system_prompt_for_hooks( - api_kwargs, request_messages - ) - _invoke_hook( - "pre_api_request", - task_id=effective_task_id, - turn_id=turn_id, - api_request_id=api_request_id, - session_id=agent.session_id or "", - user_message=original_user_message, - conversation_history=list(messages), - platform=agent.platform or "", - model=agent.model, - provider=agent.provider, - base_url=agent.base_url, - api_mode=agent.api_mode, - api_call_count=api_call_count, - retry_count=retry_count, - request_messages=list(request_messages) - if isinstance(request_messages, list) - else [], - system_prompt=system_prompt_for_hooks, - message_count=len(api_messages), - tool_count=len(agent.tools or []), - approx_input_tokens=approx_tokens, - request_char_count=total_chars, - max_tokens=agent.max_tokens, - started_at=api_start_time, - middleware_trace=list(_llm_middleware_trace), - request=_request_payload, - ) - except Exception: - pass + _fire_pre_api_request_hook( + agent, api_kwargs, api_messages, _llm_middleware_trace, messages=messages, + original_user_message=original_user_message, approx_tokens=approx_tokens, + total_chars=total_chars, retry_count=retry_count, api_call_count=api_call_count, + api_request_id=api_request_id, api_start_time=api_start_time, + effective_task_id=effective_task_id, turn_id=turn_id, + ) if env_var_enabled("HERMES_DUMP_REQUESTS"): agent._dump_api_request_debug(api_kwargs, reason="preflight") - # Private to the in-process MoA facade; add after middleware/hooks/debug - # dumps so none serializes it into the provider payload. + # Private to the in-process MoA facade; added after middleware/hooks/debug dumps so + # none serializes it into the provider payload. Re-read the live client: + # rotation/fallback/cleanup rebuild agent.client between attempts; a native OpenAI + # client rejects this key (TypeError). if _moa_prepared_request is not None and agent.provider == "moa": - # Re-read the live client: rotation/fallback/cleanup rebuild - # agent.client between attempts; a native OpenAI client rejects this - # key (TypeError). if _moa_client_consumes_prepared_request(agent.client): api_kwargs["_moa_prepared_request"] = _moa_prepared_request else: @@ -187,4 +179,7 @@ def build_api_request( "prepared prompt without the MoA handshake", type(agent.client).__name__, ) - return _verdict("fallthrough") + return ApiRequestBuild( + "fallthrough", api_messages, _moa_prepared_request, tools_for_api, api_kwargs, + _original_api_kwargs, _llm_middleware_trace, + ) diff --git a/agent/turn_empty_response.py b/agent/turn_empty_response.py index 75a8ab063f..537b0735d6 100644 --- a/agent/turn_empty_response.py +++ b/agent/turn_empty_response.py @@ -1,11 +1,10 @@ """Empty / thinking-only final-response recovery ladder for the conversation turn loop. -Extracted from ``run_conversation``. Runs when the model returned no visible text after -```` blocks. Ladder order is load-bearing: partial-stream recovery → reuse prior -turn content (housekeeping tools only) → one post-tool-call nudge (#9400) → thinking-only -prefill continuation (×2) → empty-response retries (budgeted, deterministic-empty -short-circuit) → fallback provider → terminal ``(empty)`` sentinel. Nothing here imports -``agent.conversation_loop`` at module level (cycle); loop-internal helpers resolve lazily. +Runs when the model returned no visible text after ```` blocks. Ladder order is +load-bearing: partial-stream recovery → reuse prior turn content (housekeeping tools only) +→ one post-tool-call nudge → thinking-only prefill continuation (×2) → empty-response +retries (budgeted, deterministic-empty short-circuit) → fallback provider → terminal +``(empty)`` sentinel. Nothing here imports ``agent.conversation_loop`` at module level. """ from __future__ import annotations @@ -21,6 +20,8 @@ from agent.turn_recovery import interruptible_backoff_sleep logger = logging.getLogger("agent.conversation_loop") +_INLINE_THINK_RE = re.compile(r'||', re.IGNORECASE) + @dataclass class EmptyResponseVerdict: @@ -40,6 +41,101 @@ class EmptyResponseVerdict: preflight_compression_blocked: bool +def _retry_empty( + agent: Any, response: Any, finish_reason: str, empty_candidate: bool, *, messages: Any, + conversation_history: Any, api_call_count: int, +) -> tuple: + """Budgeted empty-response retry. Each empty attempt re-bills the full input, so the + signature is recorded and deterministic empties stop burning paid retries (fails + open: missing usage or any output keeps the budget). Returns + ``(action_or_None, interrupt_result, deterministic_empty)``.""" + from agent.conversation_loop import jittered_backoff + + if empty_candidate: + _empty_guard.record_empty_attempt(agent, finish_reason=finish_reason, response=response) + budget = ( + _empty_guard.empty_retry_budget(agent, response) + if empty_candidate else _empty_guard.DEFAULT_EMPTY_RETRY_BUDGET + ) + deterministic = empty_candidate and _empty_guard.deterministic_empty(agent) + if not (empty_candidate and agent._empty_content_retries < budget and not deterministic): + return None, None, deterministic + agent._empty_content_retries += 1 + n = agent._empty_content_retries + wait_time = jittered_backoff(n, base_delay=5.0, max_delay=60.0) + logger.warning( + "Empty response (no content or reasoning) — retry %d/%d in %.1fs (model=%s)", + n, budget, wait_time, agent.model, + ) + _budget_note = ( + " — high-cost request, reduced retry budget" + if budget < _empty_guard.DEFAULT_EMPTY_RETRY_BUDGET else "" + ) + agent._buffer_status( + f"⚠️ Empty response from model — retrying ({n}/{budget}) in {wait_time:.0f}s{_budget_note}" + ) + _interrupted = interruptible_backoff_sleep( + agent, wait_time, None, + messages=messages, + conversation_history=conversation_history, + api_call_count=api_call_count, + abort_message="Interrupt detected during empty-response retry wait, aborting.", + interrupt_text=( + f"Operation interrupted: retrying empty response from model (retry {n}/{budget})." + ), + activity_label=f"empty response retry backoff ({n}/{budget})", + ) + if _interrupted is not None: + return "return", _interrupted, deterministic + return "continue", None, deterministic + + +def _terminal_empty(agent: Any, assistant_message: Any, finish_reason: str, messages: Any) -> str: + """Retries and fallback exhausted: persist the ``(empty)`` sentinel row and return the + delivery text. Reasoning is surfaced ONLY here, for delivery — the persisted row keeps + the sentinel so later "continue" turns don't replay it and loop on empties.""" + _streak_cost = _empty_guard.streak_cost_usd(agent) + if _streak_cost is not None: + agent._buffer_status( + f"ℹ️ Estimated cost of these empty attempts: ~${_streak_cost:.2f} (input tokens are billed " + f"per attempt even when no answer is produced)" + ) + agent._flush_status_buffer() + reasoning_text = agent._extract_reasoning(assistant_message) + agent._drop_trailing_empty_response_scaffolding(messages) + assistant_msg = agent._build_assistant_message(assistant_message, finish_reason) + assistant_msg["content"] = "(empty)" + assistant_msg["_empty_terminal_sentinel"] = True + append_message(messages, assistant_msg) + + if not reasoning_text: + logger.warning( + "Empty response (no content or reasoning) after %d retries. No fallback available. " + "model=%s provider=%s", + agent._empty_content_retries, agent.model, + agent.provider, + ) + agent._emit_status( + "❌ Model returned no content after all retries" + + (" and fallback attempts." if agent._fallback_chain else + ". No fallback providers configured.") + ) + return "(empty)" + + reasoning_preview = reasoning_text[:500] + "..." if len(reasoning_text) > 500 else reasoning_text + logger.warning( + "Reasoning-only response (no visible content) after exhausting retries and fallback. Reasoning: %s", reasoning_preview, + ) + agent._emit_status( + "⚠️ Model produced reasoning but no visible response after all retries. Returning empty." + ) + return ( + "⚠️ The model produced only internal reasoning and no final answer, despite retries" + + (" and fallback" if agent._fallback_chain else "") + + ". Its last reasoning, which may contain the answer:\n\n" + reasoning_preview + ) + + def recover_empty_response( agent: Any, assistant_message: Any, response: Any, finish_reason: str, *, final_response: Any, messages: List[Dict[str, Any]], api_messages: Any, conversation_history: Any, @@ -48,12 +144,8 @@ def recover_empty_response( ) -> EmptyResponseVerdict: """Recover from a final response with no visible content (see module docstring for the ladder). Role alternation is preserved: the post-tool nudge appends the empty - assistant row BEFORE the user-level hint (APIs reject tool→user). Reasoning is - surfaced only at the terminal step, for delivery — the persisted row keeps the - ``(empty)`` sentinel.""" - from agent.conversation_loop import ( - _EMPTY_TOOL_RESPONSE_NUDGE, _sync_failover_system_message, jittered_backoff - ) + assistant row BEFORE the user-level hint (APIs reject tool→user).""" + from agent.conversation_loop import _EMPTY_TOOL_RESPONSE_NUDGE, _sync_failover_system_message _turn_exit_reason = turn_exit_reason _preflight_compression_blocked = preflight_compression_blocked @@ -65,32 +157,26 @@ def recover_empty_response( preflight_compression_blocked=_preflight_compression_blocked, ) - # Partial stream recovery: content streamed before the connection - # died becomes the final response instead of fallback or retries. - _partial_streamed = ( - getattr(agent, "_current_streamed_assistant_text", "") or "" - ) + # Partial stream recovery: content streamed before the connection died becomes the + # final response instead of fallback or retries. + _partial_streamed = getattr(agent, "_current_streamed_assistant_text", "") or "" if agent._has_content_after_think_block(_partial_streamed): _turn_exit_reason = "partial_stream_recovery" _recovered = agent._strip_think_blocks(_partial_streamed).strip() logger.info( - "Partial stream content delivered (%d chars) " - "— using as final response", + "Partial stream content delivered (%d chars) — using as final response", len(_recovered), ) - agent._emit_status( - "↻ Stream interrupted — using delivered content " "as final response" - ) + agent._emit_status("↻ Stream interrupted — using delivered content " "as final response") final_response = _recovered - # A streamed fragment isn't a confirmed preview: keep - # response_previewed false so gateway fallback delivery can - # send the text plus the abnormal-turn explanation. + # A streamed fragment isn't a confirmed preview: gateway fallback delivery + # sends the text plus the abnormal-turn explanation. agent._response_was_previewed = False return _verdict("break") - # Prior turn had real content + ONLY housekeeping tools: model is - # done, reuse it. With substantive tools it was mid-task narration - # and the empty reply is a choke; let the post-tool nudge handle it. + # Prior turn had real content + ONLY housekeeping tools: model is done, reuse it. + # With substantive tools it was mid-task narration and the empty reply is a choke; + # let the post-tool nudge handle it. fallback = getattr(agent, '_last_content_with_tools', None) if fallback and getattr(agent, '_last_content_tools_all_housekeeping', False): _turn_exit_reason = "fallback_prior_turn_content" @@ -99,42 +185,28 @@ def recover_empty_response( agent._last_content_with_tools = None agent._last_content_tools_all_housekeeping = False agent._empty_content_retries = 0 - # Do NOT modify the assistant message content (injected text - # poisoned history); use the fallback as the response and break. + # Do NOT modify the assistant message content (injected text poisoned history). final_response = agent._strip_think_blocks(fallback).strip() agent._response_was_previewed = True return _verdict("break") - # ── Post-tool-call empty response nudge ─────────── - # Empty after tool results (no prior content, or only mid-task - # narration): nudge once via a user-level hint. (#9400) - _prior_was_tool = any( - m.get("role") == "tool" - for m in messages[-5:] # check recent messages - ) - # Ollama puts in content, not reasoning_content, so - # _has_structured misses it; detect here to route to prefill. - _has_inline_thinking = bool( - re.search( r'||', final_response or "", re.IGNORECASE ) - ) + # Post-tool-call empty (no prior content, or only mid-task narration): nudge once. + _prior_was_tool = any(m.get("role") == "tool" for m in messages[-5:]) + # Ollama puts in content, not reasoning_content, so _has_structured misses + # it; detect here to route to prefill. + _has_inline_thinking = bool(_INLINE_THINK_RE.search(final_response or "")) if ( _prior_was_tool and not getattr(agent, "_post_tool_empty_retried", False) and not _has_inline_thinking # thinking model still working — let prefill handle ): agent._post_tool_empty_retried = True - # Clear stale narration so it doesn't resurface - # on a later empty response after the nudge. + # Clear stale narration so it doesn't resurface on a later empty response. agent._last_content_with_tools = None agent._last_content_tools_all_housekeeping = False - logger.info( - "Empty response after tool calls — nudging model " "to continue processing" - ) - agent._buffer_status( - "⚠️ Model returned empty after tool calls — " "nudging to continue" - ) - # Append the empty assistant first so the sequence stays valid: - # tool → assistant("(empty)") → user (APIs reject tool→user). + logger.info("Empty response after tool calls — nudging model " "to continue processing") + agent._buffer_status("⚠️ Model returned empty after tool calls — " "nudging to continue") + # tool → assistant("(empty)") → user keeps the sequence valid. _nudge_msg = agent._build_assistant_message(assistant_message, finish_reason) _nudge_msg["content"] = "(empty)" _nudge_msg["_empty_recovery_synthetic"] = True @@ -144,9 +216,8 @@ def recover_empty_response( }) return _verdict("continue") - # ── Thinking-only prefill continuation ────────── - # Reasoning but no text: append as-is and continue so the model sees - # its own reasoning and writes text. Covers _has_inline_thinking. + # Thinking-only prefill: append the reasoning as-is and continue so the model sees + # its own reasoning and writes text. _has_structured = bool( getattr(assistant_message, "reasoning", None) or getattr(assistant_message, "reasoning_content", None) @@ -156,192 +227,60 @@ def recover_empty_response( if _has_structured and agent._thinking_prefill_retries < 2: agent._thinking_prefill_retries += 1 logger.info( - "Thinking-only response (no visible content) — " - "prefilling to continue (%d/2)", + "Thinking-only response (no visible content) — prefilling to continue (%d/2)", agent._thinking_prefill_retries, ) agent._buffer_status( - f"↻ Thinking-only response — prefilling to continue " - f"({agent._thinking_prefill_retries}/2)" - ) - interim_msg = agent._build_assistant_message( - assistant_message, "incomplete" + f"↻ Thinking-only response — prefilling to continue ({agent._thinking_prefill_retries}/2)" ) + interim_msg = agent._build_assistant_message(assistant_message, "incomplete") interim_msg["_thinking_prefill"] = True append_message(messages, interim_msg) agent._session_messages = messages return _verdict("continue") - # ── Empty response retry ────────────────────── - # Retry up to 3 times before fallback; covers truly empty replies - # AND reasoning-only replies after prefill exhaustion. - _truly_empty = not agent._strip_think_blocks( - final_response - ).strip() - _prefill_exhausted = ( - _has_structured and agent._thinking_prefill_retries >= 2 + # Empty-response retries: truly empty replies AND reasoning-only replies after + # prefill exhaustion. + _truly_empty = not agent._strip_think_blocks(final_response).strip() + _empty_candidate = _truly_empty and (not _has_structured or agent._thinking_prefill_retries >= 2) + action, interrupt_result, _deterministic_empty = _retry_empty( + agent, response, finish_reason, _empty_candidate, messages=messages, + conversation_history=conversation_history, api_call_count=api_call_count, ) - _empty_candidate = _truly_empty and ( - not _has_structured or _prefill_exhausted - ) - if _empty_candidate: - # Each empty attempt re-bills the full input; record its - # signature so deterministic empties stop burning paid retries. - # Fails open: missing usage or any output keeps the budget. - _empty_guard.record_empty_attempt( - agent, finish_reason=finish_reason, response=response - ) - _empty_retry_budget = ( - _empty_guard.empty_retry_budget(agent, response) - if _empty_candidate - else _empty_guard.DEFAULT_EMPTY_RETRY_BUDGET - ) - _deterministic_empty = _empty_candidate and ( - _empty_guard.deterministic_empty(agent) - ) - if ( - _empty_candidate - and agent._empty_content_retries < _empty_retry_budget - and not _deterministic_empty - ): - agent._empty_content_retries += 1 - wait_time = jittered_backoff( - agent._empty_content_retries, base_delay=5.0, max_delay=60.0 - ) - logger.warning( - "Empty response (no content or reasoning) — " - "retry %d/%d in %.1fs (model=%s)", - agent._empty_content_retries, - _empty_retry_budget, wait_time, agent.model, - ) - _budget_note = ( - " — high-cost request, reduced retry budget" - if _empty_retry_budget < _empty_guard.DEFAULT_EMPTY_RETRY_BUDGET - else "" - ) - agent._buffer_status( - f"⚠️ Empty response from model — retrying " - f"({agent._empty_content_retries}/{_empty_retry_budget}) " - f"in {wait_time:.0f}s{_budget_note}" - ) - _interrupted = interruptible_backoff_sleep( - agent, wait_time, None, - messages=messages, - conversation_history=conversation_history, - api_call_count=api_call_count, - abort_message="Interrupt detected during empty-response retry wait, aborting.", - interrupt_text=( - f"Operation interrupted: retrying empty response from model " - f"(retry {agent._empty_content_retries}/{_empty_retry_budget})." - ), - activity_label=f"empty response retry backoff ({agent._empty_content_retries}/{_empty_retry_budget})", - ) - if _interrupted is not None: - return _verdict("return", _interrupted) - return _verdict("continue") + if action is not None: + return _verdict(action, interrupt_result) if _truly_empty and _deterministic_empty: logger.warning( - "Deterministic empty response detected " - "(consecutive zero-output completions, " - "model=%s provider=%s finish_reason=%s) — " - "skipping remaining retries", + "Deterministic empty response detected (consecutive zero-output completions, " + "model=%s provider=%s finish_reason=%s) — skipping remaining retries", agent.model, agent.provider, finish_reason, ) agent._buffer_status( - "⚠️ Model is deterministically returning empty " - "(zero output tokens) — skipping further retries " + "⚠️ Model is deterministically returning empty (zero output tokens) — skipping further retries " "to avoid repeat charges" ) - # ── Exhausted retries — try fallback provider ── - # Before "(empty)", switch to the next provider in the chain. + # Exhausted retries — try the next provider in the chain before "(empty)". if _truly_empty and agent._fallback_chain: logger.warning( - "Empty response after %d retries — " - "attempting fallback (model=%s, provider=%s)", - agent._empty_content_retries, agent.model, - agent.provider, - ) - agent._buffer_status( - "⚠️ Model returning empty responses — " "switching to fallback provider..." + "Empty response after %d retries — attempting fallback (model=%s, provider=%s)", + agent._empty_content_retries, agent.model, agent.provider, ) + agent._buffer_status("⚠️ Model returning empty responses — " "switching to fallback provider...") if agent._try_activate_fallback(): - active_system_prompt = _sync_failover_system_message( - agent, api_messages, active_system_prompt) + active_system_prompt = _sync_failover_system_message(agent, api_messages, active_system_prompt) agent._empty_content_retries = 0 - agent._buffer_status( - f"↻ Switched to fallback: {agent.model} " f"({agent.provider})" - ) + agent._buffer_status(f"↻ Switched to fallback: {agent.model} " f"({agent.provider})") logger.info( - "Fallback activated after empty responses: " - "now using %s on %s", + "Fallback activated after empty responses: now using %s on %s", agent.model, agent.provider, ) - # OUTER loop: `continue` re-runs preflight against the - # fallback's window; `break` would end the turn without - # calling the fallback. Clear the preflight block. (#84733) + # OUTER loop: `continue` re-runs preflight against the fallback's window; + # `break` would end the turn without calling the fallback. _preflight_compression_blocked = False return _verdict("continue") - # Retries and fallback exhausted — fall through to "(empty)". - # Surface the buffered retry trace and, if known, what the empty - # streak cost (each attempt re-billed the full input). - _streak_cost = _empty_guard.streak_cost_usd(agent) - if _streak_cost is not None: - agent._buffer_status( - f"ℹ️ Estimated cost of these empty attempts: " - f"~${_streak_cost:.2f} (input tokens are billed " - f"per attempt even when no answer is produced)" - ) - agent._flush_status_buffer() _turn_exit_reason = "empty_response_exhausted" - reasoning_text = agent._extract_reasoning(assistant_message) - agent._drop_trailing_empty_response_scaffolding(messages) - assistant_msg = agent._build_assistant_message(assistant_message, finish_reason) - assistant_msg["content"] = "(empty)" - # Gateway failure sentinel, not content: persisting it lets later - # "continue" turns replay assistant("(empty)") and loop on empties. - assistant_msg["_empty_terminal_sentinel"] = True - append_message(messages, assistant_msg) - - if reasoning_text: - reasoning_preview = reasoning_text[:500] + "..." if len(reasoning_text) > 500 else reasoning_text - logger.warning( - "Reasoning-only response (no visible content) " - "after exhausting retries and fallback. " - "Reasoning: %s", reasoning_preview, - ) - agent._emit_status( - "⚠️ Model produced reasoning but no visible " - "response after all retries. Returning empty." - ) - else: - logger.warning( - "Empty response (no content or reasoning) " - "after %d retries. No fallback available. " - "model=%s provider=%s", - agent._empty_content_retries, agent.model, - agent.provider, - ) - agent._emit_status( - "❌ Model returned no content after all retries" - + (" and fallback attempts." if agent._fallback_chain else - ". No fallback providers configured.") - ) - - # Delivery-only: show labeled reasoning instead of bare "(empty)" - # when the model thought but wrote no text. The persisted row keeps - # the sentinel; reasoning is never promoted earlier in the ladder. - if reasoning_text: - final_response = ( - "⚠️ The model produced only internal reasoning and " - "no final answer, despite retries" - + (" and fallback" if agent._fallback_chain else "") - + ". Its last reasoning, which may contain the " - "answer:\n\n" + reasoning_preview - ) - else: - final_response = "(empty)" + final_response = _terminal_empty(agent, assistant_message, finish_reason, messages) return _verdict("break") - return _verdict("fallthrough") diff --git a/agent/turn_overflow.py b/agent/turn_overflow.py index e74855a084..dca76b10a4 100644 --- a/agent/turn_overflow.py +++ b/agent/turn_overflow.py @@ -1,11 +1,10 @@ """Overflow recovery for the conversation turn loop: 413 payload-too-large and context-length errors after ``classify_api_error``. -Extracted from ``run_conversation``'s ``except`` branch. Each path either compresses and -signals a restart, defers softly (compression lock / transient block), or ends the turn -with a typed result. Nothing here imports ``agent.conversation_loop`` at module level -(cycle); loop-internal helpers and the token estimators that tests patch on the loop -module are imported lazily inside the handlers so they keep resolving through the loop. +Each path either compresses and signals a restart, defers softly (compression lock / +transient block), or ends the turn with a typed result. Nothing here imports +``agent.conversation_loop`` at module level (cycle); loop-internal helpers and the token +estimators that tests patch on the loop module are imported lazily inside the handlers. """ from __future__ import annotations @@ -13,7 +12,7 @@ from __future__ import annotations import logging import time from dataclasses import dataclass -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple from agent.conversation_compression import ( COMPRESSION_RETRY_MESSAGES_STATUS_TEMPLATE, COMPRESSION_RETRY_TOKENS_STATUS_TEMPLATE, @@ -41,6 +40,8 @@ _GITHUB_MODELS_HINT = ( " setup` → GitHub Copilot), or pick any other provider.", ) +_MINIMAX_ANTHROPIC_URLS = ("https://api.minimax.io/anthropic", "https://api.minimaxi.com/anthropic") + @dataclass class OverflowVerdict: @@ -63,10 +64,11 @@ class OverflowVerdict: is_context_length_error: bool -@dataclass -class _Recovery: - """Mutable working state shared by the overflow sub-handlers; the loop locals - they may rebind live here and are handed back through ``verdict()``.""" +@dataclass(kw_only=True) +class _Recovery(OverflowVerdict): + """Working state shared by the overflow sub-handlers — the verdict itself, plus the + read-only call context; handlers mutate the loop-local fields and ``done()`` stamps + the action.""" agent: Any api_messages: Any @@ -74,23 +76,14 @@ class _Recovery: effective_task_id: Any api_call_count: int max_compression_attempts: int - messages: List[Dict[str, Any]] - active_system_prompt: Any - conversation_history: Any - approx_tokens: int - compression_attempts: int + action: str = "fallthrough" + result: Optional[Dict[str, Any]] = None provider_overflow_recovery_pending: bool = False is_context_length_error: bool = False - def verdict(self, action: str, result: Optional[Dict[str, Any]] = None) -> OverflowVerdict: - return OverflowVerdict( - action=action, result=result, messages=self.messages, - active_system_prompt=self.active_system_prompt, - conversation_history=self.conversation_history, approx_tokens=self.approx_tokens, - compression_attempts=self.compression_attempts, - provider_overflow_recovery_pending=self.provider_overflow_recovery_pending, - is_context_length_error=self.is_context_length_error, - ) + def done(self, action: str, result: Optional[Dict[str, Any]] = None) -> OverflowVerdict: + self.action, self.result = action, result + return self def fail_turn( self, final_response: str, *, notices: tuple = (), log: Optional[tuple] = None, @@ -118,11 +111,14 @@ class _Recovery: if compression_exhausted: result["compression_exhausted"] = True result.update(extra) - return self.verdict("return", result) + return self.done("return", result) - def exhausted(self, *, payload_too_large: bool = False) -> OverflowVerdict: - """Terminal: ``compression_attempts`` exceeded ``max_compression_attempts``.""" + def count_attempt(self, *, payload_too_large: bool = False) -> Optional[OverflowVerdict]: + """Bump ``compression_attempts``; the terminal verdict once the cap is exceeded.""" + self.compression_attempts += 1 cap = self.max_compression_attempts + if self.compression_attempts <= cap: + return None if payload_too_large: return self.fail_turn( f"Request payload too large: max compression attempts ({cap}) reached.", @@ -140,13 +136,13 @@ class _Recovery: def compress(self, request_tokens: int, *, fail_on_timeout: bool = False) -> Optional[OverflowVerdict]: """One compression pass with the summary-failure cooldown bypassed (the - provider proved the request doesn't fit, #100661). Returns ``None`` when - history was compressed, or a soft-defer verdict when another path holds the - compression lock (#69870) or a timed guard no-oped the pass (#97488): the - attempt is refunded and the turn ends WITHOUT ``compression_exhausted`` so the - gateway does not auto-reset. With ``fail_on_timeout`` a host timeout (recovery - spent its wait budget with no committed summary) ends the turn via the typed - contract, since re-sending would hit the same overflow (#98722).""" + provider proved the request doesn't fit). Returns ``None`` when history was + compressed, or a soft-defer verdict when another path holds the compression + lock or a timed guard no-oped the pass: the attempt is refunded and the turn + ends WITHOUT ``compression_exhausted`` so the gateway does not auto-reset. With + ``fail_on_timeout`` a host timeout (recovery spent its wait budget with no + committed summary) ends the turn via the typed contract, since re-sending would + hit the same overflow.""" from agent.conversation_loop import ( _COMPRESSION_TIMEOUT_FINAL_RESPONSE, _compression_deferred_result, conversation_history_after_compression, @@ -169,7 +165,7 @@ class _Recovery: if deferred is not None: self.compression_attempts -= 1 agent._persist_session(self.messages, self.conversation_history) - return self.verdict("return", deferred) + return self.done("return", deferred) if fail_on_timeout and context_compression_timed_out(agent): return self.fail_turn( _COMPRESSION_TIMEOUT_FINAL_RESPONSE, turn_exit_reason="context_compression_timeout" @@ -179,6 +175,29 @@ class _Recovery: ) return None + def compress_scored_by_tokens( + self, request_tokens: int, *, fail_on_timeout: bool = False, + ) -> Tuple[Optional[OverflowVerdict], bool, int]: + """``compress`` scored in message count / tokens (context-overflow errors ARE + token-budget errors). Same-message-count compression (tool-result pruning, + in-place summarization) can shrink the request, so re-estimate rather than trust + the array length. Returns ``(deferred_verdict, shrank, new_tokens)``.""" + from agent.conversation_loop import estimate_messages_tokens_rough + + original_len = len(self.messages) + original_tokens = estimate_messages_tokens_rough(self.messages) + deferred = self.compress(request_tokens, fail_on_timeout=fail_on_timeout) + if deferred is not None: + return deferred, False, original_tokens + messages = self.messages + new_tokens = estimate_messages_tokens_rough(messages) + shrank_tokens = new_tokens > 0 and new_tokens < original_tokens * 0.95 + if len(messages) < original_len: + self.agent._buffer_status(COMPRESSION_RETRY_MESSAGES_STATUS_TEMPLATE.format(before=original_len, after=len(messages))) + elif shrank_tokens: + self.agent._buffer_status(COMPRESSION_RETRY_TOKENS_STATUS_TEMPLATE.format(before=original_tokens, after=new_tokens)) + return None, len(messages) < original_len or shrank_tokens, new_tokens + def request_tokens(self) -> int: """Overhead-aware request size (msgs + tools + system) so LCM forced-overflow recovery arms on the TRUE request, not the tool-blind message count.""" @@ -190,13 +209,13 @@ class _Recovery: def _recover_payload_too_large(st: _Recovery, _retry: TurnRetryState) -> OverflowVerdict: """413: compress and retry. A 413 is a BYTE-size error, so progress is scored in payload bytes — never the token estimate, which is deliberately byte-blind to images - and wedged sessions on "no progress" (#88960 / #47339).""" + and wedged sessions on "no progress".""" from agent.conversation_loop import estimate_messages_tokens_rough agent = st.agent - st.compression_attempts += 1 - if st.compression_attempts > st.max_compression_attempts: - return st.exhausted(payload_too_large=True) + exhausted = st.count_attempt(payload_too_large=True) + if exhausted is not None: + return exhausted agent._buffer_status( f"⚠️ Request payload too large (413) — compression attempt " f"{st.compression_attempts}/{st.max_compression_attempts}..." @@ -223,14 +242,14 @@ def _recover_payload_too_large(st: _Recovery, _retry: TurnRetryState) -> Overflo ) time.sleep(2) # Brief pause between compression retries _retry.restart_with_compressed_messages = True - return st.verdict("break") + return st.done("break") if agent._try_strip_image_parts_from_tool_messages(st.api_messages, remember_model=False): agent._buffer_status( "📐 Compression could not reduce the request further — " "removed retained vision payloads and retrying..." ) - return st.verdict("continue") + return st.done("continue") return st.fail_turn( "Request payload too large (413). Cannot compress further.", @@ -243,8 +262,6 @@ def _clamp_output_cap(st: _Recovery, _retry: TurnRetryState, available_out: int, """Output-cap error ("max_tokens too large": input fits but input + max_tokens > window). The provider's available_tokens is the authoritative bound; also estimate the real request shape (API-only content) and use the smaller minus a margin.""" - from agent.conversation_loop import estimate_messages_tokens_rough - agent = st.agent request_input_estimate = st.request_tokens() local_available_out = old_ctx - request_input_estimate @@ -255,37 +272,27 @@ def _clamp_output_cap(st: _Recovery, _retry: TurnRetryState, available_out: int, safe_out = max(1, available_out - 64) agent._ephemeral_max_output_tokens = safe_out agent._buffer_vprint( - f"⚠️ Output cap too large for current prompt — " - f"retrying with max_tokens={safe_out:,} " - f"(provider_available={available_out:,}, " - f"estimated_request_tokens={request_input_estimate:,}; " + f"⚠️ Output cap too large for current prompt — retrying with max_tokens={safe_out:,} " + f"(provider_available={available_out:,}, estimated_request_tokens={request_input_estimate:,}; " f"context_length unchanged at {old_ctx:,})" ) # Still count against compression_attempts so a recurring error can't loop forever. - st.compression_attempts += 1 - if st.compression_attempts > st.max_compression_attempts: - return st.exhausted() + exhausted = st.count_attempt() + if exhausted is not None: + return exhausted # Also compress history so the retry doesn't spin on max_tokens alone; dropping the - # middle window makes the total fit (#55546). Compression must never turn an - # output-cap error fatal — on error, fall through and retry on max_tokens alone. + # middle window makes the total fit. Compression must never turn an output-cap error + # fatal — on error, fall through and retry on max_tokens alone. try: - original_len = len(st.messages) - original_tokens = estimate_messages_tokens_rough(st.messages) - deferred = st.compress(request_input_estimate) + deferred, _shrank, _new_tokens = st.compress_scored_by_tokens(request_input_estimate) if deferred is not None: return deferred - messages = st.messages - new_tokens = estimate_messages_tokens_rough(messages) - if len(messages) < original_len: - agent._buffer_status(COMPRESSION_RETRY_MESSAGES_STATUS_TEMPLATE.format(before=original_len, after=len(messages))) - elif new_tokens > 0 and new_tokens < original_tokens * 0.95: - agent._buffer_status(COMPRESSION_RETRY_TOKENS_STATUS_TEMPLATE.format(before=original_tokens, after=new_tokens)) except Exception: logger.warning( "%sOutput-cap compression hit an error; retrying on max_tokens only.", agent.log_prefix ) _retry.restart_with_compressed_messages = True - return st.verdict("break") + return st.done("break") def _adopt_provider_context_limit(st: _Recovery, error_msg: str, old_ctx: int) -> Optional[int]: @@ -315,13 +322,9 @@ def _adopt_provider_context_limit(st: _Recovery, error_msg: str, old_ctx: int) - agent._buffer_vprint(f"⚠️ Context length exceeded — using provider limit: {old_ctx:,} → {new_ctx:,} tokens") return new_ctx - _provider_lower = (getattr(agent, "provider", "") or "").lower() - _base_lower = (getattr(agent, "base_url", "") or "").rstrip("/").lower() is_minimax_provider = ( - _provider_lower in {"minimax", "minimax-cn"} - or _base_lower.startswith(( - "https://api.minimax.io/anthropic", "https://api.minimaxi.com/anthropic" - )) + (getattr(agent, "provider", "") or "").lower() in {"minimax", "minimax-cn"} + or (getattr(agent, "base_url", "") or "").rstrip("/").lower().startswith(_MINIMAX_ANTHROPIC_URLS) ) if is_minimax_provider and "context window exceeds limit (" in error_msg: agent._buffer_vprint( @@ -340,8 +343,6 @@ def _recover_context_length(st: _Recovery, _retry: TurnRetryState, error_msg: st """Context-length error. Two shapes: "prompt too long" = INPUT overflows the window (shrink context_length + compress); "max_tokens too large" = input fits but input + max_tokens > window (shrink the OUTPUT cap only).""" - from agent.conversation_loop import estimate_messages_tokens_rough - agent = st.agent old_ctx = agent.context_compressor.context_length @@ -350,18 +351,15 @@ def _recover_context_length(st: _Recovery, _retry: TurnRetryState, error_msg: st return _clamp_output_cap(st, _retry, available_out, old_ctx) # Output-cap error with unparseable budget: compression can't help (input already - # fits) and would death-loop on the same 400. Fail fast (#55546). + # fits) and would death-loop on the same 400. Fail fast. if is_output_cap_error(error_msg): return st.fail_turn( "max_tokens exceeds the provider's output cap for this model. " "Lower model.max_tokens in config.yaml.", notices=( - "❌ The provider rejected the request because " - "max_tokens exceeds its output cap for this model.", - " 💡 Lower model.max_tokens in your config.yaml to " - "at or below the model's max-output limit. " - "(This is an output-cap error, not a context overflow — " - "compression cannot fix it.)", + "❌ The provider rejected the request because max_tokens exceeds its output cap for this model.", + " 💡 Lower model.max_tokens in your config.yaml to at or below the model's max-output limit. " + "(This is an output-cap error, not a context overflow — compression cannot fix it.)", ), log=( f"{agent.log_prefix}Output-cap error not routed into compression " @@ -372,36 +370,24 @@ def _recover_context_length(st: _Recovery, _retry: TurnRetryState, error_msg: st new_ctx = _adopt_provider_context_limit(st, error_msg, old_ctx) - st.compression_attempts += 1 - if st.compression_attempts > st.max_compression_attempts: - return st.exhausted() + exhausted = st.count_attempt() + if exhausted is not None: + return exhausted agent._buffer_status(COMPRESSION_RETRY_TOO_LARGE_STATUS_TEMPLATE.format( tokens=st.approx_tokens, attempt=st.compression_attempts, cap=st.max_compression_attempts, )) - original_len = len(st.messages) - original_tokens = estimate_messages_tokens_rough(st.messages) - deferred = st.compress(st.request_tokens(), fail_on_timeout=True) + deferred, shrank, new_tokens = st.compress_scored_by_tokens(st.request_tokens(), fail_on_timeout=True) if deferred is not None: return deferred - - # Re-estimate: same-message-count compression (tool-result pruning, in-place - # summarization) can shrink the request (#39550). - messages = st.messages - new_tokens = estimate_messages_tokens_rough(messages) st.approx_tokens = new_tokens - shrank_tokens = new_tokens > 0 and new_tokens < original_tokens * 0.95 - if len(messages) < original_len or shrank_tokens or (new_ctx and new_ctx < old_ctx): - if len(messages) < original_len: - agent._buffer_status(COMPRESSION_RETRY_MESSAGES_STATUS_TEMPLATE.format(before=original_len, after=len(messages))) - elif shrank_tokens: - agent._buffer_status(COMPRESSION_RETRY_TOKENS_STATUS_TEMPLATE.format(before=original_tokens, after=new_tokens)) + if shrank or (new_ctx and new_ctx < old_ctx): time.sleep(2) # Brief pause between compression retries # Rebuild the full request and force normal preflight to honor it; message # count alone doesn't prove system/tool-inclusive pressure fell. st.provider_overflow_recovery_pending = True _retry.restart_with_compressed_messages = True - return st.verdict("break") + return st.done("break") # Can't compress further and already at minimum tier. return st.fail_turn( @@ -450,12 +436,12 @@ def recover_from_overflow( return _recover_payload_too_large(st, _retry) # Relay-wrapped output-cap 429s (parsed by the caller) go to the clamp, not - # failover or generic retries (#72281). The classifier also covers - # 400/disconnect + large-session heuristics. + # failover or generic retries. The classifier also covers 400/disconnect + + # large-session heuristics. st.is_context_length_error = ( classified.reason == FailoverReason.context_overflow or wrapped_output_cap_budget is not None ) if st.is_context_length_error: return _recover_context_length(st, _retry, error_msg) - return st.verdict("fallthrough") + return st.done("fallthrough") diff --git a/agent/turn_request_assembly.py b/agent/turn_request_assembly.py index d5c18d06e3..cb5490e368 100644 --- a/agent/turn_request_assembly.py +++ b/agent/turn_request_assembly.py @@ -2,17 +2,16 @@ from the transcript, append MoA context, inject prefills, run the context-engine selection hook and the send-time sanitizers, canonicalize for bit-perfect cache prefixes, build the request-local prompt-cache plan LAST (after every transcript mutation), prepare the -persistent-MoA request, then measure request pressure. Extracted from -``run_conversation``; nothing here imports ``agent.conversation_loop`` at module level -(cycle) — loop-internal helpers resolve lazily so ``patch("agent.conversation_loop.X")`` -sites keep intercepting. +persistent-MoA request, then measure request pressure. Nothing here imports +``agent.conversation_loop`` at module level (cycle) — loop-internal helpers resolve lazily +so ``patch("agent.conversation_loop.X")`` sites keep intercepting. """ from __future__ import annotations from dataclasses import dataclass import logging -from typing import Any, Dict, Optional +from typing import Any from agent.message_sanitization import _sanitize_messages_surrogates from agent.model_metadata import anchored_context_tokens @@ -38,6 +37,72 @@ class AssembledRequest: total_chars: Any +def _append_moa_context(agent: Any, api_messages: Any, moa_config: Any, original_user_message: Any) -> None: + """Run the MoA reference models and append their aggregated context to the last user + message (as a trailing text part on multimodal turns). Fail-open.""" + try: + from agent.message_content import flatten_message_text as _flatten_mt + from agent.moa_loop import _preset_temperature, aggregate_moa_context + + _moa_context = aggregate_moa_context( + user_prompt=( + original_user_message + if isinstance(original_user_message, str) + # Multimodal content list: extract visible text rather than + # str()-ing parts, which would leak base64 image payloads. + else _flatten_mt(original_user_message) + ), + api_messages=api_messages, + reference_models=moa_config.get("reference_models") or [], + aggregator=moa_config.get("aggregator") or {}, + temperature=_preset_temperature(moa_config, "reference_temperature"), + aggregator_temperature=_preset_temperature(moa_config, "aggregator_temperature"), + reference_max_tokens=moa_config.get("reference_max_tokens"), + # None = no per-preset override; inherit auxiliary.moa_reference.timeout. + reference_timeout=( + float(moa_config["reference_timeout"]) + if moa_config.get("reference_timeout") + else None + ), + degraded_reference_policy=str( + moa_config.get("degraded_reference_policy") or "loud" + ), + agent=agent, + ) + if not _moa_context: + return + for _msg in reversed(api_messages): + if _msg.get("role") == "user": + _base = _msg.get("content", "") + if isinstance(_base, str): + _msg["content"] = _base + "\n\n" + _moa_context + elif isinstance(_base, list): + _msg["content"] = [*_base, {"type": "text", "text": "\n\n" + _moa_context}] + break + except Exception as _moa_exc: + logger.warning("MoA context aggregation failed: %s", _moa_exc) + + +def _prepare_moa_request(agent: Any, api_messages: Any, pending_moa_prepared_request: Any) -> tuple: + """Persistent-MoA request: rebase the pending prepared request onto the new messages + when the client supports it, else prepare a fresh one. Returns + ``(prepared_request, api_messages, pending_moa_prepared_request)``.""" + _moa_completions = getattr(getattr(agent.client, "chat", None), "completions", None) + prepared: Any = None + if pending_moa_prepared_request is not None: + _rebase = getattr(_moa_completions, "rebase_prepared_request", None) + if callable(_rebase): + prepared = _rebase(pending_moa_prepared_request, api_messages) + pending_moa_prepared_request = None + if prepared is None: + _prepare = getattr(_moa_completions, "prepare", None) + if callable(_prepare): + prepared = _prepare(api_messages) + if prepared is not None: + api_messages = prepared["messages"] + return prepared, api_messages, pending_moa_prepared_request + + def assemble_api_request( agent: Any, *, messages: Any, current_turn_user_idx: Any, _ext_prefetch_cache: Any, _plugin_user_context: Any, moa_config: Any, active_system_prompt: Any, @@ -51,19 +116,6 @@ def assemble_api_request( _midturn_request_pressure_tokens, estimate_messages_tokens_rough, ) - def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> AssembledRequest: - return AssembledRequest( - action=action, - api_messages=api_messages, - tools_for_api=tools_for_api, - _moa_prepared_request=_moa_prepared_request, - pending_moa_prepared_request=pending_moa_prepared_request, - approx_tokens=approx_tokens, - request_pressure_tokens=request_pressure_tokens, - total_chars=total_chars, - - ) - api_messages, effective_system = build_api_messages( agent, messages, current_turn_user_idx=current_turn_user_idx, ext_prefetch_cache=_ext_prefetch_cache, plugin_user_context=_plugin_user_context, @@ -71,54 +123,9 @@ def assemble_api_request( ) if moa_config: - try: - from agent.message_content import flatten_message_text as _flatten_mt - from agent.moa_loop import _preset_temperature, aggregate_moa_context + _append_moa_context(agent, api_messages, moa_config, original_user_message) - _moa_context = aggregate_moa_context( - user_prompt=( - original_user_message - if isinstance(original_user_message, str) - # Multimodal content list: extract visible text rather than - # str()-ing parts, which would leak base64 image payloads. - else _flatten_mt(original_user_message) - ), - api_messages=api_messages, - reference_models=moa_config.get("reference_models") or [], - aggregator=moa_config.get("aggregator") or {}, - temperature=_preset_temperature(moa_config, "reference_temperature"), - aggregator_temperature=_preset_temperature(moa_config, "aggregator_temperature"), - reference_max_tokens=moa_config.get("reference_max_tokens"), - # None = no per-preset override; inherit - # auxiliary.moa_reference.timeout via call_llm. - reference_timeout=( - float(moa_config["reference_timeout"]) - if moa_config.get("reference_timeout") - else None - ), - degraded_reference_policy=str( - moa_config.get("degraded_reference_policy") or "loud" - ), - agent=agent, - ) - if _moa_context: - for _msg in reversed(api_messages): - if _msg.get("role") == "user": - _base = _msg.get("content", "") - if isinstance(_base, str): - _msg["content"] = _base + "\n\n" + _moa_context - elif isinstance(_base, list): - # Multimodal turn: append MoA context as a trailing text - # part instead of silently dropping it. - _msg["content"] = [ - *_base, {"type": "text", "text": "\n\n" + _moa_context} - ] - break - except Exception as _moa_exc: - logger.warning("MoA context aggregation failed: %s", _moa_exc) - - # Inject ephemeral prefill messages right after the system prompt - # but before conversation history. Same API-call-time-only pattern. + # Ephemeral prefill messages go right after the system prompt, API-call-time only. if agent.prefill_messages: sys_offset = 1 if (api_messages and api_messages[0].get("role") == "system") else 0 for idx, pfm in enumerate(agent.prefill_messages): @@ -140,17 +147,14 @@ def assemble_api_request( api_messages = agent._sanitize_api_messages(api_messages) # One-time repeated-heal notice goes out via the status/warning callback, NEVER - # appended to messages: the cached prompt prefix stays byte-identical (#96870). + # appended to messages: the cached prompt prefix stays byte-identical. try: - from agent.agent_runtime_helpers import ( - consume_pending_sanitizer_heal_notice, - ) + from agent.agent_runtime_helpers import consume_pending_sanitizer_heal_notice _heal_notice = consume_pending_sanitizer_heal_notice() if _heal_notice: agent._emit_warning(_heal_notice) except Exception: - # A notice hiccup must never break the send path. logger.debug("sanitizer heal notice delivery failed", exc_info=True) # Drop thinking-only assistant turns + merge adjacent users, API copy only: @@ -171,24 +175,21 @@ def assemble_api_request( _sanitize_messages_surrogates(api_messages) # No send-time pad loop here: ``repair_empty_non_final_messages`` (inside - # ``_sanitize_api_messages``) is the single owner of empty-turn repair, and its - # non-whitespace placeholder survives normalization regardless of ordering. + # ``_sanitize_api_messages``) is the single owner of empty-turn repair. # Build the request-local cache sections LAST, after every transcript mutation; # the canonical tool registry stays undecorated. Marked ``content`` becomes text # blocks the whitespace pass skips, so the same row's bytes vary across turns. tools_for_api = agent.tools if agent._use_prompt_caching and agent.provider != "moa": - from agent.prompt_caching import ( - envelope_tool_part_cache_markers_supported, - ) + from agent.prompt_caching import envelope_tool_part_cache_markers_supported _static_system_prefix = getattr(agent, "_cached_system_prompt_static", None) _initial_cache_plan = build_prompt_cache_plan( api_messages, tools_for_api, # Clamp per-destination: a configured 1h regresses to 5m on - # Qwen/Alibaba routes, whose context cache is 5m-only (#84733). + # Qwen/Alibaba routes, whose context cache is 5m-only. cache_ttl=effective_cache_ttl( agent._cache_ttl, provider=agent.provider, model=agent.model ), @@ -198,7 +199,7 @@ def assemble_api_request( ), direct_native_tool_cache=agent._direct_native_anthropic_tool_cache_capability(), # LiteLLM-style envelope routes forward part-level markers into - # tool_result.content[] → non-retryable 400 (#89886). + # tool_result.content[] → non-retryable 400. tool_part_markers=envelope_tool_part_cache_markers_supported( getattr(agent, "provider", ""), getattr(agent, "base_url", "") ), @@ -211,50 +212,35 @@ def assemble_api_request( # prepared request instead of running the advisors again. _moa_prepared_request = None if agent.provider == "moa": - _moa_completions = getattr(getattr(agent.client, "chat", None), "completions", None) - if pending_moa_prepared_request is not None: - _rebase_moa_request = getattr(_moa_completions, "rebase_prepared_request", None) - if callable(_rebase_moa_request): - _moa_prepared_request = _rebase_moa_request( - pending_moa_prepared_request, api_messages - ) - pending_moa_prepared_request = None - if _moa_prepared_request is None: - _prepare_moa_request = getattr(_moa_completions, "prepare", None) - if callable(_prepare_moa_request): - _moa_prepared_request = _prepare_moa_request(api_messages) - if _moa_prepared_request is not None: - api_messages = _moa_prepared_request["messages"] + _moa_prepared_request, api_messages, pending_moa_prepared_request = _prepare_moa_request( + agent, api_messages, pending_moa_prepared_request + ) # One image-stripped estimate feeds both figures; tools counted separately (50+ # tools ≈ 20-30K tokens); total_chars is a rough proxy for logs/hooks only. - # Charge stale thinking only when the active route replays it (#84371). + # Charge stale thinking only when the active route replays it. from agent.turn_context import _agent_stale_thinking_on_wire if _agent_stale_thinking_on_wire(agent): approx_tokens = estimate_messages_tokens_rough(api_messages) else: - approx_tokens = estimate_messages_tokens_rough( - api_messages, charge_stale_thinking=False - ) + approx_tokens = estimate_messages_tokens_rough(api_messages, charge_stale_thinking=False) # Route-aware: native Responses compaction prunes the wire payload, so the raw - # history figure overstates it and fires needless local compression (#96995). + # history figure overstates it and fires needless local compression. request_pressure_tokens = _midturn_request_pressure_tokens( agent, api_messages, effective_system or "", approx_tokens ) # Usage-anchored override: real prompt_tokens (incl. system + tool schemas) + # delta estimate replaces the whole-history heuristic when the anchor is fresh. - _anchored_pressure = anchored_context_tokens( - messages, getattr(agent, "_usage_anchor", None) - ) + _anchored_pressure = anchored_context_tokens(messages, getattr(agent, "_usage_anchor", None)) if _anchored_pressure is not None: request_pressure_tokens = _anchored_pressure - total_chars = approx_tokens * 4 # Stash the rough estimate so update_from_response() can pair it with the real # count (should_defer_preflight_to_real_usage). getattr: test doubles lack it. - _note_rough = getattr( - agent.context_compressor, "note_request_rough_estimate", None - ) + _note_rough = getattr(agent.context_compressor, "note_request_rough_estimate", None) if callable(_note_rough): _note_rough(request_pressure_tokens) - return _verdict("fallthrough") + return AssembledRequest( + "fallthrough", api_messages, tools_for_api, _moa_prepared_request, + pending_moa_prepared_request, approx_tokens, request_pressure_tokens, approx_tokens * 4, + ) diff --git a/agent/turn_response_check.py b/agent/turn_response_check.py index 05499b2030..cc04110267 100644 --- a/agent/turn_response_check.py +++ b/agent/turn_response_check.py @@ -2,9 +2,8 @@ spinner, validate the response shape (retry / eager fallback / terminal invalid-response result), derive ``finish_reason`` per api_mode, route content-policy refusals and ``length`` truncation, fold usage into the compressor, and mark the logical relay call -complete. Extracted from ``run_conversation``; nothing here imports -``agent.conversation_loop`` at module level (cycle) — loop-internal helpers resolve lazily so -existing ``patch("agent.conversation_loop.X")`` sites keep intercepting. +complete. Nothing here imports ``agent.conversation_loop`` at module level (cycle) — +loop-internal helpers resolve lazily so ``patch("agent.conversation_loop.X")`` keeps intercepting. """ from __future__ import annotations @@ -14,6 +13,7 @@ import logging import time from typing import Any, Dict, Optional +from agent.turn_api_call import stop_thinking_spinner from agent.turn_truncation import handle_content_policy_refusal, recover_from_truncation from agent.turn_usage import record_response_usage @@ -42,6 +42,45 @@ class ResponseCheckVerdict: result: Optional[Dict[str, Any]] = None +def _codex_finish_reason(response: Any) -> str: + """Responses API max-output exhaustion is a normal Codex incomplete turn: route it to + the Codex continuation path (``"incomplete"``), not the length rollback.""" + status = getattr(response, "status", None) + if isinstance(status, str): + status = status.strip().lower() + incomplete_details = getattr(response, "incomplete_details", None) + if isinstance(incomplete_details, dict): + incomplete_reason = incomplete_details.get("reason") + else: + incomplete_reason = getattr(incomplete_details, "reason", None) + if incomplete_reason is not None: + incomplete_reason = str(incomplete_reason).strip().lower() + if status == "incomplete" and incomplete_reason in {"max_output_tokens", "length"}: + return "incomplete" + if status == "incomplete" and incomplete_reason == "content_filter": + return "content_filter" + return "stop" + + +def _derive_finish_reason(agent: Any, response: Any, messages: Any) -> str: + if agent.api_mode == "codex_responses": + return _codex_finish_reason(response) + transport = agent._get_transport() + if agent.api_mode == "anthropic_messages": + return transport.map_finish_reason(response.stop_reason) + normalized = transport.normalize_response(response) # Bedrock already normalized at dispatch + finish_reason = normalized.finish_reason + if agent.api_mode != "bedrock_converse" and agent._should_treat_stop_as_truncated( + finish_reason, normalized, messages + ): + agent._vprint( + f"{agent.log_prefix}⚠️ Treating suspicious Ollama/GLM stop response as truncated", + force=True, + ) + return "length" + return finish_reason + + def check_api_response( agent: Any, *, response: Any, _retry: Any, thinking_spinner: Any, messages: Any, api_messages: Any, api_kwargs: Any, active_system_prompt: Any, conversation_history: Any, @@ -54,10 +93,7 @@ def check_api_response( """Verify ``response`` in the original order. The retry buffer is NOT cleared on success (bytes back != usable content); ``_preflight_compression_blocked``/``_last_preflight_pressure`` reset only when the usage fold re-arms the compression budget.""" - from agent.conversation_loop import ( - validate_response_shape, - ) - api_duration = None + from agent.conversation_loop import validate_response_shape def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> ResponseCheckVerdict: return ResponseCheckVerdict( @@ -74,25 +110,17 @@ def check_api_response( api_duration = time.time() - api_start_time - # Stop thinking spinner silently -- the response box or tool - # execution messages that follow are more informative. - if thinking_spinner: - thinking_spinner.stop("") - thinking_spinner = None - if agent.thinking_callback: - agent.thinking_callback("") + # Silent stop: the response box / tool messages that follow are more informative. + thinking_spinner = stop_thinking_spinner(agent, thinking_spinner) if not agent.quiet_mode: agent._vprint(f"{agent.log_prefix}⏱️ API call completed in {api_duration:.2f}s") if agent.verbose_logging: - # Log response with provider info if available resp_model = getattr(response, 'model', 'N/A') if response else 'N/A' logging.debug(f"API Response received - Model: {resp_model}, Usage: {response.usage if hasattr(response, 'usage') else 'N/A'}") - # Validate response shape before proceeding response_invalid, error_details = validate_response_shape(agent, response) - if response_invalid: _iv = retry_invalid_response( agent, response=response, error_details=error_details, _retry=_retry, @@ -112,54 +140,9 @@ def check_api_response( return _verdict(_iv.action, _iv.result) agent._turn_received_provider_response = True + finish_reason = _derive_finish_reason(agent, response, messages) - # Check finish_reason before proceeding - if agent.api_mode == "codex_responses": - status = getattr(response, "status", None) - if isinstance(status, str): - status = status.strip().lower() - incomplete_details = getattr(response, "incomplete_details", None) - incomplete_reason = None - if isinstance(incomplete_details, dict): - incomplete_reason = incomplete_details.get("reason") - else: - incomplete_reason = getattr(incomplete_details, "reason", None) - if incomplete_reason is not None: - incomplete_reason = str(incomplete_reason).strip().lower() - if status == "incomplete" and incomplete_reason in {"max_output_tokens", "length"}: - # Responses API max-output exhaustion is a normal Codex - # incomplete turn: use the Codex continuation path, not the - # length rollback. - finish_reason = "incomplete" - elif status == "incomplete" and incomplete_reason == "content_filter": - finish_reason = "content_filter" - else: - finish_reason = "stop" - elif agent.api_mode == "anthropic_messages": - _tfr = agent._get_transport() - finish_reason = _tfr.map_finish_reason(response.stop_reason) - elif agent.api_mode == "bedrock_converse": - # Bedrock response already normalized at dispatch — use transport - _bt_fr = agent._get_transport() - _bedrock_result = _bt_fr.normalize_response(response) - finish_reason = _bedrock_result.finish_reason - else: - _cc_fr = agent._get_transport() - _finish_result = _cc_fr.normalize_response(response) - finish_reason = _finish_result.finish_reason - assistant_message = _finish_result - if agent._should_treat_stop_as_truncated( - finish_reason, assistant_message, messages - ): - agent._vprint( - f"{agent.log_prefix}⚠️ Treating suspicious Ollama/GLM stop response as truncated", - force=True, - ) - finish_reason = "length" - - # ── Content-policy refusal (HTTP 200) ────────────────── - # Refusal finish reasons (``content_filter``, ``guardrail_intervened``) - # are deterministic: one fallback try, else return the refusal. + # HTTP-200 refusals are deterministic: one fallback try, else return the refusal. if finish_reason == "content_filter": _rv = handle_content_policy_refusal( agent, response, _retry, thinking_spinner=thinking_spinner, messages=messages, @@ -194,12 +177,8 @@ def check_api_response( truncated_tool_call_retries = _tv.truncated_tool_call_retries retry_count = _tv.retry_count compression_attempts = _tv.compression_attempts - if _tv.action == "return": - return _verdict("return", _tv.result) - if _tv.action == "break": - return _verdict("break") - if _tv.action == "continue": - return _verdict("continue") + if _tv.action in ("return", "break", "continue"): + return _verdict(_tv.action, _tv.result) # Fold provider usage into compressor / anchors / session counters / state.db # (agent/turn_usage.py). A rearmed budget also clears the preflight-block latch. @@ -213,10 +192,8 @@ def check_api_response( _preflight_compression_blocked = False _last_preflight_pressure = None - _retry.has_retried_429 = False # Reset on success - # Don't clear the retry buffer: bytes back != usable content; it is - # cleared once genuine content lands. Clearing Nous rate-limit state - # proves the limit reset so other sessions may resume. + _retry.has_retried_429 = False + # Clearing Nous rate-limit state proves the limit reset so other sessions may resume. if agent.provider == "nous": try: from agent.nous_rate_guard import clear_nous_rate_limit @@ -225,12 +202,9 @@ def check_api_response( pass from agent import relay_llm - relay_llm.complete_logical_call( - api_request_id, outcome="success" - ) + relay_llm.complete_logical_call(api_request_id, outcome="success") agent._touch_activity(f"API call #{api_call_count} completed") - return _verdict("break") # Success, exit retry loop - return _verdict("fallthrough") + return _verdict("break") @dataclass @@ -278,20 +252,11 @@ def retry_invalid_response( status_code=getattr(getattr(response, "error", None), "code", None), retry_count=retry_count, max_retries=max_retries, retryable=True, reason="invalid_response", ) - # Stop spinner silently — retry status is now buffered - # and only surfaced if every retry+fallback exhausts. - if thinking_spinner: - thinking_spinner.stop("") - thinking_spinner = None - if agent.thinking_callback: - agent.thinking_callback("") - - # Invalid response — could be rate limiting, provider timeout, - # upstream server error, or malformed response. + # Retry status is buffered and only surfaced if every retry+fallback exhausts. + thinking_spinner = stop_thinking_spinner(agent, thinking_spinner) retry_count += 1 - # Eager fallback: empty/malformed responses often mean rate limiting - # — switch now instead of extended backoff. + # Eager fallback: empty/malformed responses often mean rate limiting. if agent._fallback_index < len(agent._fallback_chain): agent._buffer_status("⚠️ Empty/malformed response — switching to fallback...") if agent._try_activate_fallback(): @@ -304,15 +269,12 @@ def retry_invalid_response( error_msg, provider_name, _failure_hint = describe_invalid_response( agent, response, api_duration ) - agent._buffer_vprint(f"⚠️ Invalid API response (attempt {retry_count}/{max_retries}): {', '.join(error_details)}") agent._buffer_vprint(f" 🏢 Provider: {provider_name}") - cleaned_provider_error = agent._clean_error_message(error_msg) - agent._buffer_vprint(f" 📝 Provider message: {cleaned_provider_error}") + agent._buffer_vprint(f" 📝 Provider message: {agent._clean_error_message(error_msg)}") agent._buffer_vprint(f" ⏱️ {_failure_hint}") if retry_count >= max_retries: - # Try fallback before giving up if agent._has_pending_fallback(): agent._buffer_status(f"⚠️ Max retries ({max_retries}) for invalid responses — trying fallback...") if agent._try_activate_fallback(): @@ -333,17 +295,15 @@ def retry_invalid_response( "completed": False, "api_calls": api_call_count, "error": _final_response, - "failed": True # Mark as failure for filtering + "failed": True, }) - # Backoff before retry — jittered exponential: 5s base, 120s cap wait_time = jittered_backoff(retry_count, base_delay=5.0, max_delay=120.0) agent._buffer_vprint(f"⏳ Retrying in {wait_time:.1f}s ({_failure_hint})...") logger.warning("Invalid API response (retry %d/%d): %s | Provider: %s", retry_count, max_retries, ', '.join(error_details), provider_name) - # A redirect cancels only the live request; the helper preserves the - # pending correction (restart_with_redirected_messages) instead of - # destroying it with clear_interrupt(). + # A redirect cancels only the live request; the helper preserves the pending + # correction (restart_with_redirected_messages) instead of clear_interrupt()-ing it. _interrupted = interruptible_backoff_sleep( agent, wait_time, _retry, messages=messages, conversation_history=conversation_history, api_call_count=api_call_count, @@ -355,5 +315,4 @@ def retry_invalid_response( return _verdict("return", _interrupted) if _retry.restart_with_redirected_messages: return _verdict("break") # rebuild this iteration from the correction - return _verdict("continue") # Retry the API call - return _verdict("fallthrough") + return _verdict("continue") diff --git a/agent/turn_response_intake.py b/agent/turn_response_intake.py index 8ea50d6fce..8c4b5949ea 100644 --- a/agent/turn_response_intake.py +++ b/agent/turn_response_intake.py @@ -1,8 +1,7 @@ """Response intake for the conversation turn loop: normalize the raw provider response into the assistant message, splice agent-as-provider projections, fire ``post_api_request``, relay reasoning to the progress callback, and apply the incomplete-scratchpad / Codex-incomplete -continuation guards. Extracted from ``run_conversation``; nothing here imports -``agent.conversation_loop`` at module level (cycle). +continuation guards. Nothing here imports ``agent.conversation_loop`` at module level (cycle). """ from __future__ import annotations @@ -15,10 +14,12 @@ from typing import Any, Dict, Optional from agent.provider_projection import splice_provider_projection from agent.trajectory import has_incomplete_scratchpad -from agent.turn_truncation import continue_codex_incomplete +from agent.turn_truncation import continue_codex_incomplete, normalize_response_for_agent, partial_result logger = logging.getLogger("agent.conversation_loop") +_REASONING_TAG_RE = re.compile(r'') + @dataclass class ResponseIntakeVerdict: @@ -32,68 +33,34 @@ class ResponseIntakeVerdict: result: Optional[Dict[str, Any]] = None -def normalize_model_response( - agent: Any, *, response: Any, messages: Any, api_messages: Any, conversation_history: Any, +def _coerce_content_text(raw: Any) -> str: + """Some OpenAI-compatible servers (llama-server) return content as dict/list, which + crashes downstream ``.strip()``; normalize to str (multimodal lists → text parts).""" + if isinstance(raw, dict): + return raw.get("text", "") or raw.get("content", "") or json.dumps(raw) + if isinstance(raw, list): + parts = [] + for part in raw: + if isinstance(part, str): + parts.append(part) + elif isinstance(part, dict) and part.get("type") == "text": + parts.append(part.get("text", "")) + elif isinstance(part, dict) and "text" in part: + parts.append(str(part["text"])) + return "\n".join(parts) + return str(raw) + + +def _fire_post_api_request_hook( + agent: Any, response: Any, assistant_message: Any, finish_reason: Any, *, api_messages: Any, api_call_count: Any, api_duration: Any, api_start_time: Any, api_request_id: Any, effective_task_id: Any, turn_id: Any, -) -> ResponseIntakeVerdict: - """Normalize ``response`` into ``assistant_message`` (str content, never dict/list) and run - the post-response hooks and continuation guards, in the original order.""" - from agent.conversation_loop import ( - _moa_reference_metrics_for_hook, - ) - assistant_message = None - finish_reason = None - - def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> ResponseIntakeVerdict: - return ResponseIntakeVerdict( - action=action, assistant_message=assistant_message, finish_reason=finish_reason, - result=result, - ) - - _transport = agent._get_transport() - _normalize_kwargs = {} - if agent.api_mode == "anthropic_messages": - _normalize_kwargs["strip_tool_prefix"] = agent._is_anthropic_oauth - normalized = _transport.normalize_response(response, **_normalize_kwargs) - assistant_message = normalized - finish_reason = normalized.finish_reason - - # Some OpenAI-compatible servers (llama-server) return content as dict/list, - # which crashes downstream .strip(); normalize to str. - if assistant_message.content is not None and not isinstance(assistant_message.content, str): - raw = assistant_message.content - if isinstance(raw, dict): - assistant_message.content = raw.get("text", "") or raw.get("content", "") or json.dumps(raw) - elif isinstance(raw, list): - # Multimodal content list — extract text parts - parts = [] - for part in raw: - if isinstance(part, str): - parts.append(part) - elif isinstance(part, dict) and part.get("type") == "text": - parts.append(part.get("text", "")) - elif isinstance(part, dict) and "text" in part: - parts.append(str(part["text"])) - assistant_message.content = "\n".join(parts) - else: - assistant_message.content = str(raw) - - # ── Agent-as-provider projection ────────────────────────────── - # Splice the provider-agent's own tool work in as call/result rows before - # this turn's assistant message; no-op for ordinary providers. - splice_provider_projection(agent, response, messages) +) -> None: + from agent.conversation_loop import _moa_reference_metrics_for_hook try: - from hermes_cli.lifecycle import ( - has_hook, invoke_hook as _invoke_hook - ) + from hermes_cli.lifecycle import has_hook, invoke_hook as _invoke_hook if has_hook("post_api_request"): - _assistant_tool_calls = ( - getattr(assistant_message, "tool_calls", None) or [] - ) - _assistant_text = assistant_message.content or "" - _api_ended_at = api_start_time + api_duration _invoke_hook( "post_api_request", task_id=effective_task_id, @@ -108,13 +75,10 @@ def normalize_model_response( api_call_count=api_call_count, api_duration=api_duration, started_at=api_start_time, - ended_at=_api_ended_at, - # First stream chunk time (epoch s) from - # interruptible_streaming_api_call; None if not streamed / no - # chunk. TTFB = first_chunk_at - started_at. - first_chunk_at=getattr( - agent, "_last_api_first_chunk_at", None - ), + ended_at=api_start_time + api_duration, + # First stream chunk time (epoch s); None if not streamed / no chunk. + # TTFB = first_chunk_at - started_at. + first_chunk_at=getattr(agent, "_last_api_first_chunk_at", None), finish_reason=finish_reason, message_count=len(api_messages), response_model=getattr(response, "model", None), @@ -123,73 +87,86 @@ def normalize_model_response( ), usage=agent._usage_summary_for_api_request_hook(response), assistant_message=assistant_message, - assistant_content_chars=len(_assistant_text), - assistant_tool_call_count=len(_assistant_tool_calls), + assistant_content_chars=len(assistant_message.content or ""), + assistant_tool_call_count=len(getattr(assistant_message, "tool_calls", None) or []), moa_references=_moa_reference_metrics_for_hook(agent), ) except Exception: pass - # Handle assistant response - if assistant_message.content and not agent.quiet_mode: + +def _relay_thinking(agent: Any, content: str) -> None: + """Relay the model's text to the progress callback: subagents send the first line to + the parent display; any agent with a structured callback gets ``reasoning.available``.""" + _think_text = _REASONING_TAG_RE.sub('', content.strip()).strip() + first_line = _think_text.split('\n')[0][:80] if _think_text else "" + if first_line and getattr(agent, '_delegate_depth', 0) > 0: + try: + agent.tool_progress_callback("_thinking", first_line) + except Exception: + pass + elif _think_text: + try: + agent.tool_progress_callback("reasoning.available", "_thinking", _think_text[:500], None) + except Exception: + pass + + +def normalize_model_response( + agent: Any, *, response: Any, messages: Any, api_messages: Any, conversation_history: Any, + api_call_count: Any, api_duration: Any, api_start_time: Any, api_request_id: Any, + effective_task_id: Any, turn_id: Any, +) -> ResponseIntakeVerdict: + """Normalize ``response`` into ``assistant_message`` (str content, never dict/list) and run + the post-response hooks and continuation guards, in the original order.""" + assistant_message = normalize_response_for_agent(agent, response) + finish_reason = assistant_message.finish_reason + + def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> ResponseIntakeVerdict: + return ResponseIntakeVerdict( + action=action, assistant_message=assistant_message, finish_reason=finish_reason, + result=result, + ) + + if assistant_message.content is not None and not isinstance(assistant_message.content, str): + assistant_message.content = _coerce_content_text(assistant_message.content) + + # Agent-as-provider projection: splice the provider-agent's own tool work in as + # call/result rows before this turn's assistant message; no-op for ordinary providers. + splice_provider_projection(agent, response, messages) + + _fire_post_api_request_hook( + agent, response, assistant_message, finish_reason, api_messages=api_messages, + api_call_count=api_call_count, api_duration=api_duration, api_start_time=api_start_time, + api_request_id=api_request_id, effective_task_id=effective_task_id, turn_id=turn_id, + ) + + content = assistant_message.content + if content and not agent.quiet_mode: if agent.verbose_logging: - agent._vprint(f"{agent.log_prefix}🤖 Assistant: {assistant_message.content}") + agent._vprint(f"{agent.log_prefix}🤖 Assistant: {content}") else: - agent._vprint(f"{agent.log_prefix}🤖 Assistant: {assistant_message.content[:100]}{'...' if len(assistant_message.content) > 100 else ''}") + agent._vprint(f"{agent.log_prefix}🤖 Assistant: {content[:100]}{'...' if len(content) > 100 else ''}") + if content and agent.tool_progress_callback: + _relay_thinking(agent, content) - # Notify progress callback of model's thinking (used by subagent - # delegation to relay the child's reasoning to the parent display). - if (assistant_message.content and agent.tool_progress_callback): - _think_text = assistant_message.content.strip() - # Strip reasoning XML tags that shouldn't leak to parent display - _think_text = re.sub( - r'', '', _think_text - ).strip() - # For subagents: relay first line to parent display (existing behaviour). - # For all agents with a structured callback: emit reasoning.available event. - first_line = _think_text.split('\n')[0][:80] if _think_text else "" - if first_line and getattr(agent, '_delegate_depth', 0) > 0: - try: - agent.tool_progress_callback("_thinking", first_line) - except Exception: - pass - elif _think_text: - try: - agent.tool_progress_callback("reasoning.available", "_thinking", _think_text[:500], None) - except Exception: - pass - - # Check for incomplete (opened but never closed) - # This means the model ran out of output tokens mid-reasoning — retry up to 2 times - if has_incomplete_scratchpad(assistant_message.content or ""): + # Incomplete (opened, never closed): the model ran out of + # output tokens mid-reasoning — retry up to 2 times, then save as partial. + if has_incomplete_scratchpad(content or ""): agent._incomplete_scratchpad_retries += 1 - agent._buffer_vprint("⚠️ Incomplete detected (opened but never closed)") - if agent._incomplete_scratchpad_retries <= 2: agent._buffer_vprint(f"🔄 Retrying API call ({agent._incomplete_scratchpad_retries}/2)...") - # Don't add the broken message, just retry - return _verdict("continue") - else: - # Max retries - discard this turn and save as partial - agent._flush_status_buffer() - agent._vprint(f"{agent.log_prefix}❌ Max retries (2) for incomplete scratchpad. Saving as partial.", force=True) - agent._incomplete_scratchpad_retries = 0 - - rolled_back_messages = agent._get_messages_up_to_last_assistant(messages) - agent._cleanup_task_resources(effective_task_id) - agent._persist_session(messages, conversation_history) - - return _verdict("return", { - "final_response": "Incomplete REASONING_SCRATCHPAD after 2 retries", - "messages": rolled_back_messages, - "api_calls": api_call_count, - "completed": False, - "partial": True, - "error": "Incomplete REASONING_SCRATCHPAD after 2 retries" - }) - - # Reset incomplete scratchpad counter on clean response + return _verdict("continue") # don't add the broken message + agent._flush_status_buffer() + agent._vprint(f"{agent.log_prefix}❌ Max retries (2) for incomplete scratchpad. Saving as partial.", force=True) + agent._incomplete_scratchpad_retries = 0 + rolled_back_messages = agent._get_messages_up_to_last_assistant(messages) + agent._cleanup_task_resources(effective_task_id) + agent._persist_session(messages, conversation_history) + return _verdict("return", partial_result( + rolled_back_messages, api_call_count, "Incomplete REASONING_SCRATCHPAD after 2 retries" + )) agent._incomplete_scratchpad_retries = 0 if agent.api_mode == "codex_responses" and finish_reason == "incomplete": @@ -200,6 +177,6 @@ def normalize_model_response( if _codex_result is not None: return _verdict("return", _codex_result) return _verdict("continue") - elif hasattr(agent, "_codex_incomplete_retries"): + if hasattr(agent, "_codex_incomplete_retries"): agent._codex_incomplete_retries = 0 return _verdict("fallthrough") diff --git a/agent/turn_truncation.py b/agent/turn_truncation.py index c7d5e063b0..8dd7ba548c 100644 --- a/agent/turn_truncation.py +++ b/agent/turn_truncation.py @@ -1,11 +1,10 @@ """Truncation recovery (``finish_reason == "length"``) for the conversation turn loop. -Extracted from ``run_conversation``. Handles thinking-budget exhaustion, repetition- -dominated truncation (#86581), content-filter stream stalls escalated to the fallback -chain (#32421), text continuation nudges (up to 4, with the ceiling exit that drops the -fragment trail), truncated tool-call retries with max_tokens boosts, and the final -roll-back. Nothing here imports ``agent.conversation_loop`` at module level (cycle); -loop-internal helpers are imported lazily so tests patching them on the loop keep working. +Handles thinking-budget exhaustion, repetition-dominated truncation, content-filter stream +stalls escalated to the fallback chain, text continuation nudges (up to 4, with the ceiling +exit that drops the fragment trail), truncated tool-call retries with max_tokens boosts, and +the final roll-back. Nothing here imports ``agent.conversation_loop`` at module level +(cycle); loop-internal helpers are imported lazily so tests patching them keep working. """ from __future__ import annotations @@ -19,11 +18,66 @@ from agent.error_classifier import FailoverReason from agent.message_metadata import append_message from agent.message_sanitization import close_interrupted_tool_sequence from agent.repetition_guard import is_repetition_dominated +from agent.turn_api_call import stop_thinking_spinner from agent.turn_retry_state import TurnRetryState from hermes_constants import PARTIAL_STREAM_STUB_ID logger = logging.getLogger("agent.conversation_loop") +_CONTINUABLE_MODES = {"chat_completions", "bedrock_converse", "anthropic_messages"} +_THINK_TAG_RE = re.compile(r'<(?:think|thinking|reasoning|REASONING_SCRATCHPAD)[^>]*>', re.IGNORECASE) +_TRUNCATED_FINAL = "Response truncated due to output length limit" +_FIRST_TRUNCATED_FINAL = "First response truncated due to output length limit" + +_THINKING_EXHAUSTED = ( + "💭 Reasoning exhausted the output token budget — no visible response was produced.", + "⚠️ **Thinking Budget Exhausted**\n\nThe model used all its output tokens on reasoning " + "and had none left for the actual response.\n\nTo fix this:\n" + "→ Lower reasoning effort: `/reasoning low` or `/reasoning minimal`\n" + "→ Or switch to a larger/non-reasoning model with `/model`", + "Model used all output tokens on reasoning with none left " + "for the response. Try lowering reasoning effort or increasing max_tokens.", +) +_REPETITION_DOMINATED = ( + "🔁 Response dominated by repeated text — stopping instead of continuing a degenerate response.", + "⚠️ **Response Stopped — Repetition Detected**\n\nThe model fell into a repetition loop while " + "writing this response, so continuing would only produce more repeated text. The partial response " + "was discarded.\n\n→ Switch to a different model with `/model`\n" + "→ Or resend your message (your conversation history is preserved)", + "Model output entered a repetition loop and was truncated mid-loop; refusing to continue a " + "degenerate response.", +) +_CEILING_NO_TEXT = ( + "⚠️ **No visible answer was produced.** The model hit its output-token limit on every " + "continuation attempt — its reasoning consumed the entire budget each time.\n\nTo fix this:\n" + "→ Lower reasoning effort: `/reasoning low` or `/reasoning none`\n→ Or raise max_tokens for this model" +) + + +def normalize_response_for_agent(agent: Any, response: Any) -> Any: + """One OpenAI-style message from any transport; Anthropic strips the OAuth tool prefix.""" + if agent.api_mode == "anthropic_messages": + return agent._get_transport().normalize_response( + response, strip_tool_prefix=agent._is_anthropic_oauth + ) + return agent._get_transport().normalize_response(response) + + +def partial_result( + messages: List[Dict[str, Any]], api_call_count: int, final_response: str, + error: Optional[str] = None, *, failed: bool = False, +) -> Dict[str, Any]: + """Typed incomplete-turn result (``partial`` unless ``failed``); ``error`` defaults to + ``final_response``.""" + return { + "final_response": final_response, + "messages": messages, + "api_calls": api_call_count, + "completed": False, + ("failed" if failed else "partial"): True, + "error": final_response if error is None else error, + } + @dataclass class TruncationVerdict: @@ -45,6 +99,206 @@ class TruncationVerdict: compression_attempts: int +@dataclass(kw_only=True) +class _Trunc(TruncationVerdict): + """Working state for the truncation phases — the verdict itself, plus the read-only + call context; phases mutate the loop-local fields and ``done()`` stamps the action.""" + + agent: Any + response: Any + finish_reason: str + conversation_history: Any + api_call_count: int + effective_task_id: Any + current_turn_user_idx: Any + action: str = "fallthrough" + result: Optional[Dict[str, Any]] = None + + def done(self, action: str, result: Optional[Dict[str, Any]] = None) -> TruncationVerdict: + self.action, self.result = action, result + return self + + def end_turn( + self, final_response: str, error: Optional[str] = None, *, + result_messages: Optional[List[Dict[str, Any]]] = None, cleanup: bool = True, + failed: bool = False, + ) -> TruncationVerdict: + """Persist and end the turn as partial (or ``failed``).""" + agent = self.agent + if cleanup: + agent._cleanup_task_resources(self.effective_task_id) + agent._persist_session(self.messages, self.conversation_history) + return self.done("return", partial_result( + self.messages if result_messages is None else result_messages, self.api_call_count, + final_response, error, failed=failed, + )) + + @property + def is_stub(self) -> bool: + return getattr(self.response, "id", "") == PARTIAL_STREAM_STUB_ID + + +def _abort_reason(agent: Any, content: Any, has_tool_calls: bool) -> Optional[tuple]: + """``(vprint, user response, error)`` when continuation must NOT be attempted: + thinking exhausted the budget (reasoning blocks with no visible text after them — + ``content=None`` from non- models is normal truncation), or a repetition loop + burned the budget on one fragment (reasoning stripped first).""" + if has_tool_calls: + return None + if content and _THINK_TAG_RE.search(content) and not agent._has_content_after_think_block(content): + return _THINKING_EXHAUSTED + visible = agent._strip_think_blocks(content) if isinstance(content, str) else content + if visible and is_repetition_dominated(visible): + return _REPETITION_DOMINATED + return None + + +def _content_filter_fallback(st: _Trunc, _retry: TurnRetryState) -> Optional[TruncationVerdict]: + """Content-filter stream stall → fallback. ``_content_filter_terminated`` is + content-deterministic, so escalate before retrying the primary; without a fallback + fall through to normal continuation (best-effort, may loop).""" + agent = st.agent + if not ( + getattr(st.response, "_content_filter_terminated", False) + and agent._fallback_index < len(agent._fallback_chain) + ): + return None + agent._vprint( + f"{agent.log_prefix}🛡️ Content filter terminated stream — activating fallback provider...", + force=True, + ) + agent._emit_status("Content filter terminated stream; switching to fallback...") + if agent._try_activate_fallback(): + # Roll partial content back to the last clean turn so the fallback gets a + # coherent continuation point; unmark survivors (their text left the partial). + if st.truncated_response_parts: + st.messages = agent._get_messages_up_to_last_assistant(st.messages) + for _frag in st.messages: + if isinstance(_frag, dict): + _frag.pop("_length_continuation_fragment", None) + _frag.pop("_length_continuation_nudge", None) + agent._session_messages = st.messages + st.length_continue_retries = 0 + st.truncated_response_parts = [] + st.retry_count = 0 + st.compression_attempts = 0 + _retry.primary_recovery_attempted = False + _retry.restart_with_rebuilt_messages = True + return st.done("break") + agent._vprint( + f"{agent.log_prefix}⚠️ No fallback provider configured — retrying with same provider " + f"(may re-hit filter)...", + force=True, + ) + return None + + +def _continue_text(st: _Trunc, _retry: TurnRetryState, assistant_message: Any) -> TruncationVerdict: + """Text truncation (no tool calls): append the fragment + a continuation nudge (up to + 4), then the ceiling exit that drops the fragment trail and keeps the stitched partial. + Never appends an interim assistant row with NO visible content — strict providers + reject it with 400 — only the nudge.""" + from agent.conversation_loop import _get_continuation_prompt, _join_truncated_parts + + agent = st.agent + messages = st.messages + st.length_continue_retries += 1 + n = st.length_continue_retries + _interim_content = getattr(assistant_message, "content", None) + if not _interim_content and not st.is_stub: + # Thinking-only truncation: continuing with thinking ON re-burns the budget. + agent._ephemeral_reasoning_off = True + if _interim_content: + interim_msg = agent._build_assistant_message(assistant_message, st.finish_reason) + interim_msg["_length_continuation_fragment"] = True # ceiling exit drops these + append_message(messages, interim_msg) + st.truncated_response_parts.append(_interim_content) + + if n < 4: + _dropped_tools = getattr(st.response, "_dropped_tool_names", None) + if st.is_stub and _dropped_tools: + agent._vprint( + f"{agent.log_prefix}↻ Stream interrupted mid " + f"tool-call ({', '.join(_dropped_tools[:3])}) — requesting chunked retry ({n}/4)..." + ) + elif st.is_stub: + agent._vprint(f"{agent.log_prefix}↻ Stream interrupted — requesting continuation ({n}/4)...") + else: + agent._vprint(f"{agent.log_prefix}↻ Requesting continuation ({n}/4)...") + append_message(messages, { + "role": "user", "content": _get_continuation_prompt(st.is_stub, _dropped_tools), + "_length_continuation_nudge": True, + }) + agent._session_messages = messages + _retry.restart_with_length_continuation = True + return st.done("break") + + partial_response = agent._strip_think_blocks(_join_truncated_parts(st.truncated_response_parts)).strip() + # The one-shot reasoning-off override must not leak into the next turn. + agent._ephemeral_reasoning_off = False + agent._vprint( + f"{agent.log_prefix}⚠️ Response still truncated after {n} continuation attempts — " + + ("keeping the partial response received so far." if partial_response + else "no visible text was produced."), + force=True, + ) + # Unanswered continue nudges made every later turn re-truncate: drop the trail. + idx = st.current_turn_user_idx + _turn_start = idx + 1 if isinstance(idx, int) and idx >= 0 else 0 + messages[_turn_start:] = [ + m for m in messages[_turn_start:] + if not (isinstance(m, dict) and ( + m.get("_length_continuation_fragment") or m.get("_length_continuation_nudge") + )) + ] + if partial_response: + append_message(messages, { + "role": "assistant", "content": partial_response, "finish_reason": "length" + }) + agent._session_messages = messages + return st.end_turn( + partial_response or _CEILING_NO_TEXT, + "Response remained truncated after 4 continuation attempts", + ) + + +def _retry_truncated_tool_call(st: _Trunc, api_kwargs: Any) -> TruncationVerdict: + """Truncated tool call: re-run the same call (up to 4×) with a boosted max_tokens — + a real output-cap truncation needs it, harmless for a network stall — else refuse to + execute incomplete arguments.""" + agent = st.agent + if st.truncated_tool_call_retries < 4: + st.truncated_tool_call_retries += 1 + n = st.truncated_tool_call_retries + if st.is_stub: + agent._buffer_vprint(f"⚠️ Stream interrupted mid tool-call — retrying ({n}/4)...") + else: + agent._buffer_vprint(f"⚠️ Truncated tool call detected — retrying API call ({n}/4)...") + _tc_boost = (agent.max_tokens if agent.max_tokens else 4096) * (2 ** n) + _tc_requested_cap = agent._requested_output_cap_from_api_kwargs(api_kwargs) + if _tc_requested_cap is not None: + _tc_boost = max(_tc_boost, _tc_requested_cap) + agent._ephemeral_max_output_tokens = min(_tc_boost, max(32768, _tc_requested_cap or 0)) + return st.done("continue") # don't append the broken response + agent._flush_status_buffer() + if st.is_stub: + agent._vprint( + f"{agent.log_prefix}⚠️ Stream kept dropping mid tool-call after 4 retries — the action was not executed.", + force=True, + ) + _final_response = "Stream repeatedly dropped mid tool-call (network); the tool was not executed" + else: + agent._vprint( + f"{agent.log_prefix}⚠️ Truncated tool call response detected again — refusing to execute incomplete tool arguments.", + force=True, + ) + _final_response = _TRUNCATED_FINAL + agent._cleanup_task_resources(st.effective_task_id) + # Prior tool batches can leave a tool-result tail; this path never reaches finalize_turn. + close_interrupted_tool_sequence(st.messages, _final_response) + return st.end_turn(_final_response, cleanup=False) + + def recover_from_truncation( agent: Any, response: Any, finish_reason: str, _retry: TurnRetryState, *, messages: List[Dict[str, Any]], conversation_history: Any, api_kwargs: Any, api_call_count: int, @@ -55,408 +309,57 @@ def recover_from_truncation( """Recover from a truncated response. Order is load-bearing: thinking exhaustion and repetition abort BEFORE any continuation; a content-filter stall escalates to the fallback chain BEFORE the primary is retried; text continuation (no tool calls) then - truncated tool-call retry; finally roll back to the last complete assistant turn. - Never appends an interim assistant row with NO visible content (strict providers - reject it with 400) — only the continuation nudge.""" - from agent.conversation_loop import _get_continuation_prompt, _join_truncated_parts - - def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> TruncationVerdict: - return TruncationVerdict( - action=action, result=result, messages=messages, - length_continue_retries=length_continue_retries, - truncated_response_parts=truncated_response_parts, - truncated_tool_call_retries=truncated_tool_call_retries, retry_count=retry_count, - compression_attempts=compression_attempts, - ) - - if getattr(response, "id", "") == PARTIAL_STREAM_STUB_ID: - agent._vprint( - f"{agent.log_prefix}⚠️ Response truncated — stream " - f"ended before completion", - force=True, - ) - else: - agent._vprint( - f"{agent.log_prefix}⚠️ Response truncated " - f"(finish_reason='length') - model hit max output tokens", - force=True, - ) - - # Normalize to one OpenAI-style message so continuation and tool- - # call retry work across transports (Anthropic reuses the loop's - # adapter). - _trunc_msg = None - _trunc_transport = agent._get_transport() - if agent.api_mode == "anthropic_messages": - _trunc_result = _trunc_transport.normalize_response( - response, strip_tool_prefix=agent._is_anthropic_oauth - ) - else: - _trunc_result = _trunc_transport.normalize_response(response) - _trunc_msg = _trunc_result + truncated tool-call retry; finally roll back to the last complete assistant turn.""" + st = _Trunc( + agent=agent, response=response, finish_reason=finish_reason, + conversation_history=conversation_history, api_call_count=api_call_count, + effective_task_id=effective_task_id, current_turn_user_idx=current_turn_user_idx, + messages=messages, length_continue_retries=length_continue_retries, + truncated_response_parts=truncated_response_parts, + truncated_tool_call_retries=truncated_tool_call_retries, retry_count=retry_count, + compression_attempts=compression_attempts, + ) + agent._vprint( + f"{agent.log_prefix}⚠️ Response truncated — stream ended before completion" + if st.is_stub else + f"{agent.log_prefix}⚠️ Response truncated (finish_reason='length') - model hit max output tokens", + force=True, + ) + _trunc_msg = normalize_response_for_agent(agent, response) _trunc_content = getattr(_trunc_msg, "content", None) if _trunc_msg else None _trunc_has_tool_calls = bool(getattr(_trunc_msg, "tool_calls", None)) if _trunc_msg else False - # ── Detect thinking-budget exhaustion ────────────── - # Only when reasoning blocks exist with no visible text after them; - # content=None from non- models is normal truncation. - _has_think_tags = bool( - _trunc_content and re.search( - r'<(?:think|thinking|reasoning|REASONING_SCRATCHPAD)[^>]*>', - _trunc_content, - re.IGNORECASE, - ) - ) - _thinking_exhausted = ( - not _trunc_has_tool_calls - and _has_think_tags - and ( - (_trunc_content is not None and not agent._has_content_after_think_block(_trunc_content)) - or _trunc_content is None - ) - ) + abort = _abort_reason(agent, _trunc_content, _trunc_has_tool_calls) + if abort is not None: + line, user_response, error = abort + agent._vprint(f"{agent.log_prefix}{line}", force=True) + return st.end_turn(user_response, error) - if _thinking_exhausted: - _exhaust_error = ( - "Model used all output tokens on reasoning with none left " - "for the response. Try lowering reasoning effort or " - "increasing max_tokens." - ) - agent._vprint( - f"{agent.log_prefix}💭 Reasoning exhausted the output token budget — " - f"no visible response was produced.", - force=True, - ) - # Return a user-friendly message as the response so CLI and - # gateway display it. - _exhaust_response = ( - "⚠️ **Thinking Budget Exhausted**\n\n" - "The model used all its output tokens on reasoning " - "and had none left for the actual response.\n\n" - "To fix this:\n" - "→ Lower reasoning effort: `/reasoning low` or `/reasoning minimal`\n" - "→ Or switch to a larger/non-reasoning model with `/model`" - ) - agent._cleanup_task_resources(effective_task_id) - agent._persist_session(messages, conversation_history) - return _verdict("return", { - "final_response": _exhaust_response, - "messages": messages, - "api_calls": api_call_count, - "completed": False, - "partial": True, - "error": _exhaust_error, - }) + if agent.api_mode in _CONTINUABLE_MODES: + cf = _content_filter_fallback(st, _retry) + if cf is not None: + return cf + if _trunc_msg is not None: + if not _trunc_has_tool_calls: + return _continue_text(st, _retry, _trunc_msg) + return _retry_truncated_tool_call(st, api_kwargs) - # ── Detect repetition-dominated truncation (#86581) ── - # A repetition loop can burn the whole budget on one fragment; abort - # like _thinking_exhausted (reasoning stripped first). - _visible_trunc = ( - agent._strip_think_blocks(_trunc_content) - if isinstance(_trunc_content, str) - else _trunc_content - ) - _repetition_dominated = ( - not _trunc_has_tool_calls - and bool(_visible_trunc) - and is_repetition_dominated(_visible_trunc) - ) - if _repetition_dominated: - _rep_error = ( - "Model output entered a repetition loop and was " - "truncated mid-loop; refusing to continue a " - "degenerate response." - ) - agent._vprint( - f"{agent.log_prefix}🔁 Response dominated by " - f"repeated text — stopping instead of " - f"continuing a degenerate response.", - force=True, - ) - _rep_response = ( - "⚠️ **Response Stopped — Repetition Detected**\n\n" - "The model fell into a repetition loop while " - "writing this response, so continuing would only " - "produce more repeated text. The partial response " - "was discarded.\n\n" - "→ Switch to a different model with `/model`\n" - "→ Or resend your message (your conversation " - "history is preserved)" - ) - agent._cleanup_task_resources(effective_task_id) - agent._persist_session(messages, conversation_history) - return _verdict("return", { - "final_response": _rep_response, - "messages": messages, - "api_calls": api_call_count, - "completed": False, - "partial": True, - "error": _rep_error, - }) - - if agent.api_mode in {"chat_completions", "bedrock_converse", "anthropic_messages"}: - assistant_message = _trunc_msg - # ── Content-filter stream stall → fallback (#32421) ── - # ``_content_filter_terminated`` is content-deterministic; - # escalate to the fallback before retrying the primary. - _cf_terminated = getattr( - response, "_content_filter_terminated", False - ) - if ( - _cf_terminated and agent._fallback_index < len(agent._fallback_chain) - ): - agent._vprint( - f"{agent.log_prefix}🛡️ Content filter terminated " - f"stream — activating fallback provider...", - force=True, - ) - agent._emit_status( - "Content filter terminated stream; switching to fallback..." - ) - if agent._try_activate_fallback(): - # Roll partial content back to the last clean turn so - # the fallback gets a coherent continuation point. - if truncated_response_parts: - messages = agent._get_messages_up_to_last_assistant(messages) - # Unmark survivors: their text left the stitched partial. - for _frag in messages: - if isinstance(_frag, dict): - _frag.pop("_length_continuation_fragment", None) - _frag.pop("_length_continuation_nudge", None) - agent._session_messages = messages - length_continue_retries = 0 - truncated_response_parts = [] - retry_count = 0 - compression_attempts = 0 - _retry.primary_recovery_attempted = False - _retry.restart_with_rebuilt_messages = True - return _verdict("break") - # No fallback available — fall through to normal - # continuation (best-effort, may loop). - agent._vprint( - f"{agent.log_prefix}⚠️ No fallback provider " - f"configured — retrying with same provider " - f"(may re-hit filter)...", - force=True, - ) - if assistant_message is not None and not _trunc_has_tool_calls: - length_continue_retries += 1 - # Never append an interim assistant message with NO visible - # content: strict providers reject it (HTTP 400), poisoning - # history. Append only the nudge. - _interim_content = getattr(assistant_message, "content", None) - _is_empty_partial_stub = ( - getattr(response, "id", "") == PARTIAL_STREAM_STUB_ID and not _interim_content - ) - if not _interim_content and not _is_empty_partial_stub: - # Thinking-only truncation: continuing with thinking ON - # re-burns the budget, so drop thinking for one request. - agent._ephemeral_reasoning_off = True - if _interim_content: - interim_msg = agent._build_assistant_message(assistant_message, finish_reason) - # Marked so the ceiling exit can drop the fragment trail. - interim_msg["_length_continuation_fragment"] = True - append_message(messages, interim_msg) - truncated_response_parts.append(_interim_content) - - if length_continue_retries < 4: - _is_partial_stream_stub = ( - getattr(response, "id", "") == PARTIAL_STREAM_STUB_ID - ) - _dropped_tools = getattr( - response, "_dropped_tool_names", None - ) - - if _is_partial_stream_stub and _dropped_tools: - _tool_list = ", ".join(_dropped_tools[:3]) - agent._vprint( - f"{agent.log_prefix}↻ Stream interrupted mid " - f"tool-call ({_tool_list}) — requesting " - f"chunked retry " - f"({length_continue_retries}/4)..." - ) - elif _is_partial_stream_stub: - agent._vprint( - f"{agent.log_prefix}↻ Stream interrupted — " - f"requesting continuation " - f"({length_continue_retries}/4)..." - ) - else: - agent._vprint( - f"{agent.log_prefix}↻ Requesting continuation " - f"({length_continue_retries}/4)..." - ) - - _continue_content = _get_continuation_prompt( - _is_partial_stream_stub, _dropped_tools - ) - continue_msg = { - "role": "user", "content": _continue_content, "_length_continuation_nudge": True - } - append_message(messages, continue_msg) - agent._session_messages = messages - _retry.restart_with_length_continuation = True - return _verdict("break") - - partial_response = agent._strip_think_blocks(_join_truncated_parts(truncated_response_parts)).strip() - # The one-shot reasoning-off override must not leak into the - # next turn when the ceiling exit skips the consuming call. - agent._ephemeral_reasoning_off = False - if partial_response: - agent._vprint( - f"{agent.log_prefix}⚠️ Response still truncated " - f"after {length_continue_retries} continuation attempts — keeping the " - f"partial response received so far.", - force=True, - ) - _ceiling_final = partial_response - else: - # Every fragment was empty (e.g. reasoning-only model): - # return an actionable message, not a bare None. - agent._vprint( - f"{agent.log_prefix}⚠️ Response still truncated " - f"after {length_continue_retries} continuation attempts — no visible " - f"text was produced.", - force=True, - ) - _ceiling_final = ( - "⚠️ **No visible answer was produced.** The " - "model hit its output-token limit on every " - "continuation attempt — its reasoning " - "consumed the entire budget each time.\n\n" - "To fix this:\n" - "→ Lower reasoning effort: `/reasoning low` " - "or `/reasoning none`\n" - "→ Or raise max_tokens for this model" - ) - # Unanswered continue nudges made every later turn re-truncate. - _turn_start = ( - current_turn_user_idx + 1 - if isinstance(current_turn_user_idx, int) - and current_turn_user_idx >= 0 - else 0 - ) - messages[_turn_start:] = [ - m for m in messages[_turn_start:] - if not ( - isinstance(m, dict) - and ( - m.get("_length_continuation_fragment") - or m.get("_length_continuation_nudge") - ) - ) - ] - if partial_response: - append_message(messages, { - "role": "assistant", "content": partial_response, "finish_reason": "length" - }) - agent._session_messages = messages - agent._cleanup_task_resources(effective_task_id) - agent._persist_session(messages, conversation_history) - return _verdict("return", { - "final_response": _ceiling_final, - "messages": messages, - "api_calls": api_call_count, - "completed": False, - "partial": True, - "error": "Response remained truncated after 4 continuation attempts", - }) - - if agent.api_mode in {"chat_completions", "bedrock_converse", "anthropic_messages"}: - assistant_message = _trunc_msg - if assistant_message is not None and _trunc_has_tool_calls: - _is_stub_stall = ( - getattr(response, "id", "") == PARTIAL_STREAM_STUB_ID - ) - if truncated_tool_call_retries < 4: - truncated_tool_call_retries += 1 - if _is_stub_stall: - # Stream broke mid tool-call (network), not a real - # output cap — say so. - agent._buffer_vprint( - f"⚠️ Stream interrupted mid tool-call — " - f"retrying ({truncated_tool_call_retries}/4)..." - ) - else: - agent._buffer_vprint( - f"⚠️ Truncated tool call detected — " - f"retrying API call " - f"({truncated_tool_call_retries}/4)..." - ) - # Boost max_tokens per retry: a real output-cap - # truncation needs it; harmless for a stall. - _tc_boost_base = agent.max_tokens if agent.max_tokens else 4096 - _tc_boost = _tc_boost_base * (2 ** truncated_tool_call_retries) - _tc_requested_cap = agent._requested_output_cap_from_api_kwargs(api_kwargs) - if _tc_requested_cap is not None: - _tc_boost = max(_tc_boost, _tc_requested_cap) - _tc_boost_cap = max(32768, _tc_requested_cap or 0) - agent._ephemeral_max_output_tokens = min(_tc_boost, _tc_boost_cap) - # Don't append the broken response; re-run the same call - # from current state. - return _verdict("continue") - agent._flush_status_buffer() - if _is_stub_stall: - agent._vprint( - f"{agent.log_prefix}⚠️ Stream kept dropping mid tool-call after 4 retries — the action was not executed.", - force=True, - ) - else: - agent._vprint( - f"{agent.log_prefix}⚠️ Truncated tool call response detected again — refusing to execute incomplete tool arguments.", - force=True, - ) - agent._cleanup_task_resources(effective_task_id) - _final_response = ( - "Stream repeatedly dropped mid tool-call (network); " - "the tool was not executed" - if _is_stub_stall - else "Response truncated due to output length limit" - ) - # Prior tool batches can leave a tool-result tail; this path - # never reaches finalize_turn (#48879). - close_interrupted_tool_sequence(messages, _final_response) - agent._persist_session(messages, conversation_history) - return _verdict("return", { - "final_response": _final_response, - "messages": messages, - "api_calls": api_call_count, - "completed": False, - "partial": True, - "error": _final_response, - }) - - # If we have prior messages, roll back to last complete state if len(messages) > 1: agent._vprint(f"{agent.log_prefix} ⏪ Rolling back to last complete assistant turn") - rolled_back_messages = agent._get_messages_up_to_last_assistant(messages) + return st.end_turn( + _TRUNCATED_FINAL, result_messages=agent._get_messages_up_to_last_assistant(messages) + ) + # First message was truncated - mark as failed + agent._flush_status_buffer() + agent._vprint(f"{agent.log_prefix}❌ First response truncated - cannot recover", force=True) + return st.end_turn(_FIRST_TRUNCATED_FINAL, cleanup=False, failed=True) - agent._cleanup_task_resources(effective_task_id) - agent._persist_session(messages, conversation_history) - return _verdict("return", { - "final_response": "Response truncated due to output length limit", - "messages": rolled_back_messages, - "api_calls": api_call_count, - "completed": False, - "partial": True, - "error": "Response truncated due to output length limit" - }) - else: - # First message was truncated - mark as failed - agent._flush_status_buffer() - agent._vprint(f"{agent.log_prefix}❌ First response truncated - cannot recover", force=True) - agent._persist_session(messages, conversation_history) - return _verdict("return", { - "final_response": "First response truncated due to output length limit", - "messages": messages, - "api_calls": api_call_count, - "completed": False, - "failed": True, - "error": "First response truncated due to output length limit" - }) - return _verdict("fallthrough") +_CODEX_REPLAY_KEYS = ( + "content", "reasoning", "reasoning_content", "reasoning_details", + "codex_reasoning_items", "codex_message_items", +) def continue_codex_incomplete( @@ -466,128 +369,84 @@ def continue_codex_incomplete( """Codex Responses ``status=incomplete`` continuation (max 3 per turn). Appends the interim assistant message (deduped on visible content only — opaque - provider state drifts per continuation, #52711; ``codex_reasoning_items`` are merged, - not overwritten, because the earlier response holds the only native-compaction + provider state drifts per continuation; ``codex_reasoning_items`` are merged, not + overwritten, because the earlier response holds the only native-compaction checkpoint) and, when a bare retry would be byte-identical, a user-role nudge — only after an assistant row, to preserve role alternation. Returns ``None`` to continue the turn loop, or the terminal ``partial`` result once retries are exhausted.""" from agent.conversation_loop import _CODEX_INCOMPLETE_NUDGE agent._codex_incomplete_retries += 1 + n = agent._codex_incomplete_retries interim_msg = agent._build_assistant_message(assistant_message, finish_reason) interim_has_content = bool((interim_msg.get("content") or "").strip()) - interim_has_reasoning = bool(interim_msg.get("reasoning", "").strip()) if isinstance(interim_msg.get("reasoning"), str) else False + _reasoning = interim_msg.get("reasoning") + interim_has_reasoning = isinstance(_reasoning, str) and bool(_reasoning.strip()) interim_has_codex_reasoning = bool(interim_msg.get("codex_reasoning_items")) interim_has_codex_message_items = bool(interim_msg.get("codex_message_items")) - if ( - interim_has_content - or interim_has_reasoning - or interim_has_codex_reasoning - or interim_has_codex_message_items - ): + if interim_has_content or interim_has_reasoning or interim_has_codex_reasoning or interim_has_codex_message_items: last_msg = messages[-1] if messages else None - # Dedup on visible content only (content + reasoning): opaque - # provider state drifts per continuation and would defeat dedup - # (#52711). - last_interim_visible = ( - agent._interim_assistant_visible_text(last_msg) if isinstance(last_msg, dict) else "" - ) + last_is_dict = isinstance(last_msg, dict) + last_interim_visible = agent._interim_assistant_visible_text(last_msg) if last_is_dict else "" current_interim_visible = agent._interim_assistant_visible_text(interim_msg) if last_interim_visible or current_interim_visible: same_visible_output = last_interim_visible == current_interim_visible else: - # Preserve the existing reasoning-only behavior when - # neither response has text eligible for interim delivery. - same_visible_output = ( + # Neither has text eligible for interim delivery: compare raw content+reasoning. + same_visible_output = last_is_dict and ( (last_msg.get("content") or "") == (interim_msg.get("content") or "") and (last_msg.get("reasoning") or "") == (interim_msg.get("reasoning") or "") - ) if isinstance(last_msg, dict) else False - visible_duplicate = ( - isinstance(last_msg, dict) + ) + if ( + last_is_dict and last_msg.get("role") == "assistant" and last_msg.get("finish_reason") == "incomplete" and same_visible_output - ) - if visible_duplicate: - # Update replay state in-place: keep the latest provider payload - # without re-emitting identical user-visible commentary. - for _key in ( - "content", - "reasoning", - "reasoning_content", - "reasoning_details", - "codex_reasoning_items", - "codex_message_items", - ): - if _key in interim_msg: - if _key == "codex_reasoning_items": - # Merge, don't overwrite: the earlier response's - # native compaction checkpoint is the only copy. See - # merge_interim_reasoning_items. - from agent.native_compaction import ( - merge_interim_reasoning_items, - ) - last_msg[_key] = merge_interim_reasoning_items( - last_msg.get(_key), interim_msg[_key] - ) - else: - last_msg[_key] = interim_msg[_key] + ): + # Duplicate: refresh replay state in place, no re-emitted commentary. + for _key in _CODEX_REPLAY_KEYS: + if _key not in interim_msg: + continue + if _key == "codex_reasoning_items": + from agent.native_compaction import merge_interim_reasoning_items + last_msg[_key] = merge_interim_reasoning_items(last_msg.get(_key), interim_msg[_key]) + else: + last_msg[_key] = interim_msg[_key] else: append_message(messages, interim_msg) agent._emit_interim_assistant_message(interim_msg) - if agent._codex_incomplete_retries < 3: - # If the interim has nothing the Responses converter will replay, a - # bare retry is byte-identical and fails identically; append a - # user-role nudge so the retry differs and asks for the answer. - interim_replayable = ( - interim_has_content or interim_has_codex_reasoning or interim_has_codex_message_items - ) - # Replayable ≠ different: an interim holding only a ``compaction`` - # checkpoint in ``codex_reasoning_items`` is replayable yet re-sends - # identically. One bare retry, then always nudge. - if not interim_replayable or agent._codex_incomplete_retries >= 2: + if n < 3: + # If the interim has nothing the Responses converter will replay, a bare retry is + # byte-identical; a replayable interim holding only a ``compaction`` checkpoint + # ALSO re-sends identically. One bare retry, then always nudge. + interim_replayable = interim_has_content or interim_has_codex_reasoning or interim_has_codex_message_items + if not interim_replayable or n >= 2: _last_msg = messages[-1] if messages else None - _already_nudged = ( - isinstance(_last_msg, dict) - and _last_msg.get("role") == "user" - and _last_msg.get("content") == _CODEX_INCOMPLETE_NUDGE - ) - # Alternation guard: the user-role nudge may only follow an - # assistant message; after a too-empty interim it would create - # user→user / tool→user. - _last_is_assistant = ( - isinstance(_last_msg, dict) and _last_msg.get("role") == "assistant" - ) - if not _already_nudged and _last_is_assistant: - append_message(messages, { - "role": "user", "content": _CODEX_INCOMPLETE_NUDGE - }) + if isinstance(_last_msg, dict): + _already_nudged = ( + _last_msg.get("role") == "user" and _last_msg.get("content") == _CODEX_INCOMPLETE_NUDGE + ) + # Alternation guard: the nudge may only follow an assistant row. + if not _already_nudged and _last_msg.get("role") == "assistant": + append_message(messages, {"role": "user", "content": _CODEX_INCOMPLETE_NUDGE}) if not agent.quiet_mode: - agent._vprint(f"{agent.log_prefix}↻ Codex response incomplete; continuing turn ({agent._codex_incomplete_retries}/3)") - # Show the continuation on the spinner/status line and gateway - # heartbeat; these retries can take minutes and otherwise look like - # infinite thinking (#64434). + agent._vprint(f"{agent.log_prefix}↻ Codex response incomplete; continuing turn ({n}/3)") + # Spinner/heartbeat notice: these retries can take minutes and otherwise look + # like infinite thinking. agent._emit_wait_notice( - f"↻ model returned reasoning with no final answer — " - f"asking it to continue " - f"({agent._codex_incomplete_retries}/3)" + f"↻ model returned reasoning with no final answer — asking it to continue ({n}/3)" ) agent._session_messages = messages return None agent._codex_incomplete_retries = 0 agent._persist_session(messages, conversation_history) - return { - "final_response": "Codex response remained incomplete after 3 continuation attempts", - "messages": messages, - "api_calls": api_call_count, - "completed": False, - "partial": True, - "error": "Codex response remained incomplete after 3 continuation attempts", - } + return partial_result( + messages, api_call_count, "Codex response remained incomplete after 3 continuation attempts" + ) @dataclass @@ -610,25 +469,13 @@ def handle_content_policy_refusal( ) -> RefusalVerdict: """HTTP-200 refusal (``finish_reason`` ``content_filter`` / ``guardrail_intervened``). Deterministic for the unchanged prompt — never retried: one configured-fallback try, - else surface the refusal (explanation may live only in the reasoning channel). The - caller stops its spinner reference; this stops the spinner object.""" + else surface the refusal (explanation may live only in the reasoning channel).""" from agent.conversation_loop import ( _CONTENT_POLICY_RECOVERY_HINT, _arm_fallback_restart, _content_policy_blocked_result ) - def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> RefusalVerdict: - return RefusalVerdict(action=action, result=result, active_system_prompt=active_system_prompt) - - _refusal_transport = agent._get_transport() - if agent.api_mode == "anthropic_messages": - _refusal_result = _refusal_transport.normalize_response( - response, strip_tool_prefix=agent._is_anthropic_oauth - ) - else: - _refusal_result = _refusal_transport.normalize_response(response) + _refusal_result = normalize_response_for_agent(agent, response) _refusal_text = (getattr(_refusal_result, "content", None) or "").strip() - # Some refusals carry the explanation only in the reasoning - # channel; fall back to it so the user sees *something*. if not _refusal_text: _refusal_text = (agent._extract_reasoning(_refusal_result) or "").strip() @@ -640,52 +487,33 @@ def handle_content_policy_refusal( status_code=None, retry_count=retry_count, max_retries=max_retries, retryable=False, reason=FailoverReason.content_policy_blocked.value, ) + stop_thinking_spinner(agent, thinking_spinner) - if thinking_spinner: - thinking_spinner.stop("") - if agent.thinking_callback: - agent.thinking_callback("") - - # Deterministic for the unchanged prompt — never retry. Try a - # configured fallback once; otherwise surface the refusal. if agent._has_pending_fallback(): - agent._buffer_status( - "⚠️ Model declined to respond (safety refusal) — trying fallback..." - ) + agent._buffer_status("⚠️ Model declined to respond (safety refusal) — trying fallback...") if agent._try_activate_fallback(): - active_system_prompt = _arm_fallback_restart( - agent, api_messages, active_system_prompt, _retry) - return _verdict("break") + active_system_prompt = _arm_fallback_restart(agent, api_messages, active_system_prompt, _retry) + return RefusalVerdict("break", None, active_system_prompt) agent._flush_status_buffer() - _refusal_log = ( - _refusal_text[:500] + "..." if len(_refusal_text) > 500 else _refusal_text - ) + _refusal_log = _refusal_text[:500] + "..." if len(_refusal_text) > 500 else _refusal_text logger.warning( - "%sModel declined to respond (finish_reason=content_filter). " - "model=%s provider=%s refusal=%s", + "%sModel declined to respond (finish_reason=content_filter). model=%s provider=%s refusal=%s", agent.log_prefix, agent.model, agent.provider, _refusal_log or "(no text)", ) - agent._emit_status( - "⚠️ The model declined to respond to this request (safety refusal)." - ) - + agent._emit_status("⚠️ The model declined to respond to this request (safety refusal).") _refusal_detail = ( - f"Model's explanation: {_refusal_text}" - if _refusal_text - else "The model returned no explanation." + f"Model's explanation: {_refusal_text}" if _refusal_text else "The model returned no explanation." ) _refusal_response = ( - "⚠️ The model declined to respond to this request " - "(safety refusal — not a Hermes/gateway failure).\n\n" + "⚠️ The model declined to respond to this request (safety refusal — not a Hermes/gateway failure).\n\n" f"{_refusal_detail}\n\n" f"{_CONTENT_POLICY_RECOVERY_HINT}" ) - agent._cleanup_task_resources(effective_task_id) agent._persist_session(messages, conversation_history) - return _verdict("return", _content_policy_blocked_result( + return RefusalVerdict("return", _content_policy_blocked_result( messages, api_call_count, final_response=_refusal_response, error_detail=_refusal_text or "model declined (content_filter)", - )) + ), active_system_prompt)