From 7fa7b58b3cc7cd70b41e6b4545bc981a22ca0608 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 18:53:53 -0700 Subject: [PATCH] refactor(agent): compact conversation_loop docstrings/comments, unify partial-turn result shape, tidy empty_response_guard/interrupt_compat/iteration_budget --- agent/conversation_loop.py | 453 +++++++++++++--------------------- agent/empty_response_guard.py | 53 ++-- agent/interrupt_compat.py | 29 +-- agent/iteration_budget.py | 14 +- 4 files changed, 205 insertions(+), 344 deletions(-) diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index 48b41bc949..8521e1699a 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -83,13 +83,11 @@ from utils import base_url_host_matches logger = logging.getLogger(__name__) -# Scaffold marker used by _apply_active_turn_redirect and the ghost-row filter -# in the api_messages loop. Module-level so both sites can never drift. +# Shared by _apply_active_turn_redirect and the api_messages ghost-row filter so both sites cannot drift. _INTERRUPT_SCAFFOLD_MARKER = "[This response was interrupted by a user correction.]" -# One-time wrap-up notice appended when a wall-clock run budget crosses 80% -# (agent.run_budget_seconds / --run-budget): stop new work, deliver current state. +# One-time wrap-up notice appended when a wall-clock run budget (--run-budget) crosses 80%. RUN_BUDGET_WRAPUP_NOTICE = ( "[SYSTEM NOTICE — run time budget nearly exhausted] " "Run time budget nearly exhausted. Stop new discovery/verification work " @@ -104,11 +102,10 @@ def _midturn_request_pressure_tokens( effective_system: str, approx_tokens: int, ) -> int: - """Token figure the mid-turn pre-API compression guard compares. - - Returns the pruned native-Responses estimate when native compaction eligibility is - proven (the generic estimate overstates the wire on compacted sessions, #96995), - else the generic message+tools figure. System prompt is counted exactly once.""" + """Token figure the mid-turn pre-API compression guard compares: the pruned + native-Responses estimate when native compaction eligibility is proven (the generic + estimate overstates the wire on compacted sessions, #96995), else messages+tools. + The system prompt is counted exactly once.""" try: from agent.codex_responses_adapter import ( estimate_native_responses_preflight_tokens, @@ -149,8 +146,7 @@ def _maybe_inject_run_budget_wrapup(agent: Any, messages: List[Dict[str, Any]]) """Inject the one-time wall-clock wrap-up notice when past 80% of budget. Appends to the NEWEST ``role:"tool"`` message (cache-safe, like /steer); latches - ``_run_budget_wrapup_injected`` only on a successful append. Returns True when - injected. Dormant unless ``run_budget_seconds`` + ``_run_budget_started_at`` set.""" + ``_run_budget_wrapup_injected`` only on a successful append.""" budget = getattr(agent, "run_budget_seconds", None) started = getattr(agent, "_run_budget_started_at", None) if ( @@ -166,19 +162,15 @@ def _maybe_inject_run_budget_wrapup(agent: Any, messages: List[Dict[str, Any]]) existing = msg.get("content", "") if isinstance(existing, str): msg["content"] = existing + f"\n\n{RUN_BUDGET_WRAPUP_NOTICE}" - else: - # Multimodal content blocks — append a text block. + else: # multimodal content blocks — append a text block try: - blocks = list(existing) if existing else [] - blocks.append({"type": "text", "text": RUN_BUDGET_WRAPUP_NOTICE}) - msg["content"] = blocks + msg["content"] = [*(existing or []), {"type": "text", "text": RUN_BUDGET_WRAPUP_NOTICE}] except Exception: return False agent._run_budget_wrapup_injected = True logger.info( "Run budget wrap-up notice injected (budget=%.0fs, elapsed=%.0fs)", - float(budget), - time.time() - started, + float(budget), time.time() - started, ) return True return False @@ -187,10 +179,8 @@ def _maybe_inject_run_budget_wrapup(agent: Any, messages: List[Dict[str, Any]]) def _restore_user_after_reference_handoff( messages: List[Dict[str, Any]], user_message: Any ) -> bool: - """Re-append this turn's real user ask when compaction left only a handoff. - - Returns True when a restore append happened; only decides whether a restorable - ask exists (#80622).""" + """Re-append this turn's real user ask when compaction left only a handoff (#80622). + Returns True when a restore append happened.""" if isinstance(user_message, str): restorable = bool(user_message.strip()) else: @@ -217,15 +207,14 @@ def _should_skip_model_call_for_reference_handoff( ) -# Fallback final_response for the sole-handoff skip (#80622). Not a replay of the -# last assistant text: finalize_turn appends final_response as a fresh assistant row. +# Fallback final_response for the sole-handoff skip (#80622); finalize_turn appends it as a +# fresh assistant row, so it must not replay the last assistant text. _HANDOFF_SKIP_FINAL_RESPONSE = ( "Context was compacted. The previous response is complete — " "awaiting your next message." ) -# Terminal final_response when compression hit its host timeout while the request -# was still oversized; resending would only bounce off the overflow error (#98722). +# Terminal final_response when compression timed out while the request was still oversized (#98722). _COMPRESSION_TIMEOUT_FINAL_RESPONSE = ( "Context compression timed out without reducing this conversation. " "No messages were dropped. Start a fresh session with /new, or check " @@ -233,8 +222,7 @@ _COMPRESSION_TIMEOUT_FINAL_RESPONSE = ( ) -# Stable prefix of the local interrupt status string; surfaces (ACP, TUI) match on -# it to treat the text as cancellation metadata rather than assistant prose. +# Stable prefix ACP/TUI match on to treat the text as cancellation metadata, not assistant prose. INTERRUPT_WAITING_FOR_MODEL_PREFIX = "Operation interrupted: waiting for model response (" @@ -245,10 +233,9 @@ def _should_rearm_compression_budget( prompt_tokens: int, threshold_tokens: int, ) -> bool: - """Return True after a provider proves a completed compaction worked. - - Rough estimates cannot rearm the anti-thrash budget; require the completed- - compaction latch and a positive normalized prompt count below the threshold.""" + """True once a provider proves a completed compaction worked: rough estimates cannot + rearm the anti-thrash budget, only the completed-compaction latch plus a positive + normalized prompt count below the threshold.""" return bool( compression_attempts and completed_compaction_pending @@ -276,11 +263,8 @@ _MAX_OUTER_LOOP_ERRORS = 8 def _is_interpreter_shutdown_error(exc: Exception) -> bool: - """Check if *exc* is a fatal interpreter-shutdown failure. - - Delegates to ``tools.interpreter_shutdown`` (one text-matching site for the - shutdown-race bug class) but keeps the RuntimeError type gate: a ValueError - carrying similar text must not match (#93269).""" + """True for a fatal interpreter-shutdown RuntimeError. The RuntimeError type gate + stays here: a ValueError carrying similar text must not match (#93269).""" if isinstance(exc, RuntimeError): from tools.interpreter_shutdown import interpreter_shutting_down @@ -289,10 +273,9 @@ def _is_interpreter_shutdown_error(exc: Exception) -> bool: def _moa_client_consumes_prepared_request(client: Any) -> bool: - """True when ``client`` is the in-process MoA facade. - - Only ``MoAChatCompletions`` exposes ``prepare()``; other clients raise TypeError on - ``_moa_prepared_request`` even while ``agent.provider`` stays ``"moa"``.""" + """True when ``client`` is the in-process MoA facade (only ``MoAChatCompletions`` exposes + ``prepare()``; other clients raise TypeError on ``_moa_prepared_request`` even while + ``agent.provider`` stays ``"moa"``).""" completions = getattr(getattr(client, "chat", None), "completions", None) return callable(getattr(completions, "prepare", None)) @@ -308,10 +291,8 @@ def _join_truncated_parts(parts: List[str]) -> str: def _moa_reference_metrics_for_hook(agent: Any) -> Any: - """Per-advisor metrics for post_api_request, or None off the MoA path. - - MoA returns only the aggregator response, so a plugin sees one generation for - the whole fan-out; this carries the per-slot advisor spend across the hook boundary.""" + """Per-advisor metrics for post_api_request, or None off the MoA path (a plugin only + sees the aggregator generation; this carries the per-slot advisor spend).""" client = getattr(agent, "client", None) getter = getattr(client, "last_reference_metrics", None) if not callable(getter): @@ -337,56 +318,36 @@ def _apply_active_turn_redirect(agent: Any, messages: List[Dict[str, Any]], text checkpoint_parts = [_INTERRUPT_SCAFFOLD_MARKER] if visible: - checkpoint_parts.extend( - ["Visible response before the interruption:", visible] - ) + checkpoint_parts += ["Visible response before the interruption:", visible] checkpoint = "\n\n".join(checkpoint_parts) - correction = ( - "[Context from the interrupted assistant response]\n" - f"{checkpoint}\n\n" - f"{text}" - ) + correction = f"[Context from the interrupted assistant response]\n{checkpoint}\n\n{text}" # The live tail is normally user or tool, so an assistant placeholder + correction - # keeps strict alternation; if the tail is already assistant, fold the checkpoint - # into the user correction instead of creating assistant→assistant. - if messages and messages[-1].get("role") == "assistant": - # Transcript shows the user's own words; the provider replays the - # scaffolded form so it still sees the interrupted context. - append_message( - messages, - {"role": "user", "content": text, "api_content": correction}, - ) - else: - # Placeholder preserves role alternation only. Scaffold bytes must never land - # here: api_content is substituted back into content on replay (#81841). - placeholder: Dict[str, Any] = { - "role": "assistant", - "content": visible or "", - } + # keeps strict alternation; if the tail is already assistant, the checkpoint is folded + # into the user correction instead of creating assistant→assistant. The placeholder + # preserves alternation only — scaffold bytes must never land in it, since api_content + # is substituted back into content on replay (#81841). + if not (messages and messages[-1].get("role") == "assistant"): + placeholder: Dict[str, Any] = {"role": "assistant", "content": visible or ""} if not visible: placeholder["display_kind"] = "hidden" - # Hidden row, but a non-empty neutral api_content so the pre-call - # sanitizer does not re-heal it every call (#88955). Never - # _INTERRUPT_SCAFFOLD_MARKER: as assistant text the model echoes it (#81841) + # Hidden row, but a non-empty neutral api_content so the pre-call sanitizer + # does not re-heal it every call (#88955). Never _INTERRUPT_SCAFFOLD_MARKER: + # as assistant text the model echoes it (#81841). from agent.agent_runtime_helpers import _INTERRUPTED_PLACEHOLDER placeholder["api_content"] = _INTERRUPTED_PLACEHOLDER append_message(messages, placeholder) - append_message( - messages, - {"role": "user", "content": text, "api_content": correction}, - ) + # Transcript shows the user's own words; the provider replays the scaffolded form. + append_message(messages, {"role": "user", "content": text, "api_content": correction}) agent._current_streamed_assistant_text = "" agent._stream_needs_break = True def _is_copilot_provider(agent: Any) -> bool: - """Delegate to ``AIAgent._is_copilot_provider`` (single owner of the check). - - ``agent.provider`` may hold the aliases ``github-copilot`` / ``github``; a bare - ``provider == "copilot"`` gate would skip credential recovery for them.""" + """Delegate to ``AIAgent._is_copilot_provider``; the fallback keeps the ``github-copilot`` / + ``github`` aliases so credential recovery is not skipped for them.""" try: return bool(agent._is_copilot_provider()) except Exception: @@ -398,11 +359,9 @@ def _is_copilot_provider(agent: Any) -> bool: def _is_stale_copilot_credential_error(status_code: Optional[int], error_message: str) -> bool: - """Detect a Copilot 400 that is really a STALE / DEGRADED credential. - - Matches status 400 AND ``model_not_available_for_integrator`` or - ``model_not_supported`` / "the requested model is not supported", so a wrong model - name never triggers the single-shot re-exchange. Caller enforces scoping/guard.""" + """Detect a Copilot 400 that is really a STALE / DEGRADED credential (status 400 AND an + integrator/model-not-supported marker, so a wrong model name never triggers the + single-shot re-exchange). Caller enforces scoping/guard.""" lowered = (error_message or "").lower() if status_code != 400 and "error code: 400" not in lowered: return False @@ -455,10 +414,8 @@ def _ollama_context_limit_error(agent: Any, request_tokens: int) -> Optional[str def _maybe_grow_local_window(agent: Any, compressor: Any, request_tokens: int) -> Optional[int]: - """Try growing the managed local model's context window before compressing. - - Returns the new window when the ladder granted one, else None (hold / at native / - not a managed local session). Cheap for non-local providers: one compare.""" + """Grow a managed local model's context window before compressing; returns the new + window when the ladder granted one, else None.""" provider = (getattr(agent, "provider", "") or "").strip().lower() if provider not in ("llamacpp", "llama.cpp", "llama-cpp", "custom"): return None @@ -483,8 +440,7 @@ def _maybe_grow_local_window(agent: Any, compressor: Any, def _ra(): - """Lazy ``run_agent`` reference so patches on ``run_agent.handle_function_call`` / - ``run_agent._set_interrupt`` / ``run_agent.OpenAI`` reach this code path.""" + """Lazy ``run_agent`` reference so patches on ``run_agent.*`` reach this code path.""" import run_agent return run_agent @@ -520,10 +476,8 @@ def _print_nous_entitlement_guidance(agent, capability: str) -> bool: def _system_prompt_for_hooks(api_kwargs: Any, request_messages: Any) -> Any: - """System prompt as actually sent to the provider, for observability hooks. - - Checks ``system`` (Anthropic), ``instructions`` (Responses/Codex), then - ``messages[0]``. Returns None when the request carries no system prompt.""" + """System prompt as sent to the provider (``system`` / ``instructions`` / ``messages[0]``) + for observability hooks; None when the request carries none.""" system_prompt = api_kwargs.get("system") if system_prompt is None: system_prompt = api_kwargs.get("instructions") @@ -554,47 +508,42 @@ def _billing_or_entitlement_message( provider_label = (provider or "").strip() or "the selected provider" model_label = (model or "").strip() or "the selected model" - # Anthropic Pro/Max OAuth surfaces exhaustion of the "extra usage" bucket as a hard - # 400; point at the settings page and cycle reset — "add credits" does not apply. + # Anthropic Pro/Max OAuth surfaces "extra usage" exhaustion as a hard 400 — "add credits" + # does not apply. ``unverified`` (#82154): the same 400 is returned for a server-side + # content-filter rejection, so hedge and name the other cause. if (provider or "").strip().lower() == "anthropic": - # ``unverified`` (#82154): the "out of extra usage" 400 is also returned for a - # server-side content-filter rejection, so hedge and name the other cause. + switch = ( + "You can also switch to an Anthropic API key or another provider with " + "/model --provider ." + ) if unverified: lines = [ - ( - f"{provider_label} reported that your Claude subscription usage may be " - f"exhausted for {model_label} (included quota + extra-usage credits) — " - "but this specific error is not proof of a billing problem." - ), + f"{provider_label} reported that your Claude subscription usage may be " + f"exhausted for {model_label} (included quota + extra-usage credits) — " + "but this specific error is not proof of a billing problem.", "If https://claude.ai/settings/usage still shows quota remaining, this is " "probably NOT a billing problem: on a Claude subscription (OAuth) token " "Anthropic returns this same message when its content filter rejects part " "of the request — typically a phrase in the system prompt.", "If usage really is exhausted: wait for the billing cycle to reset, or add " "extra usage at https://claude.ai/settings/usage", - "You can also switch to an Anthropic API key or another provider with " - "/model --provider .", - # The exhaustion latch replays the stored error without issuing - # a request, so a real fix looks like it didn't work. + switch, + # The exhaustion latch replays the stored error without a request. "Retry with a fresh credential state: `hermes auth reset anthropic`. Until " "that cooldown clears, this error can be replayed from cache without " "contacting the API.", ] else: lines = [ - ( - f"{provider_label} reported that your Claude subscription usage is " - f"exhausted for {model_label} (included quota + extra-usage credits)." - ), + f"{provider_label} reported that your Claude subscription usage is " + f"exhausted for {model_label} (included quota + extra-usage credits).", "Options: wait for the billing cycle to reset, or add extra usage at " "https://claude.ai/settings/usage", - "You can also switch to an Anthropic API key or another provider with " - "/model --provider .", + switch, ] return "\n".join(lines) - # Provider-agnostic billing URL so every text surface (CLI, gateway, TUI) shows the - # same actionable link, not just OpenRouter. + # Provider-agnostic billing URL so every text surface shows the same actionable link. try: from agent.billing_links import build_billing_block @@ -606,10 +555,8 @@ def _billing_or_entitlement_message( billing_url = None lines = [ - ( - f"{provider_label} reported that billing, credits, or account " - f"entitlement is exhausted for {model_label}." - ), + f"{provider_label} reported that billing, credits, or account " + f"entitlement is exhausted for {model_label}.", "Add credits or update billing with that provider, then retry.", ] if billing_url: @@ -631,17 +578,13 @@ def _billing_block_dict( except Exception: return None if block is not None and unverified: - # Carry the classifier's ambiguity into the structured descriptor so - # every surface rendering the block can hedge too (#82154). - block["unverified"] = True + block["unverified"] = True # every surface rendering the block can hedge too (#82154) return block def _billing_terminal_label(summary: str, unverified: bool) -> str: - """Terminal-failure prefix for a billing-classified error. - - ``unverified`` (#82154): the Anthropic "out of extra usage" 400 can be a - content-filter rejection, so the line must not assert exhaustion as fact.""" + """Terminal-failure prefix for a billing-classified error; ``unverified`` (#82154) must + not assert exhaustion as fact.""" if unverified: return ( "Provider reported usage/credit exhaustion (unverified — the same " @@ -661,10 +604,8 @@ def _billing_failure_result( model: str, guidance: Optional[str] = None, ) -> dict: - """Structured terminal result for a billing-classified failure. - - Single construction point so label, guidance, structured block and ambiguity flag - stay consistent across the non-retryable abort and max-retries paths (#82154).""" + """Structured terminal result for a billing-classified failure — the single construction + point for the non-retryable abort and max-retries paths (#82154).""" unverified = bool(getattr(classified, "billing_unverified", False)) if guidance is None: guidance = _billing_or_entitlement_message( @@ -685,11 +626,8 @@ def _billing_failure_result( "failed": True, "error": summary, "failure_reason": classified.reason.value, - # Classifier's own retry verdict so UI (agent/error_surface.py) shows Retry - # only when a re-run can differ, not re-derived from a second taxonomy. + # Classifier's own retry verdict so the UI shows Retry only when a re-run can differ. "failure_retryable": bool(classified.retryable), - # The billing verdict may rest on an ambiguous body (#82154) — carry - # that through the structured result, not just the prose. "billing_unverified": unverified, "billing_block": _billing_block_dict( provider, base_url, model, guidance, unverified=unverified @@ -782,13 +720,8 @@ def _restore_or_build_system_prompt(agent, system_message, conversation_history) session_row = agent._session_db.get_session(agent.session_id) if session_row is not None: raw_prompt = session_row.get("system_prompt") - if raw_prompt is None: - stored_state = "null" - elif raw_prompt == "": - stored_state = "empty" - else: - stored_prompt = raw_prompt - stored_state = "present" + stored_state = "null" if raw_prompt is None else ("empty" if raw_prompt == "" else "present") + stored_prompt = raw_prompt or None except Exception as exc: logger.warning( "Session DB get_session failed for system-prompt restore " @@ -905,10 +838,8 @@ def _stored_prompt_matches_runtime(agent, prompt: str) -> bool: """Return False when the persisted runtime-identity lines are stale.""" def line_value(label: str) -> str: - """Last matching line wins. - - Safe ONLY for fields in the volatile tier at the END of the prompt; embedded - project context could shadow earlier fields — see ``host_info_value``.""" + """Last matching line wins — safe ONLY for volatile-tier fields at the END of the + prompt (embedded project context could shadow earlier fields; see ``host_info_value``).""" prefix = f"{label}:" value = "" for line in prompt.splitlines(): @@ -917,10 +848,8 @@ def _stored_prompt_matches_runtime(agent, prompt: str) -> bool: return value def host_info_value(label: str) -> str: - """Read a field from the prompt's own host-info block. - - Anchors on the FIRST ``User home directory:`` line so a user's ``AGENTS.md`` row - cannot match; a false mismatch would rebuild the prompt every turn.""" + """Read a field from the prompt's own host-info block, anchored on the FIRST ``User + home directory:`` line so a user's ``AGENTS.md`` row cannot force a rebuild every turn.""" prefix = f"{label}:" lines = prompt.splitlines() for idx, line in enumerate(lines): @@ -948,9 +877,8 @@ def _stored_prompt_matches_runtime(agent, prompt: str) -> bool: return not (stored_platform and current_platform and stored_platform != current_platform) -# Named constants for the _get_continuation_prompt variants so -# _is_synthetic_compression_user_turn can recognize them by content after a crash -# persists one; SessionDB projection strips the _length_continuation_nudge tag. +# Named so _is_synthetic_compression_user_turn can recognize a crash-persisted nudge by +# content (SessionDB projection strips the _length_continuation_nudge tag). _LENGTH_CONTINUATION_NETWORK_STUB = ( "[System: The previous response was cut off by a " "network error mid-stream. Continue exactly where " @@ -962,8 +890,7 @@ _LENGTH_CONTINUATION_OUTPUT_LIMIT = ( "length limit. Continue exactly where you left off. Do not " "restart or repeat prior text. Finish the answer directly.]" ) -# The dropped-tools variant interpolates tool names, so -# _is_synthetic_compression_user_turn matches this prefix with str.startswith. +# The dropped-tools variant interpolates tool names; matched by prefix. _LENGTH_CONTINUATION_DROPPED_TOOLS_PREFIX = "[System: Your previous tool call " @@ -989,8 +916,8 @@ def _get_continuation_prompt(is_partial_stub: bool, dropped_tools: Optional[List return _LENGTH_CONTINUATION_OUTPUT_LIMIT -# Nudge for Codex/Responses turns that returned only internal reasoning: a bare retry -# would be byte-identical (nothing replayable emitted), so the model repeats it. +# Codex/Responses turns that returned only internal reasoning: a bare retry would be +# byte-identical, so the model repeats it. _CODEX_INCOMPLETE_NUDGE = ( "[System: Your previous response contained only internal reasoning and " "never produced a visible answer or tool call. Do not keep thinking. " @@ -999,23 +926,21 @@ _CODEX_INCOMPLETE_NUDGE = ( ) -# Re-prompt after an acknowledgment-only Codex/Responses reply; named so -# _is_synthetic_compression_user_turn can recognize it like _CODEX_INCOMPLETE_NUDGE. +# Re-prompt after an acknowledgment-only Codex/Responses reply. _CODEX_ACK_CONTINUATION_NUDGE = ( "[System: Continue now. Execute the required tool calls and only " "send your final answer after completing the task.]" ) -# Re-prompt for finish_reason="tool_calls" with empty tool_calls. Named like -# _CODEX_ACK_CONTINUATION_NUDGE: an interrupt mid-retry can persist it. +# Re-prompt for finish_reason="tool_calls" with empty tool_calls (an interrupt mid-retry can persist it). _DROPPED_TOOLCALL_NUDGE_CONTENT = ( "Your previous turn indicated a tool call but none was " "included. Do not narrate a plan or restate intent — issue " "the actual tool call now to continue the task." ) -# Re-prompt for an empty response after tool calls (#9400). Named because its -# _empty_recovery_synthetic metadata flag does not survive SessionDB projection. +# Re-prompt for an empty response after tool calls (#9400); the metadata flag does not +# survive SessionDB projection, so it is matched by content. _EMPTY_TOOL_RESPONSE_NUDGE = ( "You just executed tool calls but returned an " "empty response. Please process the tool " @@ -1023,30 +948,26 @@ _EMPTY_TOOL_RESPONSE_NUDGE = ( ) -# Shared recovery trailer for both content-policy refusal paths (HTTP-200 -# content_filter and the content_policy_blocked exception) so guidance cannot drift. +# Shared trailer for both content-policy refusal paths so guidance cannot drift. _CONTENT_POLICY_RECOVERY_HINT = ( "Try rephrasing the request, narrowing the context, or " "adding a fallback provider with `hermes fallback add`." ) -# Memo for send-path tool-call argument canonicalization, which re-runs on every -# historical call each iteration. Sound: canonicalization is pure and deterministic; -# malformed strings raise before being stored, so the repair fallback is never memoized. +# Memo for send-path tool-call argument canonicalization (re-run on every historical call +# each iteration). Sound because canonicalization is pure; malformed strings raise before +# being stored, so the repair fallback is never memoized. The byte budget exists because +# argument strings can run 100KB+, so a count bound alone does not bound memory. _CANON_ARGS_CACHE: Dict[str, str] = {} _CANON_ARGS_CACHE_MAX = 4096 -# Count bound alone does not bound MEMORY: argument strings can run 100KB+, so a byte -# budget bounds the worst case while keeping the memo effective for ~0.5-2KB args. _CANON_ARGS_CACHE_MAX_BYTES = 32 * 1024 * 1024 _canon_args_cache_bytes = 0 def _canonicalize_tool_call_arguments(arg_str: str) -> str: - """Return the canonical wire form of a tool-call arguments JSON string. - - Raises whatever ``json.loads`` raises on malformed input; the caller falls back to - ``_repair_tool_call_arguments``.""" + """Canonical wire form of a tool-call arguments JSON string; raises on malformed input + (the caller falls back to ``_repair_tool_call_arguments``).""" global _canon_args_cache_bytes cached = _CANON_ARGS_CACHE.get(arg_str) if cached is not None: @@ -1070,11 +991,9 @@ def _canonicalize_tool_call_arguments(arg_str: str) -> str: def _clone_message_for_send(msg): - """Structural clone of a history message for the per-call API copy. - - Clones every dict/list recursively while sharing immutable leaves, so in-place - send-path rewrites can never reach the persisted transcript (#80498). Cheaper than - copy.deepcopy; messages are JSON-shaped and acyclic, tuples are shared as leaves.""" + """Structural clone (dicts/lists recursively, immutable leaves shared) of a history + message for the per-call API copy, so send-path rewrites never reach the persisted + transcript (#80498). Cheaper than deepcopy: messages are JSON-shaped and acyclic.""" if isinstance(msg, dict): return { k: _clone_message_for_send(v) if isinstance(v, (dict, list)) else v @@ -1089,10 +1008,8 @@ def _clone_message_for_send(msg): def _canonicalize_api_tool_calls(api_messages) -> None: - """Canonicalize tool-call argument JSON on the send-path message copy. - - Rewrites ``tool_calls`` in place (copy-on-write for the dicts it touches; persisted - history untouched). The memo bounds parse/serialize to one per UNIQUE string.""" + """Canonicalize tool-call argument JSON on the send-path copy (copy-on-write for the + dicts it touches; persisted history untouched).""" for am in api_messages: tcs = am.get("tool_calls") if not tcs: @@ -1114,11 +1031,9 @@ def _canonicalize_api_tool_calls(api_messages) -> None: def _invalid_tool_name_error_content(name: str, valid_tool_names) -> str: - """Error-result content for a tool call whose name isn't a real tool. - - A blank name is a model echoing tool-call syntax seen in data, not a typo (#47967); - dumping the catalog feeds that loop, so send a terse error instead. A nonempty wrong - name still gets the catalog so the model can self-correct.""" + """Error content for an unknown tool name. A blank name is a model echoing tool-call + syntax seen in data (#47967) — dumping the catalog feeds that loop, so it gets a terse + error; a nonempty wrong name still gets the catalog to self-correct.""" if not (name or "").strip(): return ( "Tool call rejected: the tool name was empty. " @@ -1139,10 +1054,8 @@ def _content_policy_blocked_result( final_response: str, error_detail: str, ) -> Dict[str, Any]: - """Build the terminal turn result for a content-policy block. - - Refusals are deterministic for the unchanged prompt, so no retry; both the HTTP-200 - and exception paths return this shape with a ``content_policy_blocked:`` error.""" + """Terminal turn result for a content-policy block (deterministic for the unchanged + prompt, so no retry); shared by the HTTP-200 and exception paths.""" return { "final_response": final_response, "messages": messages, @@ -1153,15 +1066,29 @@ def _content_policy_blocked_result( } +def _partial_turn_result( + final_response: str, messages: List[Dict], api_call_count: int, **flags: Any +) -> Dict[str, Any]: + """Incomplete-turn result whose ``error`` mirrors ``final_response``; ``flags`` add the + recovery-contract keys (``failed``, ``compression_deferred``, ...).""" + return { + "final_response": final_response, + "messages": messages, + "completed": False, + "api_calls": api_call_count, + "error": final_response, + "partial": True, + **flags, + } + + def _compression_deferred_result( agent, messages: List[Dict], api_call_count: int, reason: str = "lock", ) -> Dict[str, Any]: - """Build the soft turn result for a transiently-deferred compression. - - Both ``reason="lock"`` and ``reason="transient_block"`` must end as + """Soft turn result for a transiently-deferred compression. Both reasons must end as ``compression_deferred``, never ``compression_exhausted`` — the gateway wipes the session on exhaustion (#9893/#35809). ``failed`` stays False; the turn persists.""" session = agent.session_id or "none" @@ -1193,17 +1120,10 @@ def _compression_deferred_result( agent._flush_status_buffer() except Exception: pass - return { - "final_response": _final, - "messages": messages, - "completed": False, - "api_calls": api_call_count, - "error": _final, - "partial": True, - "failed": False, - "compression_deferred": True, - "session_id": agent.session_id, - } + return _partial_turn_result( + _final, messages, api_call_count, + failed=False, compression_deferred=True, session_id=agent.session_id, + ) def _provider_overflow_exhausted_result( @@ -1224,29 +1144,19 @@ def _provider_overflow_exhausted_result( f"{request_pressure_tokens:,}", ) agent._persist_session(messages, conversation_history) - final_response = ( + return _partial_turn_result( "Context length exceeded: compression could not reduce the rebuilt " - "request below the safe threshold." + "request below the safe threshold.", + messages, api_call_count, + failed=True, compression_exhausted=True, + turn_exit_reason="context_compression_exhausted", ) - return { - "final_response": final_response, - "messages": messages, - "completed": False, - "api_calls": api_call_count, - "error": final_response, - "partial": True, - "failed": True, - "compression_exhausted": True, - "turn_exit_reason": "context_compression_exhausted", - } def _rewrite_system_content_blocks(system_message: dict, effective: str) -> bool: - """Rewrite a cache-decorated system message in place, keeping its blocks. - - Assigning a bare string over the ``[static prefix, volatile tail]`` block list drops - both cache_control breakpoints. Only the LAST ``Model:``/``Provider:`` lines change. - Returns False when the shape cannot be safely patched.""" + """Rewrite a cache-decorated system message in place, keeping its blocks (a bare string + over the ``[static prefix, volatile tail]`` list would drop both cache_control + breakpoints). Returns False when the shape cannot be safely patched.""" content = system_message.get("content") if not isinstance(content, list) or not content: return False @@ -1268,11 +1178,8 @@ def _rewrite_system_content_blocks(system_message: dict, effective: str) -> bool def _sync_failover_system_message(agent, api_messages, active_system_prompt): - """Refresh the in-flight system message after a provider failover. - - ``try_activate_fallback`` rewrites the identity lines on ``_cached_system_prompt``, - but this call block's ``api_messages`` were built pre-failover and are reused each - retry. Mutates ``api_messages[0]`` in place; returns the new ``active_system_prompt``.""" + """Refresh the in-flight system message after a provider failover: ``api_messages`` were + built pre-failover and are reused each retry. Returns the new ``active_system_prompt``.""" sp = getattr(agent, "_cached_system_prompt", None) if not isinstance(sp, str) or not sp: return active_system_prompt @@ -1286,10 +1193,9 @@ def _sync_failover_system_message(agent, api_messages, active_system_prompt): def _arm_fallback_restart(agent, api_messages, active_system_prompt, _retry): - """After ``_try_activate_fallback`` succeeded: sync the system message to the new - provider and arm ``restart_with_rebuilt_messages`` (re-issue against the fallback, - refunding the stalled attempt). Callers also reset ``retry_count`` / - ``compression_attempts`` to 0 and ``break`` the retry loop.""" + """After a successful fallback activation: sync the system message and arm + ``restart_with_rebuilt_messages``. Callers also zero ``retry_count`` / + ``compression_attempts`` and ``break`` the retry loop.""" active_system_prompt = _sync_failover_system_message( agent, api_messages, active_system_prompt) _retry.primary_recovery_attempted = False @@ -1298,11 +1204,9 @@ def _arm_fallback_restart(agent, api_messages, active_system_prompt, _retry): def _ensure_cached_system_prompt_static(agent, system_message=None) -> None: - """Rebuild ``_cached_system_prompt_static`` when caching becomes active (#72626). - - Sessions restored under a cache-off primary skip the static-prefix rebuild; a later - failover to a cache-on provider would otherwise silently fall back to the legacy - system-plus-3 layout. Wraps ``reconstruct_static_prefix`` (memoizes failures).""" + """Rebuild ``_cached_system_prompt_static`` when caching becomes active (#72626): sessions + restored under a cache-off primary would otherwise fall back to the legacy layout after + failover to a cache-on provider.""" from agent.system_prompt import reconstruct_static_prefix reconstruct_static_prefix( @@ -1314,9 +1218,7 @@ def _peel_moa_guidance( messages: List[Dict[str, Any]], guidance: Any, ) -> List[Dict[str, Any]]: - """Remove MoA reference guidance attached by ``_attach_reference_guidance``. - - Kept adjacent to the attach so the forward/inverse shapes evolve together.""" + """Remove MoA reference guidance attached by ``_attach_reference_guidance``.""" from agent.moa_loop import peel_reference_guidance return peel_reference_guidance(messages, guidance) @@ -1330,11 +1232,8 @@ def _redecorate_prompt_cache_for_provider( moa_prepared: Optional[Dict[str, Any]] = None, tools_for_api: Optional[List[Dict[str, Any]]] = None, ) -> tuple[List[Dict[str, Any]], Optional[Dict[str, Any]]] | tuple[List[Dict[str, Any]], Optional[Dict[str, Any]], List[Dict[str, Any]]]: - """Strip and re-apply cache_control for the *current* provider policy. - - Decoration runs once per call block for the primary provider, but failover - ``continue`` paths reuse ``api_messages`` (#72626), so reshape at the top of each - retry from the mutated in-flight request. MoA guidance is peeled and rebased.""" + """Strip and re-apply cache_control for the *current* provider policy — failover + ``continue`` paths reuse ``api_messages`` (#72626). MoA guidance is peeled and rebased.""" messages: List[Dict[str, Any]] = [ dict(m) if isinstance(m, dict) else m for m in (api_messages or []) ] @@ -1362,9 +1261,7 @@ def _redecorate_prompt_cache_for_provider( _ensure_cached_system_prompt_static(agent, system_message=system_message) static = getattr(agent, "_cached_system_prompt_static", None) direct_tool_cache = getattr( - agent, - "_direct_native_anthropic_tool_cache_capability", - lambda: False, + agent, "_direct_native_anthropic_tool_cache_capability", lambda: False )() from agent.prompt_caching import envelope_tool_part_cache_markers_supported @@ -1419,10 +1316,8 @@ def _apply_context_engine_selection( *, logger: Any, ) -> List[Dict[str, Any]]: - """Run the optional per-turn ``ContextEngine.select_context()`` hook. - - Returns the (possibly replaced) request list. Fail-open: a missing hook, exception, - or invalid return yields ``api_messages`` unchanged; history is never mutated.""" + """Run the optional per-turn ``ContextEngine.select_context()`` hook, fail-open: any + exception or invalid return yields ``api_messages`` unchanged; history is never mutated.""" engine = getattr(agent, "context_compressor", None) if not _engine_overrides_hook(engine, "select_context"): return api_messages @@ -1430,14 +1325,17 @@ def _apply_context_engine_selection( session_label = getattr(agent, "session_id", None) or "-" # Structural clones: the engine must not be able to write through nested # containers into persisted history; only the request list is acted on (#80498). - _conv_copy = [_clone_message_for_send(m) for m in conversation_messages] \ - if conversation_messages is not None else None - _incoming_copy = _clone_message_for_send(incoming_message) if isinstance(incoming_message, dict) else incoming_message try: selected = engine.select_context( api_messages, - conversation_messages=_conv_copy, - incoming_message=_incoming_copy, + conversation_messages=( + [_clone_message_for_send(m) for m in conversation_messages] + if conversation_messages is not None else None + ), + incoming_message=( + _clone_message_for_send(incoming_message) + if isinstance(incoming_message, dict) else incoming_message + ), budget_tokens=getattr(engine, "context_length", 0) or 0, ) except Exception: @@ -1455,7 +1353,6 @@ def _apply_context_engine_selection( # buggy engine would otherwise replace the request instead of failing open. if isinstance(selected, list) and selected and all(isinstance(m, dict) for m in selected): return selected - logger.warning( "Context engine select_context returned an invalid value " "(not a non-empty list of dicts); ignoring (session=%s)", @@ -1472,10 +1369,8 @@ def _notify_context_engine_turn_complete( logger: Any, **meta: Any, ) -> None: - """Notify the active context engine that a user turn has finished. - - Fail-open: a missing/no-op hook or any exception is swallowed. ``messages`` is - passed as a copy so the engine cannot mutate the persisted transcript.""" + """Notify the active context engine that a user turn has finished (fail-open; the engine + gets a copy so it cannot mutate the persisted transcript).""" engine = getattr(agent, "context_compressor", None) if not _engine_overrides_hook(engine, "on_turn_complete"): return @@ -1496,10 +1391,8 @@ def _notify_context_engine_turn_complete( def _decode_inline_moa_turn(user_message, persist_user_message): - """Decode a MoA preset encoded into ``user_message`` (``hermes_cli.moa_config``). - - Returns ``(user_message, moa_config, persist_user_message)``; unchanged with - ``moa_config=None`` when nothing is encoded or decoding fails.""" + """Decode a MoA preset encoded into ``user_message``; returns ``(user_message, + moa_config, persist_user_message)``, unchanged with ``moa_config=None`` otherwise.""" try: from hermes_cli.moa_config import decode_moa_turn @@ -1514,34 +1407,24 @@ def _decode_inline_moa_turn(user_message, persist_user_message): def _preflight_timeout_result(agent, exc, conversation_history) -> Dict[str, Any]: - """Typed recovery result when turn-start preflight compression timed out (#98424): - no provider call was sent. Surfaces hide raw exception text, which would bury the - actionable guidance and skip the compression_exhausted recovery contract.""" + """Typed recovery result when turn-start preflight compression timed out (#98424): no + provider call was sent, and surfaces would otherwise hide the actionable guidance.""" logger.warning( "Turn-start preflight compression timed out — ending turn with " "typed recovery result: %s", exc, ) - # Clear the tripwire slot note_turn_start registered; the early return skips the - # persist funnel that clears it. The user row is deliberately NOT persisted: - # the gateway skips persistence for compression_exhausted results (#7100). + # Clear the tripwire slot note_turn_start registered (the early return skips the persist + # funnel). The user row is deliberately NOT persisted (#7100). from agent.agent_runtime_helpers import note_turn_persisted note_turn_persisted(agent) # Not _COMPRESSION_TIMEOUT_FINAL_RESPONSE — that describes a different state # (compression ran, could not reduce); the exception text carries the guidance. - _final_response = str(exc) - return { - "final_response": _final_response, - "messages": list(conversation_history or []), - "completed": False, - "api_calls": 0, - "error": _final_response, - "partial": True, - "failed": True, - "compression_exhausted": True, - "turn_exit_reason": "context_compression_timeout", - } + return _partial_turn_result( + str(exc), list(conversation_history or []), 0, + failed=True, compression_exhausted=True, turn_exit_reason="context_compression_timeout", + ) @dataclass diff --git a/agent/empty_response_guard.py b/agent/empty_response_guard.py index 9f163d1ed7..492dfcb765 100644 --- a/agent/empty_response_guard.py +++ b/agent/empty_response_guard.py @@ -1,30 +1,22 @@ """Deterministic-empty detection and cost-aware retry budgets. -On an empty completion the loop retries up to 3 times, then walks the -fallback chain — each attempt re-bills the full input. Signaled refusals -(``content_filter``, Anthropic ``refusal``, Bedrock guardrails) are already -terminal; this module handles *unsignaled* empties (success, zero output -tokens, generic finish reason — typical of portal-proxied refusals). +On an empty completion the loop retries up to 3 times, then walks the fallback chain — +each attempt re-bills the full input. Signaled refusals (``content_filter``, Anthropic +``refusal``, Bedrock guardrails) are already terminal; this handles *unsignaled* empties +(success, zero output tokens, generic finish reason — typical of portal-proxied refusals). Two independent guards, both failing OPEN to legacy behaviour: 1. Deterministic-empty: two consecutive empties, both with usage present and - ``output_tokens == 0``, from the same (model, provider, finish_reason) → - the same prompt will keep producing the same empty, so skip remaining - retries and go straight to the fallback chain (a different model may - behave differently). Missing usage or ``output_tokens > 0`` (think-block - stripping, whitespace, flaky decoding) never classifies as deterministic. -2. Cost-aware budget: when one empty attempt's estimated input cost exceeds - the threshold (default $0.25), the retry budget drops from 3 to 1. - Unknown pricing / missing usage / included routes leave it untouched. + ``output_tokens == 0``, from the same (model, provider, finish_reason) → skip the + remaining retries and go straight to the fallback chain. Missing usage or + ``output_tokens > 0`` (think-block stripping, whitespace, flaky decoding) never classifies. +2. Cost-aware budget: when one empty attempt's estimated input cost exceeds the threshold + (default $0.25), the retry budget drops from 3 to 1. Unknown pricing / missing usage / + included routes leave it untouched. -Config (``config.yaml``, resolved once by ``agent_init`` and stashed on the -agent so the hot loop never re-reads config — no env vars):: - - agent: - empty_response_guard: - enabled: true # false = legacy fixed 3-retry behaviour - cost_threshold_usd: 0.25 # per-attempt cost that halves the budget +Config ``agent.empty_response_guard.{enabled, cost_threshold_usd}`` is resolved once by +``agent_init`` and stashed on the agent so the hot loop never re-reads config (no env vars). """ from __future__ import annotations @@ -41,10 +33,9 @@ REDUCED_EMPTY_RETRY_BUDGET = 1 DEFAULT_COST_THRESHOLD_USD = Decimal("0.25") DEFAULT_GUARD_ENABLED = True -# Agent-object attribute names. State is scoped to one consecutive empty -# streak: cleared whenever ``_empty_content_retries == 0`` at record time, so -# every existing counter-reset site (turn start, compaction, tool success, -# fallback activation) is honoured without touching it. +# Agent-object attribute names. State is scoped to one consecutive empty streak: cleared +# whenever ``_empty_content_retries == 0`` at record time, so every existing counter-reset +# site (turn start, compaction, tool success, fallback activation) is honoured. _ATTEMPTS_ATTR = "_empty_attempt_history" _STREAK_COST_ATTR = "_empty_streak_cost_usd" _ENABLED_ATTR = "_empty_guard_enabled" @@ -160,9 +151,8 @@ def _zero_output(agent: Any, response: Any) -> tuple: def record_empty_attempt(agent: Any, *, finish_reason: str, response: Any) -> None: """Record one empty completion in the current streak. - Call BEFORE ``_empty_content_retries`` is incremented: a counter of 0 - marks a new streak and clears prior history. - """ + Call BEFORE ``_empty_content_retries`` is incremented: a counter of 0 marks a new + streak and clears prior history.""" attempts = _attempts(agent) if getattr(agent, "_empty_content_retries", 0) == 0: attempts.clear() @@ -181,9 +171,8 @@ def record_empty_attempt(agent: Any, *, finish_reason: str, response: Any) -> No def deterministic_empty(agent: Any) -> bool: - """True when >= 2 consecutive attempts ALL have usage present, zero output - and an identical signature. Any missing-usage / non-zero attempt → False - (fail open — transients deserve their retries).""" + """True when >= 2 consecutive attempts ALL have usage present, zero output and an + identical signature. Any missing-usage / non-zero attempt → False (fail open).""" if not guard_enabled(agent): return False attempts = getattr(agent, _ATTEMPTS_ATTR, None) or [] @@ -193,8 +182,8 @@ def deterministic_empty(agent: Any) -> bool: def empty_retry_budget(agent: Any, response: Any) -> int: - """Empty-retry budget for the current streak (3, or 1 when a single - attempt is estimated to cost more than the configured threshold).""" + """Empty-retry budget for the current streak (3, or 1 when a single attempt is + estimated to cost more than the configured threshold).""" if not guard_enabled(agent): return DEFAULT_EMPTY_RETRY_BUDGET cost = _estimate_attempt_cost(agent, response) diff --git a/agent/interrupt_compat.py b/agent/interrupt_compat.py index 22f2784c79..50a4dfa7b9 100644 --- a/agent/interrupt_compat.py +++ b/agent/interrupt_compat.py @@ -13,12 +13,9 @@ def _accepts_keyword(callable_obj: Any, name: str) -> bool: except (TypeError, ValueError): return False return any( - parameter.kind is inspect.Parameter.VAR_KEYWORD - or ( - parameter.name == name - and parameter.kind is not inspect.Parameter.POSITIONAL_ONLY - ) - for parameter in parameters + p.kind is inspect.Parameter.VAR_KEYWORD + or (p.name == name and p.kind is not inspect.Parameter.POSITIONAL_ONLY) + for p in parameters ) @@ -30,20 +27,14 @@ def request_hard_interrupt( ) -> bool: """Request an explicit stop, falling back to the legacy interrupt ABI. - New agents expose ``hard_interrupt(message=None)``. Third-party agents and - old test doubles may only expose ``interrupt(message=None)``; keep those - usable without sending newer keyword arguments they do not know. - - ``message`` is diagnostic/control-plane text. ``tool_reason`` is a trusted, - fixed category that may be exposed in model-visible tool cancellation - output. It is only forwarded when the modern callable explicitly supports - that channel. - Returns ``False`` only when neither callable is available. + New agents expose ``hard_interrupt(message=None)``; third-party agents and old test + doubles may only expose ``interrupt(message=None)`` and must not receive keyword + arguments they do not know. ``tool_reason`` is a trusted, fixed category that may be + exposed in model-visible tool cancellation output, forwarded only when the callable + explicitly supports it. Returns ``False`` only when neither callable is available. """ - # Avoid treating a dynamic ``__getattr__`` proxy (notably an unspecced - # ``MagicMock`` or a third-party RPC facade) as if it genuinely implements - # the new ABI. Static lookup proves the attribute exists on the instance or - # its type before normal descriptor binding retrieves the callable. + # Static lookup first: a dynamic ``__getattr__`` proxy (unspecced MagicMock, RPC + # facade) must not be treated as genuinely implementing the new ABI. try: inspect.getattr_static(agent, "hard_interrupt") except AttributeError: diff --git a/agent/iteration_budget.py b/agent/iteration_budget.py index 0bb517bfcf..edcb32a101 100644 --- a/agent/iteration_budget.py +++ b/agent/iteration_budget.py @@ -1,10 +1,9 @@ """Per-agent iteration budget — thread-safe consume/refund counter. -Each ``AIAgent`` (parent or subagent) holds its own :class:`IterationBudget`: -the parent's cap is ``max_iterations`` (default 500), each subagent's is -``delegation.max_iterations`` (default 50), so total iterations across parent -+ subagents can exceed the parent's cap. ``run_agent`` re-exports the class so -``from run_agent import IterationBudget`` keeps working. +Each ``AIAgent`` (parent or subagent) holds its own :class:`IterationBudget`: the parent's +cap is ``max_iterations`` (default 500), each subagent's ``delegation.max_iterations`` +(default 50), so total iterations across parent + subagents can exceed the parent's cap. +``run_agent`` re-exports the class so ``from run_agent import IterationBudget`` keeps working. """ from __future__ import annotations @@ -13,9 +12,8 @@ import threading class IterationBudget: - """Thread-safe iteration counter. ``execute_code`` (programmatic tool - calling) iterations are refunded via :meth:`refund` so they don't eat - into the budget.""" + """Thread-safe iteration counter; ``execute_code`` (programmatic tool calling) + iterations are refunded via :meth:`refund` so they don't eat into the budget.""" def __init__(self, max_total: int): self.max_total = max_total