From f76fe598a31ee52e5d0d6e048cf41c55f4d5a2ee Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 19:11:11 -0700 Subject: [PATCH] refactor(agent/conversation_loop): lift the API retry loop into _run_api_retry_loop; drop blank lines after lazy imports --- agent/conversation_loop.py | 130 ++++++++-------------- agent/deadline.py | 4 - agent/empty_response_guard.py | 2 - tests/agent/test_surrogate_chokepoints.py | 2 +- 4 files changed, 49 insertions(+), 89 deletions(-) diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index a1013b649f..58c42d9685 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -96,10 +96,7 @@ def _midturn_request_pressure_tokens( 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, - ) - + from agent.codex_responses_adapter import (estimate_native_responses_preflight_tokens) native = estimate_native_responses_preflight_tokens( agent, api_messages, system_prompt=effective_system or "", tools=getattr(agent, "tools", None) or None, @@ -183,7 +180,6 @@ def _should_skip_model_call_for_reference_handoff( ) -> bool: """Guard post-compaction continues against sole-handoff active turns (#80622).""" from agent.context_compressor import reference_handoff_would_drive_next_model_call - # A restored ask is an actionable non-synthetic user row appended after the # handoff — by construction the handoff no longer drives. return reference_handoff_would_drive_next_model_call(messages) and not ( @@ -235,9 +231,7 @@ _LOCAL_PROCESSING_MODULES = frozenset({ "message_sanitization", "chat_completion_helpers", # only local when NOT also an API-call module }) -_API_CALL_MODULES = frozenset({ - "chat_completion_helpers", -}) +_API_CALL_MODULES = frozenset({"chat_completion_helpers"}) # Max outer-loop exceptions per user turn before giving up; only exceptions that # ESCAPE the inner retry/fallback machinery count, so this can be small (#92450). @@ -249,7 +243,6 @@ def _is_interpreter_shutdown_error(exc: Exception) -> bool: stays here: a ValueError carrying similar text must not match (#93269).""" if isinstance(exc, RuntimeError): from tools.interpreter_shutdown import interpreter_shutting_down - return interpreter_shutting_down(exc) return False @@ -317,7 +310,6 @@ def _apply_active_turn_redirect(agent: Any, messages: List[Dict[str, Any]], text # 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) # Transcript shows the user's own words; the provider replays the scaffolded form. @@ -374,7 +366,6 @@ def _ollama_context_limit_error(agent: Any, request_tokens: int) -> Optional[str request_tokens, len(getattr(agent, "tools", None) or []), getattr(agent, "session_id", None) or "none", ) - return ( f"Ollama loaded `{model}` with only {runtime_ctx:,} tokens of runtime context, but Hermes " f"needs at least {MINIMUM_CONTEXT_LENGTH:,} tokens for reliable tool use.\n\n" @@ -398,7 +389,6 @@ def _maybe_grow_local_window(agent: Any, compressor: Any, return None try: from hermes_cli.local_runtime.growth import maybe_grow_window - current_window = int(getattr(compressor, "context_length", 0) or 0) if current_window <= 0: return None @@ -423,7 +413,6 @@ def _nous_entitlement_message(capability: str) -> str: format_nous_portal_entitlement_message, get_nous_portal_account_info, ) - account_info = get_nous_portal_account_info(force_fresh=True) return format_nous_portal_entitlement_message(account_info, capability=capability) or "" except Exception: @@ -514,7 +503,6 @@ def _billing_or_entitlement_message( # Provider-agnostic billing URL so every text surface shows the same actionable link. try: from agent.billing_links import build_billing_block - _link = build_billing_block(provider=provider, base_url=base_url, model=model) if _link.provider_label: provider_label = _link.provider_label @@ -539,7 +527,6 @@ def _billing_block_dict( """Best-effort structured billing descriptor (None if billing_links is unavailable).""" try: from agent.billing_links import build_billing_block - block = build_billing_block( provider=provider, base_url=str(base_url), model=model, message=message ).to_dict() @@ -627,11 +614,9 @@ def _bot_chat_prompt_stale(agent, stored_prompt: str) -> bool: stored_bot_chat_prompt_needs_upgrade, stored_prompt_capability_stale, ) - home = None try: from agent.system_prompt import _agent_home - home = _agent_home(agent) except Exception: pass @@ -662,7 +647,6 @@ def _persist_system_prompt(agent, failure_message: str, *, persist_tools: bool = agent._session_db.update_system_prompt(agent.session_id, agent._cached_system_prompt) if persist_tools: from tools.mcp_tool import persist_agent_tool_names - persist_agent_tool_names(agent) except Exception as exc: logger.warning(failure_message, agent.session_id, exc) @@ -703,7 +687,6 @@ def _restore_or_build_system_prompt(agent, system_message, conversation_history) # dir; a capability refresh must rebuild THROUGH it or new skills are lost. try: from agent.prompt_builder import clear_skills_system_prompt_cache - clear_skills_system_prompt_cache(clear_snapshot=True) except Exception: pass @@ -725,7 +708,6 @@ def _restore_or_build_system_prompt(agent, system_message, conversation_history) saved_tools = session_row.get("tool_names") if session_row else None if saved_tools: from tools.mcp_tool import restore_agent_tool_prefix - restore_agent_tool_prefix(agent, json.loads(saved_tools)) except Exception: logger.debug("tool prefix restore skipped", exc_info=True) @@ -735,7 +717,6 @@ def _restore_or_build_system_prompt(agent, system_message, conversation_history) # fresh-per-turn gateway agents fall back to the single-breakpoint layout # (reconstruct_static_prefix gates on _use_prompt_caching, fails open to legacy). from agent.system_prompt import reconstruct_static_prefix, restore_plugin_prompt_sections - restore_plugin_prompt_sections(agent, stored_prompt) reconstruct_static_prefix(agent, system_message=system_message) return @@ -774,7 +755,6 @@ def _restore_or_build_system_prompt(agent, system_message, conversation_history) # session open, so this is idempotent (skips when _credits_state exists). Fail-open. try: from agent.credits_tracker import seed_credits_at_session_start - seed_credits_at_session_start(agent) except Exception: logger.debug("cold-start credits seed failed (fail-open)", exc_info=True) @@ -936,10 +916,7 @@ def _clone_message_for_send(msg): for k, v in msg.items() } if isinstance(msg, list): - return [ - _clone_message_for_send(v) if isinstance(v, (dict, list)) else v - for v in msg - ] + return [_clone_message_for_send(v) if isinstance(v, (dict, list)) else v for v in msg] return msg @@ -1091,9 +1068,7 @@ def _rewrite_system_content_blocks(system_message: dict, effective: str) -> bool content = system_message.get("content") if not isinstance(content, list) or not content: return False - if not all( - isinstance(part, dict) and part.get("type") == "text" for part in content - ): + if not all(isinstance(part, dict) and part.get("type") == "text" for part in content): return False if len(content) == 1: content[0]["text"] = effective @@ -1139,7 +1114,6 @@ def _ensure_cached_system_prompt_static(agent, system_message=None) -> None: 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(agent, system_message=system_message, log_label="failover redecoration") @@ -1149,7 +1123,6 @@ def _peel_moa_guidance( ) -> List[Dict[str, Any]]: """Remove MoA reference guidance attached by ``_attach_reference_guidance``.""" from agent.moa_loop import peel_reference_guidance - return peel_reference_guidance(messages, guidance) @@ -1173,7 +1146,6 @@ def _redecorate_prompt_cache_for_provider( planned_tools = strip_anthropic_tool_cache_control( tools_for_api if tools_for_api is not None else getattr(agent, "tools", []) ) - if prepared is not None and getattr(agent, "provider", None) == "moa": # Prepared MoA state is canonical: the synchronous acting-aggregator # sender owns its destination-local cache plan after it resolves the slot. @@ -1191,7 +1163,6 @@ def _redecorate_prompt_cache_for_provider( agent, "_direct_native_anthropic_tool_cache_capability", lambda: False )() from agent.prompt_caching import envelope_tool_part_cache_markers_supported - plan = build_prompt_cache_plan( messages, planned_tools, @@ -1229,7 +1200,6 @@ def _engine_overrides_hook(engine: Any, name: str) -> bool: return False try: from agent.context_engine import ContextEngine as _CE - return getattr(hook, "__func__", None) is not getattr(_CE, name) except Exception: return True @@ -1318,7 +1288,6 @@ def _decode_inline_moa_turn(user_message, persist_user_message): moa_config, persist_user_message)``, unchanged with ``moa_config=None`` otherwise.""" try: from hermes_cli.moa_config import decode_moa_turn - _decoded_message, _decoded_moa_config = decode_moa_turn(user_message) if _decoded_moa_config is not None: if persist_user_message is None: @@ -1338,7 +1307,6 @@ def _preflight_timeout_result(agent, exc, conversation_history) -> Dict[str, Any # 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. @@ -1460,6 +1428,38 @@ def _run_phase(fn, agent, state: _LoopState, **extra): return verdict +def _run_api_retry_loop(agent, s: _LoopState) -> Optional[Dict[str, Any]]: + """One API call with its retry/recovery loop (guard → build → call → check, error handlers). + + Returns a turn result dict when a phase ends the turn, else None once the loop is left + (success, a restart armed on ``s._retry``, interrupt, or retries exhausted).""" + while s.retry_count < s.max_retries: + _ng = _run_phase(nous_rate_limit_guard, agent, s) + if _ng.action == "return": + return _ng.result + if _ng.action == "break": + return None + try: + _run_phase(build_api_request, agent, s) + if _run_phase(perform_api_call, agent, s).action == "break": + return None + _rc = _run_phase(check_api_response, agent, s) + if _rc.action == "return": + return _rc.result + if _rc.action == "break": + return None + except InterruptedError: + if _run_phase(handle_api_interrupt, agent, s).action == "break": + return None + except Exception as api_error: + _ae = _run_phase(handle_api_error, agent, s, api_error=api_error) + if _ae.action == "return": + return _ae.result + if _ae.action == "break": + return None + return None + + def run_conversation( agent, user_message: Any, @@ -1553,7 +1553,6 @@ def run_conversation( _preflight_compression_blocked=_ctx.preflight_compression_blocked, max_compression_attempts=getattr(agent, "max_compression_attempts", 3), ) - # Opt-in runtime: api_mode == codex_app_server hands the whole turn to the codex # app-server subprocess (see agent/transports/codex_app_server_session.py). if agent.api_mode == "codex_app_server": @@ -1587,35 +1586,9 @@ def run_conversation( s.api_request_id = f"{s.turn_id}:api:{s.api_call_count}" agent._current_api_request_id = s.api_request_id - while s.retry_count < s.max_retries: - _ng = _run_phase(nous_rate_limit_guard, agent, s) - if _ng.action == "return": - return _ng.result - if _ng.action == "break": - break - - try: - _run_phase(build_api_request, agent, s) - if _run_phase(perform_api_call, agent, s).action == "break": - break - _rc = _run_phase(check_api_response, agent, s) - if _rc.action == "return": - return _rc.result - if _rc.action == "break": - break - if _rc.action == "continue": - continue - except InterruptedError: - if _run_phase(handle_api_interrupt, agent, s).action == "break": - break - except Exception as api_error: - _ae = _run_phase(handle_api_error, agent, s, api_error=api_error) - if _ae.action == "return": - return _ae.result - if _ae.action == "break": - break - if _ae.action == "continue": - continue + early_result = _run_api_retry_loop(agent, s) + if early_result is not None: + return early_result _rs = _run_phase(apply_retry_restarts, agent, s) if _rs.action == "break": @@ -1629,22 +1602,15 @@ def run_conversation( return _ri.result if _ri.action == "continue": continue - if s.assistant_message.tool_calls: - _tr = _run_phase(run_tool_round, agent, s) - if _tr.action == "return": - return _tr.result - if _tr.action == "break": - break - if _tr.action == "continue": - continue - else: - _fr = _run_phase(finish_text_response, agent, s) - if _fr.action == "return": - return _fr.result - if _fr.action == "break": - break - if _fr.action == "continue": - continue + _v = _run_phase( + run_tool_round if s.assistant_message.tool_calls else finish_text_response, agent, s + ) + if _v.action == "return": + return _v.result + if _v.action == "break": + break + if _v.action == "continue": + continue except Exception as e: if _run_phase(handle_outer_loop_error, agent, s, e=e).action == "break": break diff --git a/agent/deadline.py b/agent/deadline.py index fc9d468724..33453a746c 100644 --- a/agent/deadline.py +++ b/agent/deadline.py @@ -121,7 +121,6 @@ def _timeouts_section() -> dict: """Read the ``timeouts:`` root section from config.yaml (read-only, fail-open).""" try: from hermes_cli.config import load_config_readonly - section = load_config_readonly().get("timeouts") return section if isinstance(section, dict) else {} except Exception: @@ -353,7 +352,6 @@ def kill_process_tree(pid: int, *, sig: Optional[int] = None) -> bool: if sys.platform == "win32": try: from hermes_cli._subprocess_compat import windows_hide_flags - creationflags = windows_hide_flags() except Exception: creationflags = 0 @@ -369,13 +367,11 @@ def kill_process_tree(pid: int, *, sig: Optional[int] = None) -> bool: return False import signal as _signal - if sig is None: sig = _signal.SIGKILL try: import psutil - descendants = psutil.Process(int(pid)).children(recursive=True) except Exception: # Already gone, or psutil unavailable — the group signal still covers same-session descendants. diff --git a/agent/empty_response_guard.py b/agent/empty_response_guard.py index 492dfcb765..7e7e66ab5e 100644 --- a/agent/empty_response_guard.py +++ b/agent/empty_response_guard.py @@ -106,7 +106,6 @@ def _normalized_usage(agent: Any, response: Any, what: str) -> Any: return None try: from agent.usage_pricing import normalize_usage - return normalize_usage(raw_usage, provider=getattr(agent, "provider", None), api_mode=getattr(agent, "api_mode", None)) except Exception: # noqa: BLE001 — pricing must never break the loop @@ -121,7 +120,6 @@ def _estimate_attempt_cost(agent: Any, response: Any) -> Optional[Decimal]: return None try: from agent.usage_pricing import estimate_usage_cost - result = estimate_usage_cost( getattr(agent, "model", "") or "", canonical, provider=getattr(agent, "provider", None), base_url=getattr(agent, "base_url", None), api_key=getattr(agent, "api_key", None), diff --git a/tests/agent/test_surrogate_chokepoints.py b/tests/agent/test_surrogate_chokepoints.py index 979ede7fe5..24b87822df 100644 --- a/tests/agent/test_surrogate_chokepoints.py +++ b/tests/agent/test_surrogate_chokepoints.py @@ -185,7 +185,7 @@ def test_conversation_loop_sanitizes_api_kwargs_after_build(): build_idx = src.index("api_kwargs = agent._build_api_kwargs(api_messages)") sanitize_idx = src.index("_sanitize_structure_surrogates(api_kwargs)") assert build_idx < sanitize_idx - loop_src = inspect.getsource(cl.run_conversation) + loop_src = inspect.getsource(cl._run_api_retry_loop) assert loop_src.index("build_api_request,") < loop_src.index("perform_api_call,")