ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 11 Packages Projects Releases Wiki Activity

Merge branch 'simp/r3-08-g4' into simp/r3-08

Browse Source
This commit is contained in:
Teknium
2026-09-02 19:29:30 -07:00
parent 08d95abdd3 b741b8de79
commit 0c3dc9c728
8 changed files with 992 additions and 1365 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/turn_api_call.py
+61 -104
View File
@@ -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")
agent/turn_api_request.py
+88 -93
View File
@@ -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,
)
agent/turn_empty_response.py
+151 -212
View File
@@ -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
``<think>`` 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 ``<think>`` 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'<think>|<thinking>|<reasoning>', 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 <think> in content, not reasoning_content, so
# _has_structured misses it; detect here to route to prefill.
_has_inline_thinking = bool(
re.search( r'<think>|<thinking>|<reasoning>', 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 <think> 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")
agent/turn_overflow.py
+84 -98
View File
@@ -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")
agent/turn_request_assembly.py
+90 -104
View File
@@ -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,
)
agent/turn_response_check.py
+61 -102
View File
@@ -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")
agent/turn_response_intake.py
+100 -123
View File
@@ -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'</?(?:REASONING_SCRATCHPAD|think|reasoning)>')
@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'</?(?:REASONING_SCRATCHPAD|think|reasoning)>', '', _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 <REASONING_SCRATCHPAD> (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 <REASONING_SCRATCHPAD> (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 <REASONING_SCRATCHPAD> 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")
agent/turn_truncation.py
+357 -529
View File
@@ -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-<think> 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-<think> 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)