diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index 1c49a59e2a3c0..d766b4c269b91 100644 --- a/agent/chat_completion_helpers.py +++ b/agent/chat_completion_helpers.py @@ -978,6 +978,11 @@ def _call(): agent._touch_activity( f"codex stream killed after {int(_elapsed)}s with no first byte" ) + # This is a stale-derived timeout just like the generic stale-call + # kill below. Record its provenance before returning to the retry + # loop so bounded eager fallback and the stale give-up breaker + # continue to apply to Codex TTFB stalls. + _bump_stale_streak(agent) # Wait briefly for the worker to notice the closed connection. t.join(timeout=2.0) if result["error"] is None and result["response"] is None: @@ -1024,6 +1029,9 @@ def _call(): agent._touch_activity( f"codex stream killed after {int(_event_stale_elapsed)}s with no SSE events" ) + # Preserve stale-timeout provenance for the retry/fallback gate + # and the cross-turn stale-stream circuit breaker. + _bump_stale_streak(agent) t.join(timeout=2.0) if result["error"] is None and result["response"] is None: result["error"] = TimeoutError( diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index 3ca96898cf997..658e29ea7d878 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -4248,21 +4248,28 @@ def _perform_api_call(next_api_kwargs): # Fall through to normal error handling if compression # is exhausted or didn't help. - # Eager fallback for rate-limit errors (429 or quota exhaustion) - # and transport errors (connection failure / timeout / provider - # overloaded). Rate limits and billing: switch immediately — - # the primary provider won't recover within the retry window. - # Transport errors: allow 1 retry first (transient hiccups - # recover), then fall back if the provider is truly unreachable. + # Eager fallback for rate-limit errors (429 or quota exhaustion), + # provider overload, and stale-stream failures. Ordinary fast + # connection errors deliberately complete the normal retry cycle + # so the primary-client recovery path below can rebuild the + # transport and grant one fresh retry cycle before fallback. + # Stale streams remain eager after one retry: each attempt can + # consume the full multi-minute stale timeout (#22277). is_rate_limited = classified.reason in { FailoverReason.rate_limit, FailoverReason.billing, FailoverReason.upstream_rate_limit, } - _is_transport_failure = classified.reason in { - FailoverReason.timeout, - FailoverReason.overloaded, - } + _is_overload_failure = ( + classified.reason == FailoverReason.overloaded + ) + _is_stale_timeout = ( + classified.reason == FailoverReason.timeout + and getattr(agent, "_consecutive_stale_streams", 0) > 0 + ) + _is_eager_transport_failure = ( + _is_overload_failure or _is_stale_timeout + ) # Z.AI Coding Plan GLM-5.2 overload 429s classify as # `overloaded` (to spare the credential pool), but `overloaded` # is excluded from `is_rate_limited` — the gate for the adaptive @@ -4277,7 +4284,7 @@ def _perform_api_call(next_api_kwargs): max_retries = max(max_retries, zai_coding_overload_retry_ceiling()) _should_fallback = ( is_rate_limited - or (_is_transport_failure and retry_count >= 2) + or (_is_eager_transport_failure and retry_count >= 2) ) if _should_fallback and agent._fallback_index < len(agent._fallback_chain): # Don't eagerly fallback if credential pool rotation may @@ -4308,7 +4315,7 @@ def _perform_api_call(next_api_kwargs): agent._buffer_status( "⚠️ Billing or credits exhausted — switching to fallback provider..." ) - elif _is_transport_failure: + elif _is_eager_transport_failure: agent._buffer_status( "⚠️ Provider unreachable — switching to fallback provider..." ) diff --git a/tests/agent/test_codex_ttfb_watchdog.py b/tests/agent/test_codex_ttfb_watchdog.py index 66208a8e1a896..becd45f06cd8f 100644 --- a/tests/agent/test_codex_ttfb_watchdog.py +++ b/tests/agent/test_codex_ttfb_watchdog.py @@ -57,10 +57,6 @@ def _make_codex_agent(tmp_path, monkeypatch): return agent - - - - def test_ttfb_includes_silent_hang_hint_for_gpt_5_5(tmp_path, monkeypatch): """The no-first-byte watchdog should surface the same actionable hint as the stale-call timeout path when the model matches the silent-hang heuristic.""" @@ -102,14 +98,12 @@ def fake_hang(api_kwargs, client=None, on_first_delta=None): assert "gpt-5.3-codex" in message assert "gpt-5.4-codex" in message assert "codex_ttfb_kill" in closes + assert agent._consecutive_stale_streams == 1 assert statuses, "expected a user-facing watchdog status" assert any("gpt-5.4" in s and "gpt-5.3-codex" in s for s in statuses) finally: stop["flag"] = True - - - def test_ttfb_does_not_kill_when_events_flow(tmp_path, monkeypatch): """Once a stream event has arrived, a generation that runs past the TTFB cutoff is NOT killed by the watchdog — it completes normally.""" @@ -148,11 +142,48 @@ def fake_stream(api_kwargs, client=None, on_first_delta=None): assert "codex_ttfb_kill" not in closes +def test_event_idle_kill_records_stale_provenance(tmp_path, monkeypatch): + """The idle watchdog preserves stale provenance for bounded fallback.""" + from agent import chat_completion_helpers as h + + agent = _make_codex_agent(tmp_path, monkeypatch) + monkeypatch.setenv("HERMES_CODEX_TTFB_TIMEOUT_SECONDS", "10") + monkeypatch.setenv("HERMES_CODEX_EVENT_STALE_TIMEOUT_SECONDS", "0.4") + closes: list = [] + dummy_client = SimpleNamespace() + monkeypatch.setattr(agent, "_create_request_openai_client", lambda **k: dummy_client) + monkeypatch.setattr( + agent, + "_abort_request_openai_client", + lambda c, reason=None: closes.append(reason), + ) + monkeypatch.setattr( + agent, + "_close_request_openai_client", + lambda c, reason=None: closes.append(reason), + ) + stop = {"flag": False} + def fake_stream(api_kwargs, client=None, on_first_delta=None): + agent._codex_stream_last_event_ts = time.time() + deadline = time.time() + 30 + while time.time() < deadline and not stop["flag"] and not agent._interrupt_requested: + time.sleep(0.02) + raise RuntimeError("connection closed") + monkeypatch.setattr(agent, "_run_codex_stream", fake_stream) + try: + with pytest.raises(TimeoutError) as excinfo: + h.interruptible_api_call(agent, {"model": "gpt-5.5", "input": "hi"}) + assert "after first byte" in str(excinfo.value) + assert "codex_stream_idle_kill" in closes + assert "codex_ttfb_kill" not in closes + assert agent._consecutive_stale_streams == 1 + finally: + stop["flag"] = True @pytest.mark.parametrize( "stale_timeout", @@ -345,7 +376,3 @@ def fake_hang(api_kwargs, client=None, on_first_delta=None): assert "with no response" in str(excinfo.value) finally: stop["flag"] = True - - - - diff --git a/tests/run_agent/test_fast_transport_recovery_before_fallback.py b/tests/run_agent/test_fast_transport_recovery_before_fallback.py new file mode 100644 index 0000000000000..3abcaa7ada55b --- /dev/null +++ b/tests/run_agent/test_fast_transport_recovery_before_fallback.py @@ -0,0 +1,161 @@ +"""Regression coverage for #69186. + +Fast SDK/socket connection errors should reach the existing primary-client +recovery cycle before fallback. Stale-stream failures remain eligible for +bounded eager fallback so #22277 does not regress. +""" + +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +from run_agent import AIAgent + + +class APIConnectionError(Exception): + pass + + +def _tool_defs(): + return [ + { + "type": "function", + "function": { + "name": "web_search", + "description": "search", + "parameters": {"type": "object", "properties": {}}, + }, + } + ] + + +def _response(content: str): + message = SimpleNamespace(content=content, tool_calls=None) + choice = SimpleNamespace(message=message, finish_reason="stop") + return SimpleNamespace(choices=[choice], model="test-model", usage=None) + + +def _agent(): + fallback_chain = [ + { + "provider": "deepseek", + "model": "deepseek-chat", + "base_url": "https://api.deepseek.com", + } + ] + with ( + patch("run_agent.get_tool_definitions", return_value=_tool_defs()), + patch("run_agent.check_toolset_requirements", return_value={}), + patch("run_agent.OpenAI", return_value=MagicMock()), + ): + agent = AIAgent( + api_key="primary-key-abcdef12", + base_url="https://primary.example.com/v1", + provider="custom", + model="primary-model", + quiet_mode=True, + skip_context_files=True, + skip_memory=True, + fallback_model=fallback_chain, + ) + agent.client = MagicMock() + agent._api_max_retries = 3 + return agent + + +def _fallback_client(): + client = MagicMock() + client.api_key = "fallback-key-abcdef12" + client.base_url = "https://api.deepseek.com" + client._custom_headers = None + client.default_headers = None + return client + + +def _run(agent, api_call, *, fallback_client=None): + patches = [ + patch.object(agent, "_interruptible_api_call", side_effect=api_call), + patch.object(agent, "_persist_session"), + patch.object(agent, "_save_trajectory"), + patch.object(agent, "_cleanup_task_resources"), + patch("run_agent.OpenAI", return_value=MagicMock()), + patch("agent.agent_runtime_helpers.time.sleep"), + patch("agent.conversation_loop.time.sleep"), + patch("agent.model_metadata.get_model_context_length", return_value=200000), + ] + if fallback_client is not None: + patches.extend([ + patch( + "agent.auxiliary_client.resolve_provider_client", + return_value=(fallback_client, "deepseek-chat"), + ), + patch( + "hermes_cli.model_normalize.normalize_model_for_provider", + side_effect=lambda model, provider: model, + ), + ]) + + entered = [] + try: + for ctx in patches: + entered.append(ctx) + ctx.__enter__() + return agent.run_conversation("hello") + finally: + for ctx in reversed(entered): + ctx.__exit__(None, None, None) + + +def test_fast_connection_errors_recover_primary_before_fallback(): + """Three fast failures exhaust retries, then a rebuilt primary succeeds.""" + agent = _agent() + calls = [] + + def api_call(_kwargs): + calls.append((agent.provider, agent.model)) + if len(calls) <= 3: + raise APIConnectionError("Connection error.") + return _response("Recovered on rebuilt primary") + + with ( + patch.object( + agent, + "_try_recover_primary_transport", + wraps=agent._try_recover_primary_transport, + ) as recover, + patch.object( + agent, + "_try_activate_fallback", + wraps=agent._try_activate_fallback, + ) as fallback, + ): + result = _run(agent, api_call, fallback_client=_fallback_client()) + + assert result["completed"] is True + assert result["final_response"] == "Recovered on rebuilt primary" + assert calls == [("custom", "primary-model")] * 4 + recover.assert_called_once() + fallback.assert_not_called() + + +def test_stale_stream_timeout_keeps_bounded_eager_fallback(): + """A stale-detector-derived timeout still falls back after one retry.""" + agent = _agent() + agent._consecutive_stale_streams = 1 + calls = [] + + def api_call(_kwargs): + calls.append((agent.provider, agent.model)) + if agent.provider == "custom": + raise APIConnectionError("Connection closed after stale-stream kill.") + return _response("Recovered via fallback") + + result = _run(agent, api_call, fallback_client=_fallback_client()) + + assert result["completed"] is True + assert result["final_response"] == "Recovered via fallback" + assert calls == [ + ("custom", "primary-model"), + ("custom", "primary-model"), + ("deepseek", "deepseek-chat"), + ] + assert agent._fallback_activated is True