diff --git a/agent/turn_api_error.py b/agent/turn_api_error.py index 61a5ca287d..0c05c78b6f 100644 --- a/agent/turn_api_error.py +++ b/agent/turn_api_error.py @@ -80,11 +80,6 @@ def handle_api_error( """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).""" - from agent.conversation_loop import ( - _arm_fallback_restart, - _is_copilot_provider, - _is_stale_copilot_credential_error, - ) _provider_overflow_recovery_pending = False def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> ApiErrorVerdict: @@ -313,6 +308,96 @@ def handle_api_error( if _ov.action == "continue": return _verdict("continue") + _ue = settle_unrecovered_error( + agent, + api_error=api_error, + classified=classified, + _retry=_retry, + status_code=status_code, + error_msg=error_msg, + 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, + ) + active_system_prompt = _ue.active_system_prompt + retry_count = _ue.retry_count + compression_attempts = _ue.compression_attempts + if _ue.action != "fallthrough": + return _verdict(_ue.action, _ue.result) + return _verdict("fallthrough") + + +@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, +) -> 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, + ) + # Non-retryable: ValueError/TypeError are local bugs, except # UnicodeEncodeError (surrogate path above) and json.JSONDecodeError, a # transient provider/network failure that must be retried (#14782). diff --git a/agent/turn_response_check.py b/agent/turn_response_check.py index a65c63956d..a14e636763 100644 --- a/agent/turn_response_check.py +++ b/agent/turn_response_check.py @@ -74,10 +74,6 @@ def check_api_response( (bytes back != usable content); ``_preflight_compression_blocked``/``_last_preflight_pressure`` reset only when the usage fold re-arms the compression budget.""" from agent.conversation_loop import ( - _arm_fallback_restart, - describe_invalid_response, - jittered_backoff, - interruptible_backoff_sleep, validate_response_shape, ) api_duration = None @@ -122,101 +118,33 @@ def check_api_response( response_invalid, error_details = validate_response_shape(agent, response) if response_invalid: - 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, + _iv = retry_invalid_response( + agent, + response=response, + error_details=error_details, + _retry=_retry, + thinking_spinner=thinking_spinner, + messages=messages, + api_messages=api_messages, api_kwargs=api_kwargs, - error_type="InvalidAPIResponse", - error_message=", ".join(error_details) or "Invalid API response", - status_code=getattr(getattr(response, "error", None), "code", None), + active_system_prompt=active_system_prompt, + conversation_history=conversation_history, retry_count=retry_count, max_retries=max_retries, - retryable=True, - reason="invalid_response", - ) - # Stop spinner silently — retry status is now buffered - # and only surfaced if every retry+fallback exhausts. - if thinking_spinner: - thinking_spinner.stop("") - thinking_spinner = None - if agent.thinking_callback: - agent.thinking_callback("") - - # Invalid response — could be rate limiting, provider timeout, - # upstream server error, or malformed response. - retry_count += 1 - - # Eager fallback: empty/malformed responses often mean rate limiting - # — switch now instead of extended backoff. - if agent._fallback_index < len(agent._fallback_chain): - agent._buffer_status("⚠️ Empty/malformed response — switching to fallback...") - if agent._try_activate_fallback(): - active_system_prompt = _arm_fallback_restart( - agent, api_messages, active_system_prompt, _retry) - retry_count = 0 - compression_attempts = 0 - return _verdict("break") - - error_msg, provider_name, _failure_hint = describe_invalid_response( - agent, response, api_duration - ) - - agent._buffer_vprint(f"⚠️ Invalid API response (attempt {retry_count}/{max_retries}): {', '.join(error_details)}") - agent._buffer_vprint(f" 🏢 Provider: {provider_name}") - cleaned_provider_error = agent._clean_error_message(error_msg) - agent._buffer_vprint(f" 📝 Provider message: {cleaned_provider_error}") - agent._buffer_vprint(f" ⏱️ {_failure_hint}") - - if retry_count >= max_retries: - # Try fallback before giving up - if agent._has_pending_fallback(): - agent._buffer_status(f"⚠️ Max retries ({max_retries}) for invalid responses — trying fallback...") - if agent._try_activate_fallback(): - active_system_prompt = _arm_fallback_restart( - agent, api_messages, active_system_prompt, _retry) - retry_count = 0 - compression_attempts = 0 - return _verdict("break") - # Terminal — flush buffered retry trace so user sees what happened. - agent._flush_status_buffer() - agent._emit_status(f"❌ Max retries ({max_retries}) exceeded for invalid responses. Giving up.") - logger.error("%sInvalid API response after %d retries.", agent.log_prefix, max_retries) - agent._persist_session(messages, conversation_history) - _final_response = f"Invalid API response after {max_retries} retries: {_failure_hint}" - return _verdict("return", { - "final_response": _final_response, - "messages": messages, - "completed": False, - "api_calls": api_call_count, - "error": _final_response, - "failed": True # Mark as failure for filtering - }) - - # Backoff before retry — jittered exponential: 5s base, 120s cap - wait_time = jittered_backoff(retry_count, base_delay=5.0, max_delay=120.0) - agent._buffer_vprint(f"⏳ Retrying in {wait_time:.1f}s ({_failure_hint})...") - logger.warning("Invalid API response (retry %d/%d): %s | Provider: %s", retry_count, max_retries, ', '.join(error_details), provider_name) - - # A redirect cancels only the live request; the helper preserves the - # pending correction (restart_with_redirected_messages) instead of - # destroying it with clear_interrupt(). - _interrupted = interruptible_backoff_sleep( - agent, wait_time, _retry, - messages=messages, - conversation_history=conversation_history, + compression_attempts=compression_attempts, api_call_count=api_call_count, - abort_message="Interrupt detected during retry wait, aborting.", - interrupt_text=f"Operation interrupted during retry ({_failure_hint}, attempt {retry_count}/{max_retries}).", - activity_label=f"retry backoff ({retry_count}/{max_retries})", + api_request_id=api_request_id, + api_start_time=api_start_time, + api_duration=api_duration, + effective_task_id=effective_task_id, + turn_id=turn_id, ) - if _interrupted is not None: - return _verdict("return", _interrupted) - if _retry.restart_with_redirected_messages: - return _verdict("break") # rebuild this iteration from the correction - return _verdict("continue") # Retry the API call + thinking_spinner = _iv.thinking_spinner + active_system_prompt = _iv.active_system_prompt + retry_count = _iv.retry_count + compression_attempts = _iv.compression_attempts + if _iv.action != "fallthrough": + return _verdict(_iv.action, _iv.result) agent._turn_received_provider_response = True @@ -362,3 +290,158 @@ def check_api_response( agent._touch_activity(f"API call #{api_call_count} completed") return _verdict("break") # Success, exit retry loop return _verdict("fallthrough") + + +@dataclass +class InvalidResponseVerdict: + """``action``: ``"continue"`` (retry the API call after backoff), ``"break"`` (fallback + armed / redirect pending) or ``"return"`` (``result``: terminal invalid-response result or + interrupt during backoff). Rebinds ``thinking_spinner``/``active_system_prompt``/ + ``retry_count``/``compression_attempts``.""" + + action: str + thinking_spinner: Any + active_system_prompt: Any + retry_count: Any + compression_attempts: Any + result: Optional[Dict[str, Any]] = None + + +def retry_invalid_response( + agent: Any, + *, + response: Any, + error_details: Any, + _retry: Any, + thinking_spinner: Any, + messages: Any, + api_messages: Any, + api_kwargs: Any, + active_system_prompt: Any, + conversation_history: Any, + retry_count: Any, + max_retries: Any, + compression_attempts: Any, + api_call_count: Any, + api_request_id: Any, + api_start_time: Any, + api_duration: Any, + effective_task_id: Any, + turn_id: Any, +) -> InvalidResponseVerdict: + """Malformed/empty provider response: fire the error hook, stop the spinner, eager + fallback (empty responses often mean rate limiting), terminal result at max retries, + else jittered backoff that preserves a pending redirect.""" + from agent.conversation_loop import ( + _arm_fallback_restart, + describe_invalid_response, + interruptible_backoff_sleep, + jittered_backoff, + ) + + def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> InvalidResponseVerdict: + return InvalidResponseVerdict( + action=action, + thinking_spinner=thinking_spinner, + active_system_prompt=active_system_prompt, + retry_count=retry_count, + compression_attempts=compression_attempts, + result=result, + ) + + 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="InvalidAPIResponse", + error_message=", ".join(error_details) or "Invalid API response", + status_code=getattr(getattr(response, "error", None), "code", None), + retry_count=retry_count, + max_retries=max_retries, + retryable=True, + reason="invalid_response", + ) + # Stop spinner silently — retry status is now buffered + # and only surfaced if every retry+fallback exhausts. + if thinking_spinner: + thinking_spinner.stop("") + thinking_spinner = None + if agent.thinking_callback: + agent.thinking_callback("") + + # Invalid response — could be rate limiting, provider timeout, + # upstream server error, or malformed response. + retry_count += 1 + + # Eager fallback: empty/malformed responses often mean rate limiting + # — switch now instead of extended backoff. + if agent._fallback_index < len(agent._fallback_chain): + agent._buffer_status("⚠️ Empty/malformed response — switching to fallback...") + if agent._try_activate_fallback(): + active_system_prompt = _arm_fallback_restart( + agent, api_messages, active_system_prompt, _retry) + retry_count = 0 + compression_attempts = 0 + return _verdict("break") + + error_msg, provider_name, _failure_hint = describe_invalid_response( + agent, response, api_duration + ) + + agent._buffer_vprint(f"⚠️ Invalid API response (attempt {retry_count}/{max_retries}): {', '.join(error_details)}") + agent._buffer_vprint(f" 🏢 Provider: {provider_name}") + cleaned_provider_error = agent._clean_error_message(error_msg) + agent._buffer_vprint(f" 📝 Provider message: {cleaned_provider_error}") + agent._buffer_vprint(f" ⏱️ {_failure_hint}") + + if retry_count >= max_retries: + # Try fallback before giving up + if agent._has_pending_fallback(): + agent._buffer_status(f"⚠️ Max retries ({max_retries}) for invalid responses — trying fallback...") + if agent._try_activate_fallback(): + active_system_prompt = _arm_fallback_restart( + agent, api_messages, active_system_prompt, _retry) + retry_count = 0 + compression_attempts = 0 + return _verdict("break") + # Terminal — flush buffered retry trace so user sees what happened. + agent._flush_status_buffer() + agent._emit_status(f"❌ Max retries ({max_retries}) exceeded for invalid responses. Giving up.") + logger.error("%sInvalid API response after %d retries.", agent.log_prefix, max_retries) + agent._persist_session(messages, conversation_history) + _final_response = f"Invalid API response after {max_retries} retries: {_failure_hint}" + return _verdict("return", { + "final_response": _final_response, + "messages": messages, + "completed": False, + "api_calls": api_call_count, + "error": _final_response, + "failed": True # Mark as failure for filtering + }) + + # Backoff before retry — jittered exponential: 5s base, 120s cap + wait_time = jittered_backoff(retry_count, base_delay=5.0, max_delay=120.0) + agent._buffer_vprint(f"⏳ Retrying in {wait_time:.1f}s ({_failure_hint})...") + logger.warning("Invalid API response (retry %d/%d): %s | Provider: %s", retry_count, max_retries, ', '.join(error_details), provider_name) + + # A redirect cancels only the live request; the helper preserves the + # pending correction (restart_with_redirected_messages) instead of + # destroying it with clear_interrupt(). + _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 during retry ({_failure_hint}, attempt {retry_count}/{max_retries}).", + activity_label=f"retry backoff ({retry_count}/{max_retries})", + ) + if _interrupted is not None: + return _verdict("return", _interrupted) + if _retry.restart_with_redirected_messages: + return _verdict("break") # rebuild this iteration from the correction + return _verdict("continue") # Retry the API call + return _verdict("fallthrough") diff --git a/agent/turn_tool_round.py b/agent/turn_tool_round.py index 5d13fb5309..3eb4843e12 100644 --- a/agent/turn_tool_round.py +++ b/agent/turn_tool_round.py @@ -62,7 +62,6 @@ def run_tool_round( Hermes; a failed canonical append ends the turn rather than running tools from process-only state.""" from agent.conversation_loop import ( - _STALE_MARKER_RE, _invalid_tool_name_error_content, ) @@ -122,94 +121,14 @@ def run_tool_round( if tc.function.name not in agent.valid_tool_names ] - assistant_msg = agent._build_assistant_message(assistant_message, finish_reason) - - turn_content = assistant_message.content or "" - - # A bare bracketed token (e.g. ``[memory]``) beside a function call is - # protocol scaffolding; persisting it lets the post-tool fallback replay - # it forever (#78148). - if ( - assistant_message.tool_calls - and _STALE_MARKER_RE.fullmatch(turn_content.strip()) - ): - logger.warning( - "Discarding bare tool-call marker from assistant content: %s", - turn_content, - ) - turn_content = "" - assistant_msg["content"] = "" - - # Classify tools regardless of visible content: a substantive tool-only - # turn must invalidate any older housekeeping fallback. - _HOUSEKEEPING_TOOLS = frozenset({ - "memory", "todo_list", "skill_manage", "session_search", - }) - _all_housekeeping = all( - tc.function.name in _HOUSEKEEPING_TOOLS - for tc in assistant_message.tool_calls - ) - - # Substantive tools clear any older fallback so a two-turn-old - # housekeeping narration isn't attributed to the preceding tool turn. - if assistant_message.tool_calls and not _all_housekeeping: - agent._last_content_with_tools = None - agent._last_content_tools_all_housekeeping = False - # Also clear the mute flag a prior housekeeping turn may have set, - # else _vprint suppresses this turn's tool progress until the - # no-tool-call branch clears it. - agent._mute_post_response = False - - # Content + tool_calls in one turn: keep the content as a fallback final - # response in case the follow-up turn after tools is empty. - if turn_content and agent._has_content_after_think_block(turn_content): - agent._last_content_with_tools = turn_content - # Mute only when EVERY tool call is post-response housekeeping - # (memory, todo, skill_manage); substantive tools keep output on. - agent._last_content_tools_all_housekeeping = _all_housekeeping - if _all_housekeeping and agent._has_stream_consumers(): - agent._mute_post_response = True - elif agent._should_emit_quiet_tool_messages(): - clean = agent._strip_think_blocks(turn_content).strip() - if clean: - agent._vprint(f" ┊ 💬 {clean}") - - # Pop thinking-only prefill message(s) before appending - # (tool-call path — same rationale as the final-response path). - _had_prefill = False - while ( - messages - and isinstance(messages[-1], dict) - and messages[-1].get("_thinking_prefill") - ): - messages.pop() - _had_prefill = True - - # Tool calls after a prefill recovery reset the prefill counter, so - # each tool-call success is a fresh start, not a cumulative burn. - if _had_prefill: - agent._thinking_prefill_retries = 0 - agent._empty_content_retries = 0 - # Re-arm the post-tool nudge so it can fire on a LATER tool round. - agent._post_tool_empty_retried = False - # A landed tool call recovers any dropped-tool-call stall; refresh that - # budget so it guards each stall independently, not the whole run. - agent._dropped_toolcall_retries = 0 - - previous_msg = messages[-1] if messages else None - current_interim_visible = agent._interim_assistant_visible_text(assistant_msg) - previous_interim_visible = ( - agent._interim_assistant_visible_text(previous_msg) - if isinstance(previous_msg, dict) - else "" - ) - duplicate_previous_interim = ( - bool(current_interim_visible) - and isinstance(previous_msg, dict) - and previous_msg.get("role") == "assistant" - and previous_msg.get("finish_reason") == "incomplete" - and previous_interim_visible == current_interim_visible + _st = stage_tool_call_message( + agent, + assistant_message=assistant_message, + finish_reason=finish_reason, + messages=messages, ) + assistant_msg = _st.assistant_msg + duplicate_previous_interim = _st.duplicate_previous_interim append_message(messages, assistant_msg) # Mixed batch: error-result invalid calls and drop them from execution. @@ -348,3 +267,128 @@ def run_tool_round( # Continue loop for next response return _verdict("continue") return _verdict("fallthrough") + + +@dataclass +class StagedToolCallMessage: + """Always ``action == "fallthrough"``. ``assistant_msg`` is the transcript row to append; + ``duplicate_previous_interim`` suppresses re-emitting interim commentary the previous + ``incomplete`` row already showed.""" + + action: str + assistant_msg: Any + duplicate_previous_interim: Any + + +def stage_tool_call_message( + agent: Any, + *, + assistant_message: Any, + finish_reason: Any, + messages: Any, +) -> StagedToolCallMessage: + """Build the assistant tool-call row and update the per-turn fallback/mute state: drop a bare + bracketed marker beside a call (#78148), classify housekeeping-only rounds, keep visible + content as the empty-follow-up fallback, pop thinking-only prefills (resetting their + counters), re-arm the post-tool nudge and the dropped-tool-call stall budget.""" + from agent.conversation_loop import ( + _STALE_MARKER_RE, + ) + + def _verdict(action: str, result: Optional[Dict[str, Any]] = None) -> StagedToolCallMessage: + return StagedToolCallMessage( + action=action, + assistant_msg=assistant_msg, + duplicate_previous_interim=duplicate_previous_interim, + + ) + + assistant_msg = agent._build_assistant_message(assistant_message, finish_reason) + + turn_content = assistant_message.content or "" + + # A bare bracketed token (e.g. ``[memory]``) beside a function call is + # protocol scaffolding; persisting it lets the post-tool fallback replay + # it forever (#78148). + if ( + assistant_message.tool_calls + and _STALE_MARKER_RE.fullmatch(turn_content.strip()) + ): + logger.warning( + "Discarding bare tool-call marker from assistant content: %s", + turn_content, + ) + turn_content = "" + assistant_msg["content"] = "" + + # Classify tools regardless of visible content: a substantive tool-only + # turn must invalidate any older housekeeping fallback. + _HOUSEKEEPING_TOOLS = frozenset({ + "memory", "todo_list", "skill_manage", "session_search", + }) + _all_housekeeping = all( + tc.function.name in _HOUSEKEEPING_TOOLS + for tc in assistant_message.tool_calls + ) + + # Substantive tools clear any older fallback so a two-turn-old + # housekeeping narration isn't attributed to the preceding tool turn. + if assistant_message.tool_calls and not _all_housekeeping: + agent._last_content_with_tools = None + agent._last_content_tools_all_housekeeping = False + # Also clear the mute flag a prior housekeeping turn may have set, + # else _vprint suppresses this turn's tool progress until the + # no-tool-call branch clears it. + agent._mute_post_response = False + + # Content + tool_calls in one turn: keep the content as a fallback final + # response in case the follow-up turn after tools is empty. + if turn_content and agent._has_content_after_think_block(turn_content): + agent._last_content_with_tools = turn_content + # Mute only when EVERY tool call is post-response housekeeping + # (memory, todo, skill_manage); substantive tools keep output on. + agent._last_content_tools_all_housekeeping = _all_housekeeping + if _all_housekeeping and agent._has_stream_consumers(): + agent._mute_post_response = True + elif agent._should_emit_quiet_tool_messages(): + clean = agent._strip_think_blocks(turn_content).strip() + if clean: + agent._vprint(f" ┊ 💬 {clean}") + + # Pop thinking-only prefill message(s) before appending + # (tool-call path — same rationale as the final-response path). + _had_prefill = False + while ( + messages + and isinstance(messages[-1], dict) + and messages[-1].get("_thinking_prefill") + ): + messages.pop() + _had_prefill = True + + # Tool calls after a prefill recovery reset the prefill counter, so + # each tool-call success is a fresh start, not a cumulative burn. + if _had_prefill: + agent._thinking_prefill_retries = 0 + agent._empty_content_retries = 0 + # Re-arm the post-tool nudge so it can fire on a LATER tool round. + agent._post_tool_empty_retried = False + # A landed tool call recovers any dropped-tool-call stall; refresh that + # budget so it guards each stall independently, not the whole run. + agent._dropped_toolcall_retries = 0 + + previous_msg = messages[-1] if messages else None + current_interim_visible = agent._interim_assistant_visible_text(assistant_msg) + previous_interim_visible = ( + agent._interim_assistant_visible_text(previous_msg) + if isinstance(previous_msg, dict) + else "" + ) + duplicate_previous_interim = ( + bool(current_interim_visible) + and isinstance(previous_msg, dict) + and previous_msg.get("role") == "assistant" + and previous_msg.get("finish_reason") == "incomplete" + and previous_interim_visible == current_interim_visible + ) + return _verdict("fallthrough")