"""Recovery-branch handlers for the conversation turn's inner retry loop. When the model call raises, one-shot recovery chains run before the generic retry/backoff path. Handlers return ``True`` (request repaired in place; loop ``continue``s with the same ``retry_count``) or ``False`` (fall through). Guards live on ``TurnRetryState``; handlers mutate ``agent`` / ``messages`` / ``api_messages`` in place. Logger name stays ``agent.conversation_loop`` (caplog pins); that module is only imported lazily (cycle + patch sites). """ from __future__ import annotations import logging import locale import math import re import time from dataclasses import dataclass from typing import Any, Dict, List, Optional, Tuple from agent.conversation_compression import COMPRESSION_RETRY_CONTEXT_REDUCED_STATUS_TEMPLATE from agent.fast_mode import fast_mode_unprovisioned, mark_fast_mode_unavailable from agent.model_metadata import is_output_cap_error, parse_available_output_tokens_from_error from agent.retry_utils import is_zai_coding_overload_error, zai_coding_overload_retry_ceiling from agent.error_classifier import FailoverReason, classify_api_error from agent.message_sanitization import ( _looks_like_corrupt_image_rejection, _looks_like_image_content_rejection, _sanitize_messages_non_ascii, _sanitize_messages_surrogates, _sanitize_structure_non_ascii, _sanitize_structure_surrogates, _strip_images_from_messages, _strip_non_ascii, close_interrupted_tool_sequence, ) from agent.thinking_timeout_guidance import build_thinking_timeout_guidance, is_thinking_timeout from agent.vision_message_prep import _provider_model_key from agent.turn_failure_copy import ( CONTENT_POLICY_NEXT_STEPS, content_policy_copy, exhausted_copy, limit_reset_copy, nonretryable_copy, provider_label_for, site_copy, stamp_failure, ) from agent.turn_retry_state import TurnRetryState from hermes_constants import display_hermes_home from utils import base_url_host_matches logger = logging.getLogger("agent.conversation_loop") def _runtime_uses_ascii_encoding() -> bool: """Return whether the process genuinely needs an ASCII-only request fallback.""" encoding = locale.getpreferredencoding(False).strip().lower().replace("_", "-") return encoding in {"ascii", "us-ascii", "ansi-x3.4-1968"} def _vlines(agent: Any, *lines: str) -> None: """Force-``_vprint`` each line prefixed with ``agent.log_prefix``.""" for line in lines: agent._vprint(f"{agent.log_prefix}{line}", force=True, diagnostic=True) def _plines(agent: Any, *lines: str) -> None: """``print`` each line prefixed with ``agent.log_prefix``.""" from gateway.warning_notifications import render_notification render_notification( lambda: [print(f"{agent.log_prefix}{line}") for line in lines], platform=getattr(agent, "_notification_platform", getattr(agent, "platform", "cli")), user_config=getattr(agent, "_notification_config", None)) def _blines(agent: Any, *lines: str) -> None: """``_buffer_vprint`` each line (surfaces only if every retry+fallback exhausts).""" for line in lines: agent._buffer_vprint(line) def _image_error_max_dimension(error: Exception) -> Optional[int]: """Extract a provider-reported image dimension ceiling, if present.""" parts = [] for value in (error, getattr(error, "message", None), getattr(error, "body", None)): if value: try: parts.append(str(value)) except Exception: pass text = " ".join(parts).lower() # OpenAI Codex Responses reports a tile-patch budget (ceil(w/32)×ceil(h/32)) # instead of a pixel ceiling. A square image is the worst case for the budget, # so a per-side cap of isqrt(limit)*32 px keeps isqrt(limit)² ≤ limit — for the # 30000-patch ceiling that is 5536 px. Without this the caller falls back to # 8000 px and a 6000 px image that already exceeds the budget is skipped (#106337). if "patches after processing" in text: match = re.search(r"exceeding the limit of\s*(\d{2,7})", text) if not match: return None max_dimension = math.isqrt(int(match.group(1))) * 32 return max_dimension if 512 <= max_dimension <= 8000 else None if "image" not in text or "dimension" not in text or "max allowed size" not in text: return None match = re.search(r"max allowed size(?:\s+for [^:]+)?:\s*(\d{3,5})\s*pixels?", text) if not match: return None try: max_dimension = int(match.group(1)) except ValueError: return None return max_dimension if 512 <= max_dimension <= 8000 else None def _try_refresh_nous_paid_entitlement_credentials(agent) -> bool: """Refresh Nous runtime credentials after a fresh paid-entitlement check.""" try: from hermes_cli.nous_account import get_nous_portal_account_info if get_nous_portal_account_info(force_fresh=True).paid_service_access is not True: return False return agent._try_refresh_nous_client_credentials(force=True) except Exception: return False def _repair_transport_credentials(agent: Any) -> bool: """Strip non-ASCII from ``_client_kwargs["default_headers"]`` and the API key. Non-ASCII in the key makes httpx fail encoding the Authorization header — the usual persistent cause of UnicodeEncodeError that survives message/tool sanitization (#6843, e.g. ʋ instead of v from a bad copy-paste). Entra ID bearer providers are callables minting ASCII JWTs; skip them (``_strip_non_ascii`` would crash). Returns True when either the headers or the key were repaired. """ _client_kwargs = getattr(agent, "_client_kwargs", None) _default_headers = _client_kwargs.get("default_headers") if isinstance(_client_kwargs, dict) else None _repaired = bool(isinstance(_default_headers, dict) and _sanitize_structure_non_ascii(_default_headers)) _raw_key = getattr(agent, "api_key", None) or "" if isinstance(_raw_key, str) and _raw_key: _clean_key = _strip_non_ascii(_raw_key) if _clean_key != _raw_key: agent.api_key = _clean_key if isinstance(_client_kwargs, dict): _client_kwargs["api_key"] = _clean_key # The live client reads its own api_key copy on every request. if getattr(agent, "client", None) is not None and hasattr(agent.client, "api_key"): agent.client.api_key = _clean_key _repaired = True _vlines( agent, "⚠️ API key contained non-ASCII characters (bad copy-paste?) — stripped them. " "If auth fails, re-copy the key from your provider's dashboard.", ) return _repaired def _recover_unicode_encode_error( agent: Any, api_error: Exception, messages: List[Dict[str, Any]], api_messages: Any, api_kwargs: Any, active_system_prompt: Any, ) -> Tuple[bool, Any]: """UnicodeEncodeError recovery: lone surrogates (clipboard paste) first, then an ASCII codec under a non-UTF-8 locale. Sanitizes in place; bounded by the caller's ``_unicode_sanitization_passes < 2`` guard (surrogate strip, then ASCII-only).""" _err_str = str(api_error).lower() _is_ascii_codec = "'ascii'" in _err_str or "ascii" in _err_str # utf-8 refusing U+D800..U+DFFF ("surrogates not allowed"). _is_surrogate_error = "surrogate" in _err_str or ("'utf-8'" in _err_str and not _is_ascii_codec) # Sanitize canonical messages for surrogate recovery, but keep ASCII recovery # request-local: API copies may carry fields absent from the durable transcript. _surrogates_found = _sanitize_messages_surrogates(messages) _surrogates_found |= isinstance(api_messages, list) and _sanitize_messages_surrogates(api_messages) _surrogates_found |= isinstance(api_kwargs, dict) and _sanitize_structure_surrogates(api_kwargs) # Gate the retry on the error type, not on whether anything was found — a new # transformed field could slip through. if _surrogates_found or _is_surrogate_error: if _surrogates_found: # In-place rewrites may have popped _DB_PERSISTED_MARKER off stamped live dicts; # force a full flush scan so the repaired rows are rewritten. agent._db_flush_scan_prefix = None agent._unicode_sanitization_passes += 1 agent._buffer_vprint( "⚠️ Stripped invalid surrogate characters from messages. Retrying..." if _surrogates_found else "⚠️ Surrogate encoding error — retrying after full-payload sanitization..." ) return True, active_system_prompt if not _is_ascii_codec: return False, active_system_prompt # Error text is provider-controlled and can mention ``ascii`` even when the # process sends UTF-8. In that normal case, do not rewrite conversation, # tools, prompts, or prefill; only repair values that can poison an ASCII # transport header. If nothing was repaired, an identical retry cannot # succeed — return False so the error surfaces through the normal path # instead of burning both sanitization passes on unchanged requests. if not _runtime_uses_ascii_encoding(): if not _repair_transport_credentials(agent): return False, active_system_prompt agent._unicode_sanitization_passes += 1 _vlines( agent, "⚠️ Repaired non-ASCII request credentials/headers without changing conversation content. Retrying...", ) return True, active_system_prompt agent._force_ascii_payload = True # Strip all non-ASCII from the request-local api_messages (reused across retries). The # failed attempt's api_kwargs is NOT touched: build_api_request rebuilds it from # ``agent.tools`` on the next iteration and ``sanitize_outbound_kwargs`` strips the whole # payload under ``_force_ascii_payload``. Canonical agent state stays byte-stable. _messages_sanitized = isinstance(api_messages, list) and _sanitize_messages_non_ascii(api_messages) _system_sanitized = False if isinstance(active_system_prompt, str): _sanitized_system = _strip_non_ascii(active_system_prompt) if _sanitized_system != active_system_prompt: active_system_prompt = _sanitized_system _system_sanitized = True _transport_repaired = _repair_transport_credentials(agent) # Always retry on ASCII codec detection: _force_ascii_payload sanitizes the full # api_kwargs next iteration even when the checks above find nothing. agent._unicode_sanitization_passes += 1 _vlines( agent, "⚠️ System encoding is ASCII — stripped non-ASCII characters from request payload. Retrying..." if (_messages_sanitized or _system_sanitized or _transport_repaired) else "⚠️ System encoding is ASCII — enabling full-payload sanitization for retry...", ) return True, active_system_prompt def _strip_request_images_and_retry(agent: Any, api_messages: Any) -> bool: """Strip image parts from the per-call ``api_messages`` copy; True if anything was removed. Shared by the corrupt-image recoveries: a bad payload says nothing about the model, so it is stripped for this attempt only and the model is never recorded as image-rejecting.""" if isinstance(api_messages, list) and _strip_images_from_messages(api_messages): _vlines(agent, "⚠️ Provider rejected a corrupted image — stripped images from the retry payload and retrying...") return True return False def recover_before_classification( agent: Any, api_error: Exception, *, messages: List[Dict[str, Any]], api_messages: Any, api_kwargs: Any, active_system_prompt: Any, ) -> Tuple[bool, Any]: """Recovery branches that run BEFORE ``classify_api_error``: UnicodeEncodeError sanitization, Anthropic fast mode with no capacity (drop ``speed`` for that model), provider image-content rejection (record the (provider, model); build_api_request strips images from that model's requests only), and the Bedrock AnthropicBedrock SDK streaming fallback. Returns ``(retry_now, active_system_prompt)``; the prompt may be ASCII-sanitized in place.""" if isinstance(api_error, UnicodeEncodeError) and getattr(agent, '_unicode_sanitization_passes', 0) < 2: _recovered, active_system_prompt = _recover_unicode_encode_error( agent, api_error, messages, api_messages, api_kwargs, active_system_prompt ) if _recovered: return True, active_system_prompt # Anthropic fast mode with no capacity: a 429 whose fast-mode limit header is 0 can never # succeed at fast speed, and it says nothing about the key's standard-speed limits. Stop # sending ``speed`` to this model and retry now, before credential rotation benches the key. if fast_mode_unprovisioned(api_error, api_kwargs) and mark_fast_mode_unavailable(agent): _vlines(agent, f"⚠️ Fast mode isn't available for {agent.model} on this Anthropic organization — using standard speed for this session, retrying...") logger.warning("%sFast mode: %s has a fast-mode limit of 0; standard speed for this session", agent.log_prefix, agent.model) return True, active_system_prompt # Some providers 4xx on image_url content: record the (provider, model) and retry; # build_api_request strips images from that model's requests only. English phrase # match; extend it. _err_body = "" try: _err_body = str(getattr(api_error, "body", None) or getattr(api_error, "message", None) or str(api_error)) except Exception: pass _err_status = getattr(api_error, "status_code", None) # 4xx-only gate: 5xx/timeouts are transient and take the retry path. _status_ok = _err_status is None or (400 <= int(_err_status) < 500) # Guarded PER MODEL, not by a turn-global flag: in a fallback chain the next model can reject # images too, and a turn-wide flag would skip its recovery and fail the turn. _model_key = _provider_model_key(agent) _rejected = agent._image_rejecting_models _corrupt = _looks_like_corrupt_image_rejection(_err_body) if _status_ok and (_corrupt or (_model_key not in _rejected and _looks_like_image_content_rejection(_err_body))): # Send-path only. A rejection says what THIS model accepts, not what the conversation # holds: stripping ``messages`` (canonical history) and forcing a flush deleted every # image — and every image-only message — from state.db for good, so a later switch to a # vision model found them gone. Same failure as the ASCII strip in #117802. if _corrupt: # A bad payload says nothing about the model's capability: strip this attempt only # (like the image_corrupt branch below) and leave the model unmarked so a later good # image still reaches it. Retry only if something was stripped, or a text-only # request would loop on the same error. if _strip_request_images_and_retry(agent, api_messages): return True, active_system_prompt else: # Record the model; the retry re-enters build_api_request with the same # api_messages and strip_images_for_rejecting_model strips them there. _rejected.add(_model_key) _vlines( agent, "⚠️ Server rejected image content — sending text only to this model; " "images stay in the session history.", ) return True, active_system_prompt # AnthropicBedrock SDK raises "Unexpected event order" when Bedrock errors before # message_start; fall back to native Converse for this session. if ( isinstance(api_error, RuntimeError) and "unexpected event order" in str(api_error).lower() and getattr(agent, "provider", "") == "bedrock" and agent.api_mode == "anthropic_messages" and not getattr(agent, "_bedrock_converse_fallback_attempted", False) ): agent._bedrock_converse_fallback_attempted = True agent.api_mode = "bedrock_converse" agent._bedrock_region = getattr(agent, "_bedrock_region", None) or "us-east-1" agent.client = None # Drop the AnthropicBedrock client agent._client_kwargs = {} _vlines(agent, "⚠️ AnthropicBedrock SDK streaming failed — falling back to native Converse API for this session.") return True, active_system_prompt return False, active_system_prompt def _print_nous_401_diagnostics(agent: Any, api_error: Exception) -> None: """Nous 401 that survived a credential refresh: likely Portal OAuth expired/revoked, no credits, or agent key blocked.""" from agent.conversation_loop import _print_nous_entitlement_guidance from hermes_constants import display_hermes_home _body_text = "" try: _body = getattr(api_error, "body", None) or getattr(api_error, "response", None) if _body is not None: _body_text = str(_body)[:200] except Exception: pass _plines(agent, "🔐 Nous 401 — Portal authentication failed.") if _body_text: _plines(agent, f" Response: {_body_text}") try: from hermes_cli.anon_auth import is_anonymous_agent if is_anonymous_agent(agent): # The free tier has no credits, no agent key and no auth.json to inspect: its session # ended and could not be replaced. The two doors are a sign-in or another provider. _plines(agent, " Your session ended and Hermes couldn't start a new one.", " Sign in with a Nous account (it's free), or switch providers with /model.") return except Exception: pass if not _print_nous_entitlement_guidance(agent, "Nous model access"): _plines(agent, " Most likely: Portal OAuth expired, account out of credits, or agent key revoked.") _plines( agent, " Troubleshooting:", " • Re-authenticate: hermes auth add nous", " • Check credits / billing: https://portal.nousresearch.com", f" • Verify stored credentials: {display_hermes_home()}/auth.json", " • Switch providers temporarily: /model --provider openrouter", ) def _print_anthropic_401_diagnostics(agent: Any, key: Any) -> None: """Anthropic 401 that survived a credential refresh: show auth method + fixes.""" from agent.anthropic_credentials import _is_oauth_token from agent.azure_identity_adapter import is_token_provider from hermes_constants import display_hermes_home _plines(agent, "🔐 Anthropic 401 — authentication failed.") if is_token_provider(key): # Azure Foundry Entra ID: JWT minted per-request by an httpx hook; 401 = Azure # rejected it (RBAC, az login, IMDS). _plines( agent, " Auth method: Microsoft Entra ID (httpx event hook)", " Run `hermes doctor` for credential-chain diagnostics, or", " `az login` if your developer session expired.", ) else: auth_method = "Bearer (OAuth/setup-token)" if _is_oauth_token(key) else "x-api-key (API key)" _plines( agent, f" Auth method: {auth_method}", f" Token prefix: {key[:12]}..." if isinstance(key, str) and len(key) > 12 else " Token: (empty or short)", ) _dhh = display_hermes_home() _plines( agent, " Troubleshooting:", f" • Check ANTHROPIC_TOKEN in {_dhh}/.env for Hermes-managed OAuth/setup tokens", f" • Check ANTHROPIC_API_KEY in {_dhh}/.env for API keys or legacy token values", " • For API keys: verify at https://platform.claude.com/settings/keys", " • Hermes login (OAuth): run 'hermes auth add anthropic' to sign in again, then retry", " • Inspect what Hermes holds: hermes auth list anthropic", " • Legacy cleanup: hermes config set ANTHROPIC_TOKEN \"\"", " • Clear stale keys: hermes config set ANTHROPIC_API_KEY \"\"", ) def _refresh_credentials_after_401( agent: Any, api_error: Exception, _retry: TurnRetryState, status_code: Optional[int] ) -> bool: """Per-provider one-shot credential refresh on 401 (codex/xai, vertex, nous, copilot, anthropic), printing user-facing diagnostics when the nous/anthropic refresh fails. Returns True when a refresh succeeded and the call should be retried.""" from agent.conversation_loop import _is_copilot_provider if status_code != 401: return False if ( agent.api_mode == "codex_responses" and agent.provider in {"openai-codex", "xai-oauth"} and not _retry.codex_auth_retry_attempted ): _retry.codex_auth_retry_attempted = True if agent._try_refresh_codex_client_credentials(force=True): _label = "xAI OAuth" if agent.provider == "xai-oauth" else "Codex" agent._buffer_vprint(f"🔐 {_label} auth refreshed after 401. Retrying request...") return True if agent.api_mode == "chat_completions" and agent.provider == "vertex" and not _retry.vertex_auth_retry_attempted: _retry.vertex_auth_retry_attempted = True if agent._try_refresh_vertex_client_credentials(): agent._buffer_vprint("🔐 Vertex AI token refreshed after 401. Retrying request...") return True if ( agent.api_mode in ("chat_completions", "anthropic_messages") and agent.provider == "nous" and not _retry.nous_auth_retry_attempted ): _retry.nous_auth_retry_attempted = True if agent._try_refresh_nous_client_credentials(force=True): agent._buffer_vprint("🔐 Nous agent key refreshed after 401. Retrying request...") return True _print_nous_401_diagnostics(agent, api_error) if _is_copilot_provider(agent) and not _retry.copilot_auth_retry_attempted: _retry.copilot_auth_retry_attempted = True if agent._try_refresh_copilot_client_credentials(): agent._buffer_vprint("🔐 Copilot credentials refreshed after 401. Retrying request...") return True if ( agent.api_mode == "anthropic_messages" and hasattr(agent, '_anthropic_api_key') and not _retry.anthropic_auth_retry_attempted ): _retry.anthropic_auth_retry_attempted = True if agent._try_refresh_anthropic_client_credentials(): _plines(agent, "🔐 Anthropic credentials refreshed after 401. Retrying request...") return True _print_anthropic_401_diagnostics(agent, agent._anthropic_api_key) return False def _is_codex_token_expired(agent: Any, api_error: Exception) -> bool: """401 ``token_expired`` from the Codex backend (#88510). It rejects a stale replayed ``encrypted_content`` blob with this auth signature, so a persisted session loops on "sign in again" while a fresh session on the same bearer works. The caller treats it like ``invalid_encrypted_content`` — but only while cached reasoning items remain to strip.""" if getattr(api_error, "status_code", None) != 401: return False reason = agent._extract_api_error_context(api_error).get("reason") return isinstance(reason, str) and reason.strip().lower() == "token_expired" def reset_codex_reasoning_replay(agent: Any) -> None: """The replay verdict belongs to the route that earned it: a ``/model`` switch, fallback activation or primary restore starts the new route with replay on (#61552).""" agent._codex_reasoning_replay_enabled = True agent._codex_reasoning_replay_rejected = False def _recover_stale_codex_reasoning( agent: Any, _retry: TurnRetryState, messages: List[Dict[str, Any]], api_messages: Any, ) -> bool: """Stale ``codex_reasoning_items`` blob rejected by the provider: strip cached items (mutates persisted ``messages``) and retry once. The first rejection keeps replay on, since blobs the route mints from now on are sealed with its current key; a repeat rejection means the route cannot round-trip its own blobs, so replay is disabled for the session.""" if ( _retry.invalid_encrypted_content_retry_attempted or agent.api_mode != "codex_responses" or not bool(getattr(agent, "_codex_reasoning_replay_enabled", True)) or not any( isinstance(_m, dict) and _m.get("role") == "assistant" and isinstance(_m.get("codex_reasoning_items"), list) and _m.get("codex_reasoning_items") for _m in messages ) ): return False _retry.invalid_encrypted_content_retry_attempted = True keep_replay = not getattr(agent, "_codex_reasoning_replay_rejected", False) agent._codex_reasoning_replay_rejected = True replay_stats = agent._disable_codex_reasoning_replay(messages, keep_replay=keep_replay) # The retry is rebuilt from the request copy; with replay kept on it would resend the stale blob. for _m in api_messages if isinstance(api_messages, list) else []: if isinstance(_m, dict): _m.pop("codex_reasoning_items", None) action = "stripped stale" if keep_replay else "disabled replay for this session and stripped" _vlines( agent, f"⚠️ Encrypted reasoning replay was rejected by the provider — " f"{action} {replay_stats['items']} item(s) from {replay_stats['messages']} message(s), retrying...", ) logger.warning( "%sInvalid encrypted reasoning recovery: %s %d items from %d messages", agent.log_prefix, action, replay_stats["items"], replay_stats["messages"], ) return True def _recover_format_errors( agent: Any, api_error: Exception, classified: Any, _retry: TurnRetryState, messages: List[Dict[str, Any]], api_messages: Any, ) -> bool: """One-shot format-recovery strips: thinking-signature → invalid-encrypted-content replay disable → native-compaction reject → llama.cpp grammar strip. Returns True when the request was repaired and should be retried.""" # Upstream mutation can invalidate a thinking signature. Native Anthropic has multiple replay # carriers and preserved historical blocks, so suppress the rejected opaque blocks across all # carriers and persist that suppression. Other transports keep their established one-request # repair: drop reasoning_details only. if classified.reason == FailoverReason.thinking_signature and not _retry.thinking_sig_retry_attempted: _retry.thinking_sig_retry_attempted = True from agent.anthropic_thinking_replay import remember_rejected_thinking, tracks_rejected_thinking if tracks_rejected_thinking(agent): removed = remember_rejected_thinking(agent, api_messages) detail = "suppressed rejected Anthropic replay blocks" else: removed = 0 for message in api_messages: if isinstance(message, dict) and "reasoning_details" in message: message.pop("reasoning_details", None) removed += 1 detail = "stripped reasoning_details" _vlines(agent, f"⚠️ Thinking block signature invalid, {detail} and retrying...") logger.warning( "%sThinking block signature recovery: %s from %d carrier/message(s) " "(canonical messages unchanged)", agent.log_prefix, detail, removed, ) return True # 400 ``invalid_encrypted_content`` on a stale ``codex_reasoning_items`` blob (the 401 # ``token_expired`` twin is taken ahead of the credential pool in the caller). if classified.reason == FailoverReason.invalid_encrypted_content and _recover_stale_codex_reasoning( agent, _retry, messages, api_messages ): return True # Structured 400 naming ``context_management``: disable native compaction for the # session, retry once; local compression takes over. if ( agent.api_mode == "codex_responses" and not _retry.native_compaction_reject_retry_attempted and bool(getattr(agent, "codex_responses_native_compaction", False)) ): from agent.native_compaction import is_native_compaction_rejection if is_native_compaction_rejection(api_error, getattr(api_error, "status_code", None)): _retry.native_compaction_reject_retry_attempted = True agent.codex_responses_native_compaction = False _vlines( agent, "⚠️ Provider rejected native compaction (context_management) — disabled for this session, " "local compression stays active. Retrying...", ) logger.warning( "%sNative compaction rejection recovery: disabled " "codex_responses_native for this session and retrying", agent.log_prefix, ) return True # llama.cpp ``json-schema-to-grammar`` rejects regex escapes and most ``format`` # values: strip ``pattern``/``format`` from ``agent.tools``, retry once. if classified.reason == FailoverReason.llama_cpp_grammar_pattern and not _retry.llama_cpp_grammar_retry_attempted: _retry.llama_cpp_grammar_retry_attempted = True try: from tools.schema_sanitizer import strip_pattern_and_format _, _stripped = strip_pattern_and_format(agent.tools) except Exception as _strip_exc: # pragma: no cover — defensive logger.warning("%sllama.cpp grammar recovery: strip helper failed: %s", agent.log_prefix, _strip_exc) _stripped = 0 if _stripped: _vlines(agent, f"⚠️ llama.cpp rejected tool schema grammar — stripped {_stripped} pattern/format keyword(s), retrying...") logger.warning( "%sllama.cpp grammar recovery: stripped %d " "pattern/format keyword(s) from tool schemas", agent.log_prefix, _stripped, ) return True # Nothing to strip — fall through to normal retry rather than loop on the same error. logger.warning( "%sllama.cpp grammar error but no pattern/format " "keywords to strip — falling through to normal retry", agent.log_prefix, ) return False _WELCOME_ROUTE_HEAL_COPY = { "anon_on_paid_host": "Reconnected to the free model's own route.", "named_on_welcome_host": "Reconnected to your Nous account's own route.", } def _recover_welcome_tier(agent: Any, classified: Any, _retry: TurnRetryState) -> bool: """Two one-shot repairs for the Nous free tier, both silent on the wire and named once in chat. ``model_not_free``: the session asked the welcome host for a model it does not serve; move to the first alternate the gateway named (its own model) and retry, instead of failing the turn. ``anon_on_paid_host`` / ``named_on_welcome_host``: this process is pointed at the other identity's host (a stale route); re-read the credentials, which heals the URL, and retry. The refresh reports False when the store yields the same route, so a user-set ``NOUS_INFERENCE_BASE_URL`` falls straight through to the terminal copy. Reads the CLASSIFIER's context (``classified.error_context``): that is where ``_nous_welcome_tier`` parks ``welcome_refusal`` / ``welcome_route``. The turn's other context (``extract_api_error_context``) never carries them.""" ctx = getattr(classified, "error_context", None) or {} refusal = ctx.get("welcome_refusal") if isinstance(ctx, dict) else None if isinstance(refusal, dict) and refusal.get("reason") == "model_not_free" and not _retry.welcome_model_switch_attempted: _retry.welcome_model_switch_attempted = True alternates = [a for a in (refusal.get("alternates") or []) if isinstance(a, str) and a] requested = str(getattr(agent, "model", "") or "") target = alternates[0] if alternates else None if target and target != requested: try: agent.model = target agent._nous_model_switch = (requested, target) except Exception: return False _vlines(agent, f"↪️ {requested} isn't available without signing in; using {target} for now. Retrying...") logger.info("%sNous free tier: moved %s -> %s after model_not_free", agent.log_prefix, requested, target) return True route = ctx.get("welcome_route") if isinstance(ctx, dict) else None if route in _WELCOME_ROUTE_HEAL_COPY and not _retry.welcome_route_heal_attempted: _retry.welcome_route_heal_attempted = True try: healed = bool(agent._try_refresh_nous_client_credentials(force=True)) except Exception: healed = False if healed: _vlines(agent, f"🔐 {_WELCOME_ROUTE_HEAL_COPY[route]} Retrying request...") return True return False def recover_after_classification( agent: Any, api_error: Exception, classified: Any, _retry: TurnRetryState, *, status_code: Optional[int], error_context: Any, messages: List[Dict[str, Any]], api_messages: Any, ) -> Tuple[bool, bool]: """One-shot recovery chain that runs AFTER ``classify_api_error`` and before the generic retry path. Order is load-bearing (each branch may ``return`` early): Nous paid-entitlement refresh → Codex stale-reasoning strip on 401 ``token_expired`` → credential-pool rotation → image shrink → multimodal-tool-content strip → corrupt-image strip → Anthropic OAuth 1M-beta disable → per-provider 401 credential refresh → format-recovery strips. Returns ``(retry_now, recovered_with_pool)``; the latter feeds the Nous rate-limit guard.""" from agent.conversation_loop import _is_nous_inference_route if _recover_welcome_tier(agent, classified, _retry): return True, False # 401 ``token_expired`` while the transcript still carries ``codex_reasoning_items`` is a # stale replayed blob far more often than a dead bearer (#88510): strip BEFORE the pool # refreshes/benches every healthy entry over a session-state problem. A real expiry pays # one extra round-trip and then takes the credential path below as before. if _is_codex_token_expired(agent, api_error) and _recover_stale_codex_reasoning( agent, _retry, messages, api_messages ): return True, False if ( classified.reason == FailoverReason.billing and _is_nous_inference_route( getattr(agent, "provider", "") or "", getattr(agent, "base_url", "") or "" ) and not _retry.nous_paid_entitlement_refresh_attempted ): _retry.nous_paid_entitlement_refresh_attempted = True if _try_refresh_nous_paid_entitlement_credentials(agent): _vlines(agent, "🔐 Nous paid access verified — refreshed runtime credentials and retrying request...") return True, False recovered_with_pool, _retry.has_retried_429 = agent._recover_with_credential_pool( status_code=status_code, has_retried_429=_retry.has_retried_429, classified_reason=classified.reason, error_context=error_context, billing_unverified=classified.billing_unverified, ) if recovered_with_pool: return True, recovered_with_pool # Shrink oversized native image parts in-place and retry once. if classified.reason == FailoverReason.image_too_large and not _retry.image_shrink_retry_attempted: _retry.image_shrink_retry_attempted = True if agent._try_shrink_image_parts_in_messages( api_messages, max_dimension=_image_error_max_dimension(api_error) or 8000 ): _vlines(agent, "📐 Image(s) exceeded provider size limit — shrank and retrying...") return True, recovered_with_pool logger.info( "image-shrink recovery: no data-URL image parts found " "or shrink didn't reduce size; surfacing original error." ) # Strict OpenAI-spec providers 400 on list-type tool content: strip images, mark # (provider, model) no-list-tool-content for the session, retry once. if ( classified.reason == FailoverReason.multimodal_tool_content_unsupported and not _retry.multimodal_tool_content_retry_attempted ): _retry.multimodal_tool_content_retry_attempted = True if agent._try_strip_image_parts_from_tool_messages(api_messages): _vlines(agent, "📐 Provider rejected list-type tool content — downgraded screenshots to text and retrying...") return True, recovered_with_pool logger.info( "multimodal-tool-content recovery: no list-type tool " "messages with image parts found; surfacing original error." ) # Route rejecting a reasoning disable: a reasoning-mandatory route (Nous Portal / OpenRouter, # e.g. GLM-5.3) 400s on ``reasoning: {enabled: false}``; a chat-only OpenAI-compatible relay # 400s on the ``reasoning_effort: none`` the title/continuation disable projects (#114460). # The catalog guard in the provider profile normally swallows the first, but a process that # warmed its caps cache before the route flipped keeps sending it. One-shot: never send a # disable again this session (the wire builder omits it → route default), queue a catalog # refresh so the guard is right next time (no-op for providers without a catalog), retry. if ( classified.reason == FailoverReason.reasoning_mandatory and not _retry.reasoning_mandatory_retry_attempted ): _retry.reasoning_mandatory_retry_attempted = True sent = getattr(agent, "_wire_reasoning_config", None) if isinstance(sent, dict) and sent.get("enabled") is not False and sent.get("effort") not in (None, "none"): # The rejected request carried an ENABLED config: the route refuses that reasoning # level (#100536: ``reasoning.effort: max`` on a Responses relay). Dropping a disable # would resend the identical request; omit the reasoning fields instead (route default). agent._reasoning_effort_rejected = True _vlines(agent, f"⚠️ {agent.model} rejects reasoning effort {sent['effort']} — using the route's default for this session, retrying...") logger.warning("%sReasoning-effort recovery: dropping reasoning config for %s", agent.log_prefix, agent.model) return True, recovered_with_pool agent._reasoning_disable_rejected = True # "Reasoning is mandatory ... cannot be disabled" understands the field and refuses only the # OFF: step up to the floor effort (the closest the route allows to what the user asked for) # rather than the route default. A relay that does not know the field at all keeps the # drop (a floor would 400 the same way). from agent.error_classifier import is_reasoning_required_rejection agent._reasoning_floor_required = is_reasoning_required_rejection(str(api_error)) try: from hermes_cli.models_reasoning_caps import refresh_reasoning_caps_async refresh_reasoning_caps_async(agent.provider) except Exception: pass if agent._reasoning_floor_required: from agent.auxiliary_reasoning_floor import REASONING_FLOOR_EFFORT _vlines(agent, f"⚠️ {agent.model} cannot disable reasoning — using effort={REASONING_FLOOR_EFFORT} for this session, retrying...") logger.warning("%sReasoning-disable recovery: stepping reasoning up to %s for %s", agent.log_prefix, REASONING_FLOOR_EFFORT, agent.model) else: _vlines(agent, f"⚠️ {agent.model} rejects disabling reasoning — using the route's default for this session, retrying...") logger.warning("%sReasoning-disable recovery: dropping reasoning disable for %s", agent.log_prefix, agent.model) return True, recovered_with_pool # Provider rejected the image bytes; shrinking can't help, so strip image parts. # Strip ONLY the per-call copy: replacing msg["content"] on the shallow api_messages # rows keeps canonical history's images (transient rejection must not erase history). if classified.reason == FailoverReason.image_corrupt: if _strip_request_images_and_retry(agent, api_messages): return True, recovered_with_pool logger.info("image-corrupt recovery: no image parts found to strip; surfacing original error.") # Anthropic OAuth subscription rejected the 1M-context beta: disable it for this # session, rebuild the client, retry once. Reactive so capable subscriptions keep 1M. if ( # See PR #17680 for the original report (we chose reactive recovery over the proposed unconditional # omit so capable subscriptions don't silently lose the capability). classified.reason == FailoverReason.oauth_long_context_beta_forbidden and agent.api_mode == "anthropic_messages" and agent._is_anthropic_oauth and not _retry.oauth_1m_beta_retry_attempted ): _retry.oauth_1m_beta_retry_attempted = True if not getattr(agent, "_oauth_1m_beta_disabled", False): agent._oauth_1m_beta_disabled = True try: agent._anthropic_client.close() except Exception: pass agent._rebuild_anthropic_client() _vlines(agent, "🔕 OAuth subscription doesn't support the 1M-context beta — disabled for this session and retrying...") return True, recovered_with_pool if _refresh_credentials_after_401(agent, api_error, _retry, status_code): return True, recovered_with_pool if _recover_format_errors(agent, api_error, classified, _retry, messages, api_messages): return True, recovered_with_pool return False, recovered_with_pool def _failed_turn_result(final_response: str, messages: Any, api_call_count: int, error: str) -> Dict[str, Any]: """Base failed-turn result dict shared by the two terminal paths.""" return { "final_response": final_response, "messages": messages, "api_calls": api_call_count, "completed": False, "failed": True, "error": error, } def settle_delivered_partial(agent: Any, messages: Any, current_turn_user_idx: Any) -> str: """Visible text already delivered this turn ("" when none), collapsing any continuation trail first so the terminal persist never keeps a dangling synthetic nudge (#119001). ``build_api_request`` resets ``_current_streamed_assistant_text`` per attempt, so after a mid-stream death + continuation + pre-stream error the live accumulator is empty and the fragment rows (now collapsed into one assistant row) are the only record. """ from agent.turn_truncation import collapse_continuation_trail collapsed = collapse_continuation_trail( agent, messages, current_turn_user_idx, finish_reason="error", ) live = getattr(agent, "_current_streamed_assistant_text", "") if isinstance(live, str) and live.strip(): from agent.agent_runtime_helpers import strip_think_blocks return strip_think_blocks(agent, live).strip() or collapsed return collapsed def _with_delivered_partial(final_response: str, error_summary: str, delivered: str) -> tuple: """Prepend delivered partial text to a terminal error body ("" unchanged). Returns ``(final_response, keep_partial)``; callers set ``result["partial"]`` when ``keep_partial`` so the gateway emits ``payload.partial`` and surfaces retain the bubble instead of clearing it. ``final_response`` must stay distinct from ``error`` — that inequality is the retention contract. """ _delivered = (delivered or "").strip() if not _delivered or _delivered == (error_summary or "").strip(): return final_response, False return f"{_delivered}\n\n{final_response}", True def limit_reset_epoch(agent: Any, api_error: Exception) -> Optional[float]: """Epoch seconds when the provider says its limit lifts (Retry-After header, ``resets_at`` / ``retry_after`` body fields, "try again in N" text) — the same datum the backoff honours.""" from agent.credential_pool import _parse_absolute_timestamp try: return _parse_absolute_timestamp(agent._extract_api_error_context(api_error).get("reset_at")) except Exception: # advisory only — never break the error path return None def _stamp_limit_reset(result: Dict[str, Any], agent: Any, api_error: Exception) -> None: """``failure_resets_at`` for structured clients (Desktop card: "Limit resets at HH:mm") and the same sentence appended to the chat text every plain surface (CLI/TUI/gateway) renders (#98852).""" resets_at = limit_reset_epoch(agent, api_error) if resets_at is None: return result["failure_resets_at"] = resets_at if line := limit_reset_copy(resets_at): result["final_response"] = f"{result['final_response']}\n\n{line}" def _print_nonretryable_auth_guidance( agent: Any, classified: Any, *, status_code: Optional[int], provider: Any, base_url: Any, model: Any, ) -> None: """Actionable guidance for a terminal auth / billing error.""" from agent.conversation_loop import _print_billing_or_entitlement_guidance, _print_nous_entitlement_guidance if classified.reason == FailoverReason.billing and _print_billing_or_entitlement_guidance( agent, capability="model access", provider=provider, base_url=str(base_url), model=model, unverified=classified.billing_unverified, ): return if provider == "nous" and _print_nous_entitlement_guidance(agent, "Nous model access"): return if provider in {"openai-codex", "xai-oauth", "nous"} and status_code == 401: if provider == "openai-codex": from agent.turn_failure_copy import oauth_relogin_command _vlines( agent, " 💡 Codex OAuth token was rejected (HTTP 401). Your token may have been", " refreshed by another client (Codex CLI, VS Code) or another Hermes profile.", f" Sign this profile in again: `{oauth_relogin_command(provider)}`", ) elif provider == "xai-oauth": _vlines( agent, " 💡 xAI OAuth token was rejected (HTTP 401). To fix:", " re-authenticate with xAI Grok OAuth (SuperGrok / Premium+) from `hermes model`.", ) else: # nous _vlines( agent, " 💡 Nous Portal OAuth token was rejected (HTTP 401). Your token may be", " expired, revoked, or your account may be out of credits. To fix:", " 1. Re-authenticate: hermes portal", " 2. Check your portal account: https://portal.nousresearch.com", ) # ``:free`` is OpenRouter slug syntax; Nous Portal will reject the model # name even after a successful re-auth. if isinstance(model, str) and model.endswith(":free"): _vlines( agent, f" ⚠️ Note: `{model}` looks like an OpenRouter slug (`:free` suffix).", " Nous Portal won't recognize that model name. Either switch to a", f" Nous catalog model, or run `/model openrouter:{model}` to use OpenRouter.", ) return _vlines( agent, " 💡 Your API key was rejected by the provider. Check:", " • Is the key valid? Run: hermes setup", f" • Does your account have access to {model}?", ) if base_url_host_matches(str(base_url), "openrouter.ai"): _vlines(agent, " • Check credits: https://openrouter.ai/settings/credits") def _welcome_tier_guidance(classified: Any, *, model: Any, in_chat: bool, door: bool = True) -> str: """Copy for a Nous free-tier refusal the classifier parsed (``welcome_refusal`` / ``welcome_route`` in ``error_context``); empty for every other error.""" ctx = getattr(classified, "error_context", None) or {} refusal, route = ctx.get("welcome_refusal"), ctx.get("welcome_route") if not refusal and not route: return "" from hermes_cli.anon_auth import welcome_refusal_copy, welcome_route_refusal_copy if refusal: return welcome_refusal_copy(refusal, model=str(model or ""), in_chat=in_chat, door=door) return welcome_route_refusal_copy(str(route), in_chat=in_chat, door=door) # Closed table: every card kind the desktop has copy for. An unknown gateway reason lands on # "refused" (generic card, sentence kept) rather than a code the desktop cannot key on. _WELCOME_SURFACE_KINDS = { "rate_limited": "rate_limited", "at_capacity": "at_capacity", "admission_closed": "at_capacity", "model_not_free": "model_not_free", "feature_not_free": "model_not_free", } def _welcome_surface_kind(classified: Any) -> str: """The free-tier failure kind a client renders its card from (``error_surface`` code ``free_tier_``): the welcome refusal's reason, or the route refusal; "" otherwise.""" ctx = getattr(classified, "error_context", None) or {} refusal = ctx.get("welcome_refusal") if isinstance(ctx, dict) else None if isinstance(refusal, dict): return _WELCOME_SURFACE_KINDS.get(str(refusal.get("reason") or ""), "refused") route = ctx.get("welcome_route") if isinstance(ctx, dict) else None if route == "tier_disabled": return "disabled" # A named account on the welcome host has already signed in: no sign-in card, copy only. if route == "named_on_welcome_host": return "" return "route" if route else "" def _stamp_free_tier(result: Dict[str, Any], kind: str, message: str) -> Dict[str, Any]: """Structured free-tier failure block: ``error_surface`` keys its code on ``kind`` and a client shows ``message`` (the chat sentence) as the card body instead of its own generic copy.""" result["free_tier"] = {"kind": kind or "refused", "message": message} return result def _welcome_outage_copy(base_url: Any, classified: Any, *, anonymous: bool = False) -> str: """On the Nous free tier, a transport / server failure that outlived every retry reads as one plain sentence (the free model is having trouble) rather than the technical summary. Empty for every other route and for rate limits / billing, which have their own copy.""" try: from hermes_cli.anon_auth import FREE_TIER_OUTAGE_COPY, route_is_welcome_host # Both: an anonymous JWT sent to a user-overridden paid host never reached the free model. if not anonymous or not route_is_welcome_host(base_url): return "" # Not ``unknown``: that is the classifier's catch-all for status-less local failures, which # are not the free model's trouble. if classified.reason in (FailoverReason.timeout, FailoverReason.overloaded, FailoverReason.server_error): return FREE_TIER_OUTAGE_COPY except Exception: pass return "" # Terminal status label per non-retryable reason (default names the HTTP status). _NONRETRYABLE_LABELS = { FailoverReason.content_policy_blocked: "The provider's safety filter refused this request", FailoverReason.upstream_blocked: "A firewall/CDN in front of the provider blocked this request", FailoverReason.ssl_cert_verification: "The provider's security certificate could not be verified", # Only reached after the one-shot image shrink ran (recover_after_classification sets the flag first). FailoverReason.image_too_large: "Request still exceeded the provider's size limit after shrinking images", } def _missing_vendor_prefix_suggestion(api_error: Exception, provider: Any, model: Any) -> Optional[str]: """Prefixed catalogue id when a bare 404 most likely means ``vendor/model`` lost its prefix.""" if getattr(api_error, "status_code", None) != 404: return None try: from hermes_cli.model_normalize import suggest_prefixed_model_id return suggest_prefixed_model_id(str(provider or ""), str(model or "")) except Exception: return None def nonretryable_client_error_result( agent: Any, api_error: Exception, classified: Any, *, status_code: Optional[int], api_kwargs: Any, api_messages: Any, messages: List[Dict[str, Any]], conversation_history: Any, api_call_count: int, approx_tokens: int, provider: Any, base_url: Any, model: Any, delivered: str = "", ) -> Dict[str, Any]: """Terminal path for a non-retryable 4xx once fallback is exhausted: debug dump, flush the retry trace, print auth / billing / content-policy / TLS guidance, persist (skipped for likely context-overflow 400s so the failure does not grow the session), build result.""" # Result/guidance helpers stay in the loop module (tests import + patch them there). from agent.conversation_loop import _billing_failure_result, _content_policy_blocked_result if api_kwargs is not None: agent._dump_api_request_debug(api_kwargs, reason="non_retryable_client_error", error=api_error) # Terminal — flush buffered context so the user sees what was tried before the abort. agent._flush_status_buffer() # Summarize once: Cloudflare/proxy HTML pages and raw provider bodies must be # collapsed here or they leak verbatim via the ``error`` field. _nonretryable_summary = agent._summarize_api_error(api_error) _plabel = provider_label_for(provider) _label = _NONRETRYABLE_LABELS.get(classified.reason, f"{_plabel} rejected the request and retrying won't help") agent._emit_diagnostic_status(f"❌ {_label}: {_nonretryable_summary}") # The endpoint/status trace is developer detail: verbose only (the log has it always). if getattr(agent, "verbose_logging", False): _vlines( agent, f" 🔌 Provider: {provider} Model: {model} (HTTP {status_code})", f" 🌐 Endpoint: {base_url}", ) _welcome_hint = _welcome_tier_guidance(classified, model=model, in_chat=False) _prefix_suggestion = _missing_vendor_prefix_suggestion(api_error, provider, model) if _welcome_hint: # A free-tier gate or a wrong-host refusal: the way forward is a sign-in or another # provider, never the key/credits advice below. _vlines(agent, f" 💡 {_welcome_hint}") elif classified.is_auth or classified.reason == FailoverReason.billing: _print_nonretryable_auth_guidance( agent, classified, status_code=status_code, provider=provider, base_url=base_url, model=model ) elif classified.reason == FailoverReason.model_not_found: _vlines(agent, f" 💡 Model '{model}' isn't available on {_plabel}. Pick another with /model.") if _prefix_suggestion: _vlines(agent, f" Did you mean '{_prefix_suggestion}'? It looks like the vendor prefix is missing.") elif classified.reason not in _NONRETRYABLE_LABELS: _vlines(agent, f" 💡 Fix: pick another model (/model), or check `{display_hermes_home()}/logs/agent.log`.") # A WAF/CDN block (#53099, #70566): the key never reached the provider; the usual cause # is the SDK User-Agent, which the per-provider extra_headers override. if classified.reason == FailoverReason.upstream_blocked: _vlines( agent, " 💡 The endpoint's firewall/CDN blocked the request before it reached the model — your key", " and model access are probably fine. Relays often reject the SDK's default User-Agent:", " set `extra_headers: {User-Agent: HermesAgent/1.0}` on the custom_providers entry,", " or check the proxy/WAF rules and your network.", ) # Content-policy blocks: the provider refused this prompt, so recovery is a rephrase # or another model, not key/retry advice. if classified.reason == FailoverReason.content_policy_blocked: _vlines( agent, f" 💡 {CONTENT_POLICY_NEXT_STEPS}", " To route future blocks to another provider automatically: hermes fallback add", ) # TLS certificate failures are environment problems — name the knobs for each cause. if classified.reason == FailoverReason.ssl_cert_verification: _vlines( agent, " 💡 Hermes couldn't verify the provider's security certificate. This fails the same", " way on every retry — fix the environment, then try again:", " • Corporate TLS-inspecting proxy? Ask your administrator to install", " its root certificate in the operating system trust store.", " • Missing/stale system CA store? Refresh the OS certificate store.", " A provider-specific CA can also be configured with ssl_ca_cert.", " • Self-signed local endpoint (llama.cpp, LM Studio, vLLM)? Use http://", " for localhost, or add the server's cert to your trust store.", ) logger.error("%sNon-retryable client error: %s", agent.log_prefix, api_error) # Skip persistence on likely context-overflow (400 + large session): persisting the # failed message grows the session and repeats the failure. # Persisting the failed user message would make the session even larger, causing the same failure on the # next attempt. (#1630) if status_code == 400 and (approx_tokens > 50000 or len(api_messages) > 80): _vlines(agent, "⚠️ Skipping session persistence for large failed session to prevent growth loop.") else: agent._persist_session(messages, conversation_history) if classified.reason == FailoverReason.content_policy_blocked: return _content_policy_blocked_result( messages, api_call_count, final_response="⚠️ " + content_policy_copy(label=_plabel, summary=_nonretryable_summary), error_detail=_nonretryable_summary, ) # Billing walls get the same structured recovery descriptor as the max-retries path # so every surface renders one consistent signal. if classified.reason == FailoverReason.billing: return _billing_failure_result( classified=classified, summary=_nonretryable_summary, messages=messages, api_call_count=api_call_count, provider=provider, base_url=base_url, model=model, ) if _welcome_hint: # A free-tier refusal is fully explained by its own sentence; the raw provider summary # (status codes, JSON) is for the log, not for a first-time user's chat. _final_response = _welcome_tier_guidance(classified, model=model, in_chat=True) else: # Every surface reads final_response; the CLI hint lines above never reach chat. _final_response = nonretryable_copy( classified, provider=provider, model=model, summary=_nonretryable_summary, prefix_suggestion=_prefix_suggestion, ) # Same verdict fields as the max-retries path: without them the UI descriptor # (agent/error_surface.py) reads a rejected OAuth token as a retryable # "Provider error" and offers Retry instead of a re-login. _final_response, _keep_partial = _with_delivered_partial( _final_response, _nonretryable_summary, delivered, ) result = _failed_turn_result(_final_response, messages, api_call_count, _nonretryable_summary) result.update({ "failure_reason": classified.reason.value, "failure_retryable": bool(classified.retryable), }) if _keep_partial: result["partial"] = True _stamp_limit_reset(result, agent, api_error) if _welcome_hint and (_kind := _welcome_surface_kind(classified)): # The card form: the desktop renders the sign-in as a button, so no "To sign in" tail. _stamp_free_tier(result, _kind, _welcome_tier_guidance(classified, model=model, in_chat=True, door=False)) return result _STREAM_DROP_MARKERS = ( "connection lost", "connection reset", "connection closed", "network connection", "network error", "terminated", ) def max_retries_exhausted_result( agent: Any, api_error: Exception, classified: Any, *, max_retries: int, is_rate_limited: bool, error_msg: str, api_kwargs: Any, api_messages: Any, messages: List[Dict[str, Any]], conversation_history: Any, api_call_count: int, approx_tokens: int, provider: Any, base_url: Any, model: Any, delivered: str = "", ) -> Dict[str, Any]: """Terminal path once retries, transport recovery and fallback all failed: flush the trace, emit the billing / rate-limit / generic status, print stream-drop or thinking-timeout guidance (the latter wins), persist, build the result with ``failure_reason`` / ``failure_retryable`` / ``billing_block``.""" # Result/guidance helpers stay in the loop module (tests import + patch them there). from hermes_cli.anon_auth import is_anonymous_agent from agent.conversation_loop import ( _billing_block_dict, _billing_or_entitlement_message, _billing_terminal_label, _print_billing_or_entitlement_guidance, ) agent._flush_status_buffer() _final_summary = agent._summarize_api_error(api_error) _billing_guidance = "" _is_billing = classified.reason == FailoverReason.billing if _is_billing: if classified.billing_unverified: # Ambiguous body — hedge the terminal line. agent._emit_diagnostic_status( "❌ Provider reported usage/credit exhaustion " f"(unverified — may be a content-filter rejection) — {_final_summary}" ) else: agent._emit_diagnostic_status(f"❌ Billing or credits exhausted — {_final_summary}") _billing_kw = dict( capability="model access", provider=provider, base_url=str(base_url), model=model, unverified=classified.billing_unverified, ) _billing_guidance = _billing_or_entitlement_message(**_billing_kw) _print_billing_or_entitlement_guidance(agent, **_billing_kw) elif is_rate_limited: _reset = reset_hint(api_error) agent._emit_diagnostic_status( f"❌ Rate limited after {max_retries} retries — {_final_summary}" f"{f' (resets in {_reset})' if _reset else ''}" ) else: agent._emit_diagnostic_status(f"❌ API failed after {max_retries} retries — {_final_summary}") _vlines(agent, f" 💀 Final error: {_final_summary}") _welcome_hint = _welcome_tier_guidance(classified, model=model, in_chat=False) if _welcome_hint: _vlines(agent, f" 💡 {_welcome_hint}") # SSE stream-drop (e.g. "Network connection lost"): usually a proxy/CDN cutting a very # large tool call mid-response. _is_stream_drop = ( not getattr(api_error, "status_code", None) and any(p in error_msg for p in _STREAM_DROP_MARKERS) ) if _is_stream_drop: _vlines( agent, " 💡 The provider's stream connection keeps dropping. This often happens " "when the model tries to write a very large file in a single tool call.", " Try asking the model to use execute_code with Python's open() for " "large files, or to write the file in smaller sections.", ) # A known reasoning model hit a transport error before the first content token. # Distinct from _is_stream_drop; detection lives in agent.thinking_timeout_guidance. _is_thinking_timeout = is_thinking_timeout(classified, model, error_msg) if _is_thinking_timeout: _vlines(agent, f" 💡 {build_thinking_timeout_guidance(provider=provider, model=model).strip()}") logger.error( "%sAPI call failed after %s retries. %s | provider=%s model=%s msgs=%s tokens=~%s", agent.log_prefix, max_retries, _final_summary, provider, model, len(api_messages), f"{approx_tokens:,}", ) if api_kwargs is not None: agent._dump_api_request_debug(api_kwargs, reason="max_retries_exhausted", error=api_error) agent._persist_session(messages, conversation_history) _billing_block = None _billing_unverified = False _free_tier_kind = "" if _is_billing: _billing_unverified = classified.billing_unverified _final_response = _billing_terminal_label(_final_summary, _billing_unverified) if _billing_guidance: _final_response += f"\n\n{_billing_guidance}" # Structured recovery descriptor so every surface renders the same link + label. _billing_block = _billing_block_dict( provider, base_url, model, _billing_guidance, unverified=_billing_unverified ) else: # Every surface reads final_response (the 💡 lines above are CLI-only), so the chat # text carries the plain what-happened + next step itself. _reset_at = classified.error_context.get("reset_at") _final_response = exhausted_copy( classified.reason.value, label=provider_label_for(provider), attempts=max_retries, summary=_final_summary, reset_seconds=_reset_at - time.time() if _reset_at else None, ) if _welcome_hint: _final_response = _welcome_tier_guidance(classified, model=model, in_chat=True) _free_tier_kind = _welcome_surface_kind(classified) elif _outage := _welcome_outage_copy(base_url, classified, anonymous=is_anonymous_agent(agent)): _final_response, _free_tier_kind = _outage, "outage" if _is_thinking_timeout: # Thinking-timeout guidance overrides stream-drop guidance, which would wrongly # suggest splitting large file writes. _final_response += "\n\n" + build_thinking_timeout_guidance(provider=provider, model=model) elif _is_stream_drop: _final_response += ( "\n\nThe connection kept dropping while the model was writing — this often " "happens when it writes a very large file in one go. Ask me to write the file in " "smaller sections (or via execute_code with Python's open())." ) result = _failed_turn_result(_final_response, messages, api_call_count, _final_summary) result.update({ # Classified reason so callers (kanban worker in cli.py) can tell a quota wall # (``rate_limit`` / ``billing``) from a task failure. "failure_reason": classified.reason.value, # The classifier's own retry verdict — UI surfaces use this, not the reason string. "failure_retryable": bool(classified.retryable), # True when the billing verdict rests on an ambiguous body. "billing_unverified": _billing_unverified, # Present only for billing walls: (provider, billing_url, is_nous, message). "billing_block": _billing_block, }) # Retry-exhaustion after partial delivery (#119001): the text was already # shown, so keep it as the reply (marked failed) instead of an error-only # turn — the gateway flags ``partial`` and surfaces retain the bubble. _final_response, _keep_partial = _with_delivered_partial( _final_response, _final_summary, delivered, ) if _keep_partial: result["final_response"] = _final_response result["partial"] = True _stamp_limit_reset(result, agent, api_error) if _free_tier_kind: _stamp_free_tier(result, _free_tier_kind, ( _welcome_tier_guidance(classified, model=model, in_chat=True, door=False) if _welcome_hint else _final_response)) return result def log_api_error_attempt( agent: Any, api_error: Exception, *, retry_count: int, max_retries: int, status_code: Optional[int], elapsed_time: float, api_messages: Any, approx_tokens: int, retryable: bool = True, ) -> Tuple[str, str, Any, Any, Any]: """Log one failed API attempt (warning + buffered retry trace, OpenRouter "no tool endpoints" hint, bare-404 missing-vendor-prefix hint); the buffer only surfaces if every retry+fallback exhausts. Returns ``(error_type, error_msg, provider, base_url, model)``. ``retryable=False`` (the classifier's verdict, e.g. a 401 on a static-key route) is named on the line: a bare ``attempt 1/3`` promises a second attempt that never comes and sends readers hunting for a retry bug (#73237).""" error_type = type(api_error).__name__ error_msg = str(api_error).lower() _error_summary = agent._summarize_api_error(api_error) _attempt = f"attempt {retry_count}/{max_retries}" + ("" if retryable else ", not retryable") logger.warning( "API call failed (%s) error_type=%s %s summary=%s", _attempt, error_type, agent._client_log_context(), _error_summary, ) _provider = getattr(agent, "provider", "unknown") _base = getattr(agent, "base_url", "unknown") _model = getattr(agent, "model", "unknown") _blines(agent, f"⚠️ {_attempt[0].upper()}{_attempt[1:]} failed: {_error_summary}") # Exception class, endpoint, raw body and token counts are developer detail: verbose only. if getattr(agent, "verbose_logging", False): _status_code_str = f" [HTTP {status_code}]" if status_code else "" _blines( agent, f" 🔌 {error_type}{_status_code_str} Provider: {_provider} Model: {_model}", f" 🌐 Endpoint: {_base}", ) if status_code and status_code < 500: _err_body = getattr(api_error, "body", None) _err_body_str = str(_err_body)[:300] if _err_body else None if _err_body_str: _blines(agent, f" 📋 Details: {_err_body_str}") _blines(agent, f" ⏱️ Elapsed: {elapsed_time:.2f}s Context: {len(api_messages)} msgs, ~{approx_tokens:,} tokens") if agent._is_openrouter_url() and "support tool use" in error_msg: _blines(agent, f" 💡 No OpenRouter providers for {_model} support tool calling with your current settings.") from agent.chat_completion_helpers import _provider_preferences_for_agent if _provider_preferences_for_agent(agent).get("only"): _blines( agent, " Your provider_routing.only restriction is filtering out tool-capable providers.", " Try removing the restriction or adding providers that support tools for this model.", ) _blines(agent, f" Check which providers support tools: https://openrouter.ai/models/{_model}") # Bare 404 on a ``vendor/model`` catalogue usually means the id lost its prefix; the # provider never names the model, so we do. _suggestion = _missing_vendor_prefix_suggestion(api_error, _provider, _model) if _suggestion: _blines( agent, f" 💡 Model '{_model}' is not a valid id for provider {_provider} — it is missing its vendor prefix.", f" Did you mean '{_suggestion}'? Re-pick it with /model.", ) return error_type, error_msg, _provider, _base, _model def abort_turn_on_interrupt( agent: Any, messages: List[Dict[str, Any]], conversation_history: Any, api_call_count: int, *, abort_message: str, interrupt_text: str, ) -> Dict[str, Any]: """Announce ``abort_message``, close any open tool sequence with ``interrupt_text``, persist, clear the interrupt and return the ``interrupted`` result dict.""" _vlines(agent, f"⚡ {abort_message}") # Empty-response recovery can leave a synthetic assistant+nudge pair after an # already-executed tool result. Strip only that request-local scaffold before # closing, so this exit owner can persist its specific interrupt reason. agent._drop_trailing_empty_response_scaffolding(messages) close_interrupted_tool_sequence(messages, interrupt_text) agent._persist_session(messages, conversation_history) # The turn was stopped, not rebuilt: a pending steer was aimed at this turn's next # tool iteration, which will no longer happen — drop it (hard-cancel semantics). agent.clear_interrupt(hard_cancel=True) return { "final_response": interrupt_text, "messages": messages, "api_calls": api_call_count, "completed": False, "interrupted": True, } def interruptible_backoff_sleep( agent: Any, wait_time: float, _retry: Optional[TurnRetryState], *, messages: List[Dict[str, Any]], conversation_history: Any, api_call_count: int, abort_message: str, interrupt_text: str, activity_label: str, ) -> Optional[Dict[str, Any]]: """Sleep ``wait_time`` in 200 ms slices so interrupts are honoured promptly, touching activity every ~30 s so the gateway's inactivity monitor knows we are alive. On interrupt with ``_retry`` given and a redirect pending: preserve the redirect, arm ``_retry.restart_with_redirected_messages`` and return ``None`` (caller rebuilds the turn). Otherwise return the ``interrupted`` result dict. ``None`` when the wait completed.""" sleep_end = time.time() + wait_time _touch_counter = 0 while time.time() < sleep_end: if agent._interrupt_requested: if _retry is not None and agent.clear_interrupt(preserve_redirect=True): _retry.restart_with_redirected_messages = True return None return abort_turn_on_interrupt( agent, messages, conversation_history, api_call_count, abort_message=abort_message, interrupt_text=interrupt_text, ) time.sleep(0.2) _touch_counter += 1 if _touch_counter % 150 == 0: # 150 × 0.2s = 30s agent._touch_activity(f"{activity_label}, {int(sleep_end - time.time())}s remaining") return None _ZAI_POLICY_NOTES = { "zai_coding_overload_long": " (Z.AI Coding overload adaptive long backoff)", "zai_coding_overload_short": " (Z.AI Coding overload short retry)", } def reset_hint(api_error: Exception) -> str: """``"~13m"`` until the ``reset_at`` parsed from *api_error* (epoch s/ms or ISO-8601), else ``""``. A bare "Rate limited. Waiting 60s" hides the one fact that decides whether to wait or switch models (#26889): a per-minute throttle and a 13-minute plan window look identical without it.""" from agent.agent_runtime_helpers import extract_api_error_context from agent.credential_pool import _parse_absolute_timestamp from agent.usage_pricing import format_duration_compact reset_at = extract_api_error_context(api_error).get("reset_at") if reset_at is None: return "" remaining = (_parse_absolute_timestamp(reset_at) or 0.0) - time.time() return f"~{format_duration_compact(remaining)}" if remaining >= 1 else "" def compute_error_backoff( agent: Any, api_error: Exception, *, retry_count: int, max_retries: int, is_rate_limited: bool, is_zai_coding_overload: bool, base_url: Any, model: Any, ) -> float: """Pick the wait before the next API retry and announce it. Retry-After wins for rate limits and any other retryable error (capped at 600s: Anthropic Tier 1 buckets reset in ~171s, so a 120s cap re-tripped the limit); otherwise jittered backoff, replaced by the adaptive policy for 429s / Z.AI overloads. Normal retries are buffered; long Z.AI Coding waits surface immediately.""" # Imported lazily so tests that patch ``agent.retry_utils.jittered_backoff`` / # ``adaptive_rate_limit_backoff`` (incl. the run_agent conftest fast-backoff fixture) intercept. from agent.retry_utils import adaptive_rate_limit_backoff, jittered_backoff, parse_retry_after_seconds # Respect Retry-After on every retryable provider error, not just 429s. Retryable # 5xx responses (e.g. Cloudflare 520/524) also carry the header or a structured # ``retry_after`` problem-detail body field; ignoring either turns an origin # outage into a retry storm. _retry_after = parse_retry_after_seconds( getattr(getattr(api_error, "response", None), "headers", None) ) if _retry_after is None: _error_body = getattr(api_error, "body", None) if isinstance(_error_body, dict): # Some providers nest it as error.retry_after (the same unwrap # extract_api_error_context uses), others put it at the top level. _nested = _error_body.get("error") _payload = _nested if isinstance(_nested, dict) else _error_body _retry_after = parse_retry_after_seconds(_payload.get("retry_after")) if _retry_after is not None: # Cap at 10 minutes. Anthropic Tier 1 input-token buckets reset in ~171s, so a 120s cap # caused us to retry before the actual reset window and re-trip the limit. 600s covers all # realistic provider reset windows while still rejecting pathological values. (#26293) _retry_after = min(_retry_after, 600) if _retry_after <= 0: # A zero/expired cooldown (retry-after: 0, or an HTTP-date in the # past, which the parser clamps to 0.0) carries no usable wait — # treat it as absent so we never hot-loop the provider. _retry_after = None wait_time = _retry_after if _retry_after is not None else jittered_backoff(retry_count, base_delay=2.0, max_delay=60.0) _backoff_policy = None _adaptive = is_rate_limited or is_zai_coding_overload if _adaptive and _retry_after is None: wait_time, _backoff_policy = adaptive_rate_limit_backoff( retry_count, base_url=str(base_url), model=model, error=api_error, default_wait=wait_time, ) _reset = reset_hint(api_error) if _adaptive else "" _wait_reason = "Provider overloaded" if is_zai_coding_overload and not is_rate_limited else "Rate limited" if _adaptive: _policy_note = _ZAI_POLICY_NOTES.get(_backoff_policy or "", "") _rate_limit_status = ( f"⏱️ {_wait_reason}.{f' Resets in {_reset}.' if _reset else ''} Waiting {wait_time:.1f}s " f"(attempt {retry_count + 1}/{max_retries}){_policy_note}..." ) if _backoff_policy == "zai_coding_overload_long": agent._emit_diagnostic_status(_rate_limit_status) else: agent._buffer_diagnostic_status(_rate_limit_status) else: _retry_status = ( f"⏳ Retrying in {wait_time:.1f}s (attempt {retry_count}/{max_retries})..." ) if _retry_after is not None and _retry_after > 60: # A 5xx Retry-After can now reach the 600s cap; buffering that wait # would leave the user silent for minutes, so surface long provider # cooldowns immediately (mirrors the zai_coding_overload_long path). agent._emit_diagnostic_status(_retry_status) else: agent._buffer_diagnostic_status(_retry_status) # The buffered line only replays if every retry fails; the live status # line is the one thing the user sees meanwhile. Name the wait there so a # 60s backoff after a 5xx is not an anonymous spinner — this is transient # (rewritten by the next frame, cleared on recovery), so it does not add # the transcript chatter the buffer exists to avoid. The reset window # belongs here too: during the wait this line is the only place the user # can learn whether to sit it out or switch models. _live_reason = f"{_wait_reason.lower()} — resets in {_reset}," if _reset else "waiting on provider —" agent._emit_diagnostic_wait( f"⏳ {_live_reason} retrying in {wait_time:.0f}s (attempt {retry_count}/{max_retries})" ) logger.warning( "Retrying API call in %ss (attempt %s/%s) %s policy=%s error=%s", wait_time, retry_count, max_retries, agent._client_log_context(), _backoff_policy or "default", api_error, ) return wait_time def _codex_soft_failure_error(response: Any) -> Dict[str, Any]: """``response.error`` of a Codex ``failed``/``cancelled`` Response as ``{"code", "message"}`` (the SDK types it as ``ResponseError``; the raw-SSE assembler keeps the dict); ``{}`` when absent.""" error_obj = getattr(response, "error", None) if not error_obj: return {} if isinstance(error_obj, dict): fields = error_obj elif hasattr(error_obj, "code") or hasattr(error_obj, "message"): fields = {"code": getattr(error_obj, "code", None), "message": getattr(error_obj, "message", None)} else: fields = {"message": str(error_obj)} return {k: v for k, v in fields.items() if isinstance(v, str) and v.strip()} class _CodexSoftFailure(Exception): """A Codex HTTP-200 ``status=failed`` Response reshaped so ``classify_api_error`` and ``extract_api_error_context`` read ``response.error`` exactly like an SDK error body.""" def __init__(self, error: Dict[str, Any]) -> None: super().__init__(error.get("message") or "") self.body = {"error": error} def classify_codex_soft_failure(agent: Any, response: Any) -> Tuple[Any, Dict[str, Any]]: """``(classified, error_context)`` for a Codex ``failed``/``cancelled`` Response, or ``(None, {})`` when it is not one. The SDK never raises on these HTTP-200 soft failures, so this is the only place their quota/billing/auth semantics reach the credential pool.""" if agent.api_mode != "codex_responses": return None, {} if str(getattr(response, "status", "") or "").strip().lower() not in {"failed", "cancelled"}: return None, {} exc = _CodexSoftFailure(_codex_soft_failure_error(response)) classified = classify_api_error( exc, provider=getattr(agent, "provider", "") or "", model=getattr(agent, "model", "") or "", base_url=str(getattr(agent, "base_url", "") or ""), api_key=getattr(agent, "api_key", None), ) return classified, agent._extract_api_error_context(exc) def validate_response_shape(agent: Any, response: Any) -> Tuple[bool, List[str]]: """Validate the raw provider response via the transport; ``(response_invalid, error_details)``. A Codex ``failed``/``cancelled`` status (e.g. quota exhaustion) is invalid so the fallback chain triggers; an empty Codex ``output`` with non-empty ``output_text`` is deferred to normalization.""" if agent._get_transport().validate_response(response): return False, [] if response is None: return True, ["response is None"] if agent.api_mode == "codex_responses": _codex_resp_status = str(getattr(response, "status", "") or "").strip().lower() if _codex_resp_status in {"failed", "cancelled"}: _codex_error_msg = ( _codex_soft_failure_error(response).get("message") or f"Responses API returned status '{_codex_resp_status}'" ) logger.warning( "Codex response status='%s' (error=%s). Routing to fallback. %s", _codex_resp_status, _codex_error_msg, agent._client_log_context(), ) return True, [f"response.status={_codex_resp_status}: {_codex_error_msg}"] # Stream backfill may have failed but normalize can still recover from output_text. _out_text = getattr(response, "output_text", None) _out_text_stripped = _out_text.strip() if isinstance(_out_text, str) else "" if _out_text_stripped: logger.debug( "Codex response.output is empty but output_text is present " "(%d chars); deferring to normalization.", len(_out_text_stripped), ) return False, [] logger.warning( "Codex response.output is empty after stream backfill " "(status=%s, incomplete_details=%s, model=%s). %s", getattr(response, "status", None), getattr(response, "incomplete_details", None), getattr(response, "model", None), f"api_mode={agent.api_mode} provider={agent.provider}", ) return True, ["response.output is empty"] if agent.api_mode == "anthropic_messages": detail = "response.content invalid (not a non-empty list)" elif agent.api_mode == "bedrock_converse": detail = "Bedrock response invalid (no output or choices)" elif not hasattr(response, 'choices'): detail = "response has no 'choices' attribute" elif response.choices is None: detail = "response.choices is None" else: detail = "response.choices is empty" return True, [detail] def describe_invalid_response(agent: Any, response: Any, api_duration: float) -> Tuple[str, str, str]: """Diagnostics for an empty/malformed response: ``(error_msg, provider_name, failure_hint)``. The hint is derived from the provider error code (524/504/429/ 5xx) and the response time, instead of always assuming rate limiting.""" error_msg = "Unknown" provider_name = "Unknown" _has_error = bool(response and hasattr(response, 'error') and response.error) if _has_error: # A typed ``ResponseError`` stringifies as its repr; show the provider's message. error_msg = _codex_soft_failure_error(response).get("message") or str(response.error) if hasattr(response.error, 'metadata') and response.error.metadata: provider_name = response.error.metadata.get('provider_name', 'Unknown') elif response and hasattr(response, 'message') and response.message: error_msg = str(response.message) # OpenRouter often returns the actual model used. if provider_name == "Unknown" and response and hasattr(response, 'model') and response.model: provider_name = f"model={response.model}" if provider_name == "Unknown" and response: resp_attrs = {k: str(v)[:100] for k, v in vars(response).items() if not k.startswith('_')} if agent.verbose_logging: logging.debug(f"Response attributes for invalid response: {resp_attrs}") _resp_error_code = None if _has_error: _code_raw = getattr(response.error, 'code', None) if _code_raw is None and isinstance(response.error, dict): _code_raw = response.error.get('code') if _code_raw is not None: try: _resp_error_code = int(_code_raw) except (TypeError, ValueError): pass return error_msg, provider_name, _failure_hint_for(_resp_error_code, api_duration) def _failure_hint_for(code: Optional[int], api_duration: float) -> str: """Human-readable hint from the provider error code and response time.""" if code == 524: return f"upstream provider timed out (Cloudflare 524, {api_duration:.0f}s)" if code == 504: return f"upstream gateway timeout (504, {api_duration:.0f}s)" if code == 429: return "rate limited by upstream provider (429)" if code in {500, 502}: return f"upstream server error ({code}, {api_duration:.0f}s)" if code in {503, 529}: return f"upstream provider overloaded ({code})" if code is not None: return f"upstream error (code {code}, {api_duration:.0f}s)" if api_duration < 10: return f"fast response ({api_duration:.1f}s) — likely rate limited" if api_duration > 60: return f"slow response ({api_duration:.0f}s) — likely upstream timeout" return f"response time {api_duration:.1f}s" @dataclass class ClassifiedErrorVerdict: """Outcome of ``route_classified_error``. ``action``: ``"return"`` (terminal result), ``"break"`` (restart armed on ``_retry``), ``"continue"`` (re-enter the retry loop; Nous guard re-check) or ``"fallthrough"`` (proceed to overflow / client-error / backoff handling). The remaining fields are loop locals the router rebound or computed.""" action: str result: Optional[Dict[str, Any]] status_code: Optional[int] messages: List[Dict[str, Any]] active_system_prompt: Any conversation_history: Any retry_count: int max_retries: int compression_attempts: int provider_overflow_recovery_pending: bool is_rate_limited: bool wrapped_output_cap_budget: Optional[int] is_zai_coding_overload: bool _OVERFLOW_REASONS = frozenset({ FailoverReason.long_context_tier, FailoverReason.payload_too_large, FailoverReason.context_overflow, }) _RATE_LIMIT_REASONS = frozenset({ FailoverReason.rate_limit, FailoverReason.billing, FailoverReason.upstream_rate_limit, }) _TRANSPORT_FAILURE_REASONS = frozenset({FailoverReason.timeout, FailoverReason.overloaded}) _LONG_CONTEXT_TIER_CAP = 200000 def _cap_long_context_tier(agent: Any) -> int: """Cap the compressor's context window at the long-context tier limit; returns the previous ``context_length``.""" compressor = agent.context_compressor old_ctx = compressor.context_length if old_ctx > _LONG_CONTEXT_TIER_CAP: compressor.update_model( model=agent.model, context_length=_LONG_CONTEXT_TIER_CAP, base_url=agent.base_url, api_key=getattr(agent, "api_key", ""), provider=agent.provider, api_mode=agent.api_mode, ) # Context probing flags exist only on the built-in compressor (plugin engines # manage their own). Don't persist — a tier limit, not a model capability; # 1M should return if extra usage is enabled. if hasattr(compressor, "_context_probed"): compressor._context_probed = True compressor._context_probe_persistable = False agent._buffer_vprint( f"⚠️ Anthropic long-context tier " f"requires extra usage — reducing context: " f"{old_ctx:,} → {_LONG_CONTEXT_TIER_CAP:,} tokens" ) return old_ctx def _eager_fallback_status(classified: Any, is_upstream: bool, is_transport_failure: bool) -> str: """Status line announcing an eager fallback switch.""" if is_upstream: _upstream_name = (classified.error_context or {}).get("upstream_provider", "aggregator") return f"⚠️ Upstream {_upstream_name} rate-limited — switching to fallback model..." if classified.reason == FailoverReason.billing: if classified.billing_unverified: # Ambiguous body — don't assert billing. return ( "⚠️ Provider reported usage/credit exhaustion " "(unverified — may be a content-filter rejection) " "— switching to fallback provider..." ) return "⚠️ Billing or credits exhausted — switching to fallback provider..." if is_transport_failure: return "⚠️ Provider unreachable — switching to fallback provider..." return "⚠️ Rate limited — switching to fallback provider..." def activate_codex_app_server_fallback(agent: Any, result: Dict[str, Any]) -> bool: """The codex app-server runtime reports a failed turn as ``result["error"]`` text instead of raising, so the generic classify -> ``fallback_providers`` chain never saw it (#71633). Classify that text; on a billing / rate-limit verdict activate the configured fallback and return True so the caller re-runs the same user turn on the generic loop.""" error = result.get("error") if not error or result.get("interrupted") or not agent._has_pending_fallback(): return False classified = classify_api_error( RuntimeError(str(error)), provider=getattr(agent, "provider", "") or "", model=getattr(agent, "model", "") or "", ) if classified.reason not in _RATE_LIMIT_REASONS: return False agent._buffer_diagnostic_status( _eager_fallback_status(classified, classified.reason == FailoverReason.upstream_rate_limit, False)) return bool(agent._try_activate_fallback(reason=classified.reason)) def _is_genuine_nous_rate_limit(agent: Any, api_error: Exception, error_context: Any, classified: Any = None) -> bool: """Record a genuine account-level Nous 429 to the cross-session breaker; upstream capacity 429s (no exhausted bucket in headers or last-known state) are left alone. *error_context* is the turn's (``extract_api_error_context``); *classified* brings the classifier's own context, where a welcome-tier ``rate_limited`` refusal and its ``reset_at`` live. A long welcome reset is an exhausted allowance whatever the headers say, and the one place the user is told that signing in lifts it.""" _genuine = False try: from agent.nous_rate_guard import ( is_genuine_nous_rate_limit, is_long_welcome_rate_limit, record_nous_rate_limit) _err_resp = getattr(api_error, "response", None) _err_hdrs = getattr(_err_resp, "headers", None) if _err_resp else None from hermes_cli.anon_auth import is_anonymous_agent anonymous = is_anonymous_agent(agent) _classified_ctx = getattr(classified, "error_context", None) or {} # Only an anonymous request's fairshare body is an allowance verdict; named # requests keep the exhausted-bucket rule, whatever their host or body says. _genuine = ( (anonymous and is_long_welcome_rate_limit(_classified_ctx)) or is_genuine_nous_rate_limit(headers=_err_hdrs, last_known_state=agent._rate_limit_state)) if _genuine: _merged = {**(error_context if isinstance(error_context, dict) else {}), **_classified_ctx} record_nous_rate_limit(headers=_err_hdrs, error_context=_merged, anonymous=anonymous) else: logger.info( "Nous 429 looks like upstream capacity " "(no exhausted bucket in headers or " "last-known state) -- not tripping " "cross-session breaker." ) except Exception: pass return _genuine def route_classified_error( agent: Any, api_error: Exception, classified: Any, _retry: TurnRetryState, *, error_msg: str, error_context: Any, recovered_with_pool: bool, base_url: Any, model: Any, messages: List[Dict[str, Any]], api_messages: Any, system_message: Any, active_system_prompt: Any, conversation_history: Any, retry_count: int, max_retries: int, compression_attempts: int, max_compression_attempts: int, api_call_count: int, effective_task_id: Any, ) -> ClassifiedErrorVerdict: """Ordered (load-bearing) recovery steps between classification and overflow handling: compaction-disabled overflow → terminal error (output-cap errors exempt); Anthropic long-context tier 429 → cap at 200k and compress; eager fallback for rate-limit/billing (immediately) and transport failures (after 1 retry) unless credential-pool rotation may still recover (upstream-aggregator 429s always fall back); persistent 401/403 → fallback chain once; genuine Nous 429 → cross-session breaker + re-enter the loop exactly once.""" from agent.conversation_compression import conversation_history_after_compression from agent.conversation_loop import _arm_fallback_restart, _ra from agent.model_metadata import estimate_request_tokens_rough _provider_overflow_recovery_pending = False is_rate_limited = False _wrapped_output_cap_budget = None _is_zai_coding_overload = False status_code = getattr(api_error, "status_code", None) def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> ClassifiedErrorVerdict: return ClassifiedErrorVerdict( action=action, result=result, status_code=status_code, messages=messages, active_system_prompt=active_system_prompt, conversation_history=conversation_history, retry_count=retry_count, max_retries=max_retries, compression_attempts=compression_attempts, provider_overflow_recovery_pending=_provider_overflow_recovery_pending, is_rate_limited=is_rate_limited, wrapped_output_cap_budget=_wrapped_output_cap_budget, is_zai_coding_overload=_is_zai_coding_overload, ) def _fallback_break() -> ClassifiedErrorVerdict: nonlocal active_system_prompt, retry_count, compression_attempts active_system_prompt = _arm_fallback_restart(agent, api_messages, active_system_prompt, _retry) retry_count = 0 compression_attempts = 0 return _verdict("break") # ``compression.enabled: false`` forbids every automatic trigger, incl. these # overflow recovery paths; error out. Output-cap errors exempt. _is_output_cap_error = ( is_output_cap_error(error_msg) or parse_available_output_tokens_from_error(error_msg) is not None ) if ( classified.reason in _OVERFLOW_REASONS and not getattr(agent, "compression_enabled", True) and not _is_output_cap_error ): agent._flush_status_buffer() _vlines( agent, "❌ The conversation is too long for the model and automatic shrinking is off (compression.enabled: false).", " 💡 Run /compress to shrink it now, /new to start fresh, " "pick a model with a bigger context window, or remove attachments.", ) logger.error( f"{agent.log_prefix}Context overflow ({classified.reason.value}) with " f"auto-compaction disabled — not compressing." ) agent._persist_session(messages, conversation_history) _final_response = site_copy("compression_disabled", model=agent.model) return _verdict("return", stamp_failure({ "final_response": _final_response, "messages": messages, "completed": False, "api_calls": api_call_count, "error": _final_response, "partial": True, "failed": True, "compaction_disabled": True, }, "context_overflow", False)) # Anthropic 429 "Extra usage is required for long context requests" is a # subscription-tier limit, not transient: cap at 200k and compress. if classified.reason == FailoverReason.long_context_tier: old_ctx = _cap_long_context_tier(agent) compression_attempts += 1 if compression_attempts <= max_compression_attempts: original_len = len(messages) # Overhead-aware request size so recovery arms on the true request # (msgs + tools + system), not the tool-blind message count. messages, active_system_prompt = agent._compress_context( # Route the overhead-aware _real_tokens (computed above) into compression, not the bare # last_prompt_tokens — which is 0 in the no-usage fallback, hiding the true request size # from the engine's overflow guard (upstream PR #77169 review). messages, system_message, approx_tokens=estimate_request_tokens_rough(api_messages, tools=agent.tools or None), task_id=effective_task_id, trigger="overflow", ) conversation_history = conversation_history_after_compression(agent, messages, conversation_history) if len(messages) < original_len or old_ctx > _LONG_CONTEXT_TIER_CAP: agent._buffer_diagnostic_status( COMPRESSION_RETRY_CONTEXT_REDUCED_STATUS_TEMPLATE.format( new_ctx=_LONG_CONTEXT_TIER_CAP, old_ctx=old_ctx ) ) time.sleep(2) # Provider proved the request doesn't fit the reduced window; row count # isn't proof the rebuilt one does. Recheck before the next call. _provider_overflow_recovery_pending = True _retry.restart_with_compressed_messages = True return _verdict("break") # Compression exhausted or didn't help: fall through to normal error handling. # Eager fallback: rate-limit/billing switch immediately (primary won't recover in # the retry window); transport errors get 1 retry first. is_rate_limited = classified.reason in _RATE_LIMIT_REASONS # Some relays wrap upstream output-cap 400s as 429 (rate_limit). Only the max_tokens # clamp fixes it. Parsed once; gates the eager-fallback exemption and overflow entry. # Relay-wrapped output-cap errors: some gateways wrap an upstream "[400]: max_tokens (...) exceeds # model's maximum output tokens (...)" as HTTP 429, which classifies as rate_limit. The failure is a # deterministic request-shape problem — falling back to another provider (or burning generic retries) # can't fix it, but the output-cap clamp below can, in one retry (#72281). Parse once here; the result # gates both the eager-fallback exemption and the widened is_context_length_error entry, and is reused # as available_out inside the handler. _wrapped_output_cap_budget = ( parse_available_output_tokens_from_error(error_msg) if classified.reason == FailoverReason.rate_limit else None ) _is_transport_failure = classified.reason in _TRANSPORT_FAILURE_REASONS # Z.AI overload 429s classify `overloaded`, which `is_rate_limited` excludes. Detect # directly so the long backoff runs, and raise the ceiling to reach it. _is_zai_coding_overload = is_zai_coding_overload_error(base_url=str(base_url), model=model, error=api_error) if _is_zai_coding_overload: max_retries = max(max_retries, zai_coding_overload_retry_ceiling()) _should_fallback = ( (is_rate_limited and _wrapped_output_cap_budget is None) or (_is_transport_failure and retry_count >= 2) ) if _should_fallback and agent._fallback_index < len(agent._fallback_chain): # No eager fallback while credential pool rotation may recover. Exception: an # upstream-aggregator 429 — the pool can't help, always fall back. # Fixes #11314. _is_upstream = classified.reason == FailoverReason.upstream_rate_limit pool_may_recover = ( False if _is_upstream else _ra()._pool_may_recover_from_rate_limit(agent._credential_pool) ) if not pool_may_recover: agent._buffer_diagnostic_status(_eager_fallback_status(classified, _is_upstream, _is_transport_failure)) reset_at = error_context.get("reset_at") if isinstance(error_context, dict) else None if agent._try_activate_fallback(reason=classified.reason, reset_at=reset_at): return _fallback_break() # A 401/403 surviving credential refresh means a broken credential or endpoint: # escalate to the fallback chain once; False -> terminal handling. if ( classified.is_auth and not _retry.auth_failover_attempted and agent._fallback_index < len(agent._fallback_chain) ): _retry.auth_failover_attempted = True agent._buffer_diagnostic_status( "🔐 Authentication failed and could not be refreshed — " "switching to fallback provider..." ) if agent._try_activate_fallback(reason=classified.reason): return _fallback_break() # Nous Portal: a genuine account-level 429 is recorded to a shared file so ALL # sessions back off; is_genuine_nous_rate_limit excludes upstream 429s. if ( is_rate_limited and agent.provider == "nous" and classified.reason == FailoverReason.rate_limit and not recovered_with_pool and _is_genuine_nous_rate_limit(agent, api_error, error_context, classified) ): # Re-enter the loop exactly once so the top-of-loop Nous guard runs # (retry_count = max_retries would skip it entirely). retry_count = max(0, max_retries - 1) return _verdict("continue") # Upstream capacity 429: normal retry logic will typically succeed. return _verdict("fallthrough")