"""API-call exception handler for the conversation turn's retry loop: pre-/post-classification recovery, interpreter-shutdown abandon, classified-error routing, overflow recovery, the non-retryable client-error exit, max-retries exhaustion (primary transport recovery → fallback → terminal result) and the interruptible backoff. Nothing here imports ``agent.conversation_loop`` at module level (cycle) — loop-internal helpers resolve lazily so ``patch("agent.conversation_loop.X")`` sites keep intercepting. """ from __future__ import annotations from dataclasses import dataclass import json import logging import ssl import time from typing import Any, Dict, Optional from agent.api_error_summary import is_provider_stream_parse_error from agent.error_classifier import RETRYABLE_CLIENT_REASONS, FailoverReason, classify_api_error from agent.turn_overflow import recover_from_overflow from agent.turn_recovery import ( _NONRETRYABLE_LABELS, abort_turn_on_interrupt, compute_error_backoff, interruptible_backoff_sleep, log_api_error_attempt, max_retries_exhausted_result, nonretryable_client_error_result, recover_after_classification, recover_before_classification, route_classified_error, settle_delivered_partial, ) logger = logging.getLogger("agent.conversation_loop") @dataclass class ApiErrorVerdict: """``action``: ``"continue"`` (retry the API call), ``"break"`` (leave the retry loop: fallback armed / redirect pending) or ``"return"`` (``result`` is the turn's result dict); ``"fallthrough"`` never happens — the handler always ends in an exit. The other fields are the retry-loop locals the handler rebinds; ``_provider_overflow_recovery_pending`` is merge-only (caller sets True when set).""" action: str thinking_spinner: Any messages: Any active_system_prompt: Any conversation_history: Any approx_tokens: Any retry_count: Any max_retries: Any compression_attempts: Any _provider_overflow_recovery_pending: Any result: Optional[Dict[str, Any]] = None def handle_api_error( agent: Any, *, api_error: Any, _retry: Any, thinking_spinner: Any, messages: Any, api_messages: Any, api_kwargs: Any, system_message: Any, active_system_prompt: Any, conversation_history: Any, approx_tokens: Any, retry_count: Any, max_retries: Any, compression_attempts: Any, max_compression_attempts: Any, api_call_count: Any, api_request_id: Any, api_start_time: Any, effective_task_id: Any, turn_id: Any, current_turn_user_idx: Any = None, ) -> ApiErrorVerdict: """Recover from ``api_error`` in the original order. Every fallback activation must leave the retry loop with ``restart_with_rebuilt_messages`` armed (``"break"``) so the pre-API preflight re-runs against the fallback's context window (#84733).""" _provider_overflow_recovery_pending = False def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> ApiErrorVerdict: return ApiErrorVerdict( action=action, thinking_spinner=thinking_spinner, messages=messages, active_system_prompt=active_system_prompt, conversation_history=conversation_history, approx_tokens=approx_tokens, retry_count=retry_count, max_retries=max_retries, compression_attempts=compression_attempts, _provider_overflow_recovery_pending=_provider_overflow_recovery_pending, result=result, ) # Stop spinner silently — retry status is buffered and only flushed when every # retry+fallback is exhausted. if thinking_spinner: thinking_spinner.stop("") thinking_spinner = None if agent.thinking_callback: agent.thinking_callback("") _recovered, active_system_prompt = recover_before_classification( agent, api_error, messages=messages, api_messages=api_messages, api_kwargs=api_kwargs, active_system_prompt=active_system_prompt, ) if _recovered: return _verdict("continue") status_code = getattr(api_error, "status_code", None) error_context = agent._extract_api_error_context(api_error) # Process is exiting mid-flight: retries/rotation/fallbacks are futile and the # retry trace spams the shell. One log line. from tools.interpreter_shutdown import interpreter_shutting_down if interpreter_shutting_down(api_error): logger.warning( "%sInterpreter is shutting down — abandoning turn " "during API call #%d (%s)", agent.log_prefix, api_call_count, api_error, ) _shutdown_summary = "Turn abandoned: the process was shutting down before the model call could complete." return _verdict("return", { "final_response": _shutdown_summary, "messages": messages, "api_calls": api_call_count, "completed": False, "failed": True, "error": _shutdown_summary, "failure_reason": "interpreter_shutdown", "failure_retryable": False, }) _compressor = getattr(agent, "context_compressor", None) _ctx_len = getattr(_compressor, "context_length", 200000) if _compressor else 200000 classified = classify_api_error( api_error, provider=getattr(agent, "provider", "") or "", model=getattr(agent, "model", "") or "", approx_tokens=approx_tokens, context_length=_ctx_len, num_messages=len(api_messages) if api_messages else 0, base_url=str(getattr(agent, "base_url", "") or ""), api_key=getattr(agent, "api_key", None), ) logger.debug( "Error classified: reason=%s status=%s retryable=%s compress=%s rotate=%s fallback=%s", classified.reason.value, classified.status_code, classified.retryable, classified.should_compress, classified.should_rotate_credential, classified.should_fallback, ) agent._invoke_api_request_error_hook( task_id=effective_task_id, turn_id=turn_id, api_request_id=api_request_id, api_call_count=api_call_count, api_start_time=api_start_time, api_kwargs=api_kwargs, error_type=type(api_error).__name__, error_message=str(api_error), status_code=status_code, retry_count=retry_count, max_retries=max_retries, retryable=classified.retryable, reason=classified.reason.value, ) _recovered, recovered_with_pool = recover_after_classification( agent, api_error, classified, _retry, status_code=status_code, error_context=error_context, messages=messages, api_messages=api_messages, ) if _recovered: return _verdict("continue") retry_count += 1 elapsed_time = time.time() - api_start_time # Liveness/watchdog label only (never shown in chat), so the classifier's # "not retryable" verdict is named on the logged attempt line below instead. agent._touch_activity(f"API error recovery (attempt {retry_count}/{max_retries})") error_type, error_msg, _provider, _base, _model = log_api_error_attempt( agent, api_error, retry_count=retry_count, max_retries=max_retries, status_code=status_code, elapsed_time=elapsed_time, api_messages=api_messages, approx_tokens=approx_tokens, retryable=bool(classified.retryable), ) if agent._interrupt_requested: # Preserve a pending redirect: the user is steering, not stopping — rebuild the # turn from the correction instead of aborting. if agent.clear_interrupt(preserve_redirect=True): _retry.restart_with_redirected_messages = True return _verdict("break") return _verdict("return", abort_turn_on_interrupt( agent, messages, conversation_history, api_call_count, abort_message="Interrupt detected during error handling, aborting retries.", interrupt_text=f"Operation interrupted: handling API error ({error_type}: {agent._clean_error_message(str(api_error))}).", )) _ce = route_classified_error( agent, api_error, classified, _retry, error_msg=error_msg, error_context=error_context, recovered_with_pool=recovered_with_pool, base_url=_base, model=_model, messages=messages, api_messages=api_messages, system_message=system_message, active_system_prompt=active_system_prompt, conversation_history=conversation_history, retry_count=retry_count, max_retries=max_retries, compression_attempts=compression_attempts, max_compression_attempts=max_compression_attempts, api_call_count=api_call_count, effective_task_id=effective_task_id, ) status_code = _ce.status_code messages = _ce.messages active_system_prompt = _ce.active_system_prompt conversation_history = _ce.conversation_history retry_count = _ce.retry_count max_retries = _ce.max_retries compression_attempts = _ce.compression_attempts is_rate_limited = _ce.is_rate_limited _wrapped_output_cap_budget = _ce.wrapped_output_cap_budget _is_zai_coding_overload = _ce.is_zai_coding_overload if _ce.provider_overflow_recovery_pending: _provider_overflow_recovery_pending = True if _ce.action != "fallthrough": return _verdict(_ce.action, _ce.result) _ov = recover_from_overflow( agent, api_error, classified, _retry, status_code=status_code, error_msg=error_msg, wrapped_output_cap_budget=_wrapped_output_cap_budget, messages=messages, api_messages=api_messages, system_message=system_message, active_system_prompt=active_system_prompt, conversation_history=conversation_history, approx_tokens=approx_tokens, compression_attempts=compression_attempts, max_compression_attempts=max_compression_attempts, api_call_count=api_call_count, effective_task_id=effective_task_id, ) messages = _ov.messages active_system_prompt = _ov.active_system_prompt conversation_history = _ov.conversation_history approx_tokens = _ov.approx_tokens compression_attempts = _ov.compression_attempts is_context_length_error = _ov.is_context_length_error if _ov.provider_overflow_recovery_pending: _provider_overflow_recovery_pending = True if _ov.action != "fallthrough": return _verdict(_ov.action, _ov.result) _ue = settle_unrecovered_error( agent, api_error=api_error, classified=classified, _retry=_retry, status_code=status_code, error_msg=error_msg, error_context=error_context, is_context_length_error=is_context_length_error, is_rate_limited=is_rate_limited, _is_zai_coding_overload=_is_zai_coding_overload, _provider=_provider, _base=_base, _model=_model, messages=messages, api_messages=api_messages, api_kwargs=api_kwargs, active_system_prompt=active_system_prompt, conversation_history=conversation_history, approx_tokens=approx_tokens, retry_count=retry_count, max_retries=max_retries, compression_attempts=compression_attempts, api_call_count=api_call_count, current_turn_user_idx=current_turn_user_idx, ) active_system_prompt = _ue.active_system_prompt retry_count = _ue.retry_count compression_attempts = _ue.compression_attempts return _verdict(_ue.action, _ue.result) def _is_local_validation_error(api_error: Any) -> bool: """ValueError/TypeError are local bugs, except: UnicodeEncodeError (surrogate recovery path), json.JSONDecodeError / provider stream-parse ValueErrors (transient provider/network failure, must retry), ssl.SSLError (inherits OSError *and* ValueError — a TLS failure is not a local bug) and "NoneType is not iterable" TypeErrors (upstream shape mismatches, e.g. Codex response.completed.output=null — retryable so the fallback path runs).""" if not isinstance(api_error, (ValueError, TypeError)): return False if isinstance(api_error, (UnicodeEncodeError, json.JSONDecodeError, ssl.SSLError)): return False _text = str(api_error).lower() if is_provider_stream_parse_error(api_error): return False # corrupted provider SSE chunk (jiter), transient — not a local bug (#65147) return not (isinstance(api_error, TypeError) and "nonetype" in _text and "not iterable" in _text) @dataclass class UnrecoveredErrorVerdict: """``action``: ``"continue"`` (retry), ``"break"`` (fallback armed / redirect pending) or ``"return"`` (``result`` is the terminal result dict). Rebinds ``active_system_prompt``, ``retry_count`` and ``compression_attempts``.""" action: str active_system_prompt: Any retry_count: Any compression_attempts: Any result: Optional[Dict[str, Any]] = None def settle_unrecovered_error( agent: Any, *, api_error: Any, classified: Any, _retry: Any, status_code: Any, error_msg: Any, is_context_length_error: Any, is_rate_limited: Any, _is_zai_coding_overload: Any, _provider: Any, _base: Any, _model: Any, messages: Any, api_messages: Any, api_kwargs: Any, active_system_prompt: Any, conversation_history: Any, approx_tokens: Any, retry_count: Any, max_retries: Any, compression_attempts: Any, api_call_count: Any, error_context: Any = None, current_turn_user_idx: Any = None, ) -> UnrecoveredErrorVerdict: """Decide the fate of an API error that every recovery chain declined: local validation / non-retryable client errors (Copilot stale-credential self-heal first, then fallback, then a terminal result), max-retries exhaustion (primary transport recovery -> fallback -> terminal result), else the interruptible error backoff. ``FailoverReason.billing`` (402) is deliberately treated as non-retryable (#31273).""" from agent.conversation_loop import ( _arm_fallback_restart, _is_copilot_provider, _is_stale_copilot_credential_error ) def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> UnrecoveredErrorVerdict: return UnrecoveredErrorVerdict( action=action, active_system_prompt=active_system_prompt, retry_count=retry_count, compression_attempts=compression_attempts, result=result, ) # ``FailoverReason.billing`` (402) is deliberately NOT excluded: pool rotation and # eager fallback already gave up, so retrying only burns paid requests on a depleted # balance. Mirrors 401/403. is_local_validation_error = _is_local_validation_error(api_error) # ``recover_after_classification`` sets ``image_shrink_retry_attempted`` BEFORE it runs the shrink, # so an image-size rejection reaching this point with the flag set had nothing left to shrink (the # excess is text on a host whose cap is payload-scoped, or an unshrinkable image). Re-sending the # byte-identical body ``max_retries`` times changes nothing: treat it as a client error and try # the fallback chain now, as the format_error verdict these 400s carried before did (#112473). shrink_spent = classified.reason == FailoverReason.image_too_large and bool( getattr(_retry, "image_shrink_retry_attempted", False) ) # Same shape for the reasoning-disable rung: the retry already went out without the disable, # so a second reasoning-field rejection means the route refuses the configured reasoning # controls themselves — nothing left to drop, so take the fallback chain now instead of # replaying the identical request ``max_retries`` times (#114460). reasoning_spent = classified.reason == FailoverReason.reasoning_mandatory and bool( getattr(_retry, "reasoning_mandatory_retry_attempted", False) ) is_client_error = ( is_local_validation_error or shrink_spent or reasoning_spent or ( not classified.retryable and not classified.should_compress and classified.reason not in RETRYABLE_CLIENT_REASONS ) ) and not is_context_length_error if is_client_error: # A Codex ChatGPT-account entitlement 400 names the model: with nothing to rotate the # slug is dead for this account, so record it before the fallback walk runs (#106475). from agent.fallback_cooldown import _mark_entitlement_rejected_model _mark_entitlement_rejected_model(agent, api_error) # Copilot self-heal BEFORE fallback: a stale credential yields a 400 # ``model_not_available_for_integrator`` / ``model_not_supported``, not a 401. # Fresh token + client rebuild, one retry, SAME provider. if ( _is_copilot_provider(agent) and not _retry.copilot_stale_cred_retry_attempted and _is_stale_copilot_credential_error( status_code, str(getattr(api_error, "message", "") or api_error) ) ): _retry.copilot_stale_cred_retry_attempted = True if agent._try_recover_stale_copilot_credential(): agent._buffer_vprint( "🔐 Copilot credential re-exchanged after " "model_not_available 400. Retrying request..." ) retry_count = 0 return _verdict("continue") # ``should_fallback=False`` marks a deterministic failure no other provider can fix (the # model's own malformed tool-call JSON, #12770; MoA preset/adapter faults, #55933): skip # the cascade. An UNCLASSIFIED local ValueError/TypeError keeps its historical fallback; # a recognised verdict that opts out wins even when the exception is a ValueError subclass. _unclassified_local = is_local_validation_error and classified.reason == FailoverReason.unknown if classified.should_fallback or _unclassified_local or shrink_spent or reasoning_spent: # Announce the fallback only when a chain exists, else "trying fallback..." lies # before a silent abort. if agent._has_pending_fallback(): _label = _NONRETRYABLE_LABELS.get(classified.reason, f"Non-retryable error (HTTP {status_code})") agent._buffer_diagnostic_status(f"⚠️ {_label} — trying fallback...") 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): # Direct ``return _verdict("break")`` is load-bearing: the restart handler # re-runs the pre-API preflight against the fallback's context window. active_system_prompt = _arm_fallback_restart(agent, api_messages, active_system_prompt, _retry) retry_count = compression_attempts = 0 return _verdict("break") # Terminal from here: collapse the continuation trail once, before the persist. _delivered = settle_delivered_partial(agent, messages, current_turn_user_idx) return _verdict("return", nonretryable_client_error_result( agent, api_error, classified, status_code=status_code, api_kwargs=api_kwargs, api_messages=api_messages, messages=messages, conversation_history=conversation_history, api_call_count=api_call_count, approx_tokens=approx_tokens, provider=_provider, base_url=_base, model=_model, delivered=_delivered, )) if retry_count >= max_retries: # Before fallback, rebuild the primary client once per API call block for # transient transport errors (stale pool, TCP reset). if not _retry.primary_recovery_attempted and agent._try_recover_primary_transport( api_error, retry_count=retry_count, max_retries=max_retries, ): _retry.primary_recovery_attempted = True retry_count = 0 # Fresh attempt cycle: re-open fallback state so a follow-on 429 can still # activate fallback_providers. _retry.has_retried_429 = False agent._fallback_index = 0 agent._fallback_activated = False return _verdict("continue") if agent._has_pending_fallback(): agent._buffer_diagnostic_status(f"⚠️ Max retries ({max_retries}) exhausted — trying fallback...") 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): # Direct ``return _verdict("break")`` is load-bearing: the restart handler # re-runs the pre-API preflight against the fallback's context window. active_system_prompt = _arm_fallback_restart(agent, api_messages, active_system_prompt, _retry) retry_count = compression_attempts = 0 return _verdict("break") # Fallback first (above); only with nothing left to move to does the bounded auto-recovery # ladder park the turn on a transient outage instead of ending it (#85426, #107307). from agent.turn_recovery_autorecover import auto_recover_after_exhaustion _ladder = auto_recover_after_exhaustion( agent, api_error, classified, _retry, messages=messages, conversation_history=conversation_history, api_call_count=api_call_count, ) if _ladder is not None: if _ladder["action"] == "continue": retry_count = 0 return _verdict(_ladder["action"], _ladder.get("result")) _delivered = settle_delivered_partial(agent, messages, current_turn_user_idx) return _verdict("return", max_retries_exhausted_result( agent, api_error, classified, max_retries=max_retries, is_rate_limited=is_rate_limited, error_msg=error_msg, api_kwargs=api_kwargs, api_messages=api_messages, messages=messages, conversation_history=conversation_history, api_call_count=api_call_count, approx_tokens=approx_tokens, provider=_provider, base_url=_base, model=_model, delivered=_delivered, )) wait_time = compute_error_backoff( agent, api_error, retry_count=retry_count, max_retries=max_retries, is_rate_limited=is_rate_limited, is_zai_coding_overload=_is_zai_coding_overload, base_url=_base, model=_model, ) # Same preserve-redirect rule as the invalid-response wait: a steering correction # must survive backoff, not die as "Operation interrupted". _interrupted = interruptible_backoff_sleep( agent, wait_time, _retry, messages=messages, conversation_history=conversation_history, api_call_count=api_call_count, abort_message="Interrupt detected during retry wait, aborting.", interrupt_text=f"Operation interrupted: retrying API call after error (retry {retry_count}/{max_retries}).", activity_label=f"error retry backoff ({retry_count}/{max_retries})", ) if _interrupted is not None: return _verdict("return", _interrupted) if _retry.restart_with_redirected_messages: # Leave the retry loop — the caller rebuilds this iteration from the correction # instead of re-firing the stale request. return _verdict("break") return _verdict("fallthrough")