diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py index 875bc17d5c8..cbccc683da5 100644 --- a/agent/agent_runtime_helpers.py +++ b/agent/agent_runtime_helpers.py @@ -1275,6 +1275,12 @@ def restore_primary_runtime(agent) -> bool: agent._fallback_activated = False agent._fallback_index = 0 + # Reset the stale-call circuit breaker (#58962): the streak measured + # the FALLBACK provider we're leaving; the restored primary deserves + # a fresh stream attempt before the breaker can trip again. + from agent.chat_completion_helpers import _reset_stale_streak + _reset_stale_streak(agent) + # Undo the fallback's identity rewrite so the prompt is # byte-identical to the stored copy again (prefix cache match). from agent.chat_completion_helpers import rewrite_prompt_model_identity @@ -1992,12 +1998,13 @@ def switch_model(agent, new_model, new_provider, api_key='', base_url='', api_mo # ── Invalidate cached system prompt so it rebuilds next turn ── agent._cached_system_prompt = None - # ── Reset the cross-turn stream-stale circuit breaker (#58962) ── + # ── Reset the cross-turn stale-call circuit breaker (#58962) ── # The breaker's error text tells the user to "switch models ... then # retry"; without this reset the streak stays latched and the freshly # selected (healthy) provider would keep short-circuiting before any # stream is even attempted. - agent._consecutive_stale_streams = 0 + from agent.chat_completion_helpers import _reset_stale_streak + _reset_stale_streak(agent) # ── Update _primary_runtime so the change persists across turns ── _cc = agent.context_compressor if hasattr(agent, "context_compressor") and agent.context_compressor else None diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index 1820f32caed..9be0a7fed52 100644 --- a/agent/chat_completion_helpers.py +++ b/agent/chat_completion_helpers.py @@ -171,6 +171,52 @@ def _env_float(name: str, default: float) -> float: return default +# ── Cross-turn stale-call circuit breaker (#58962) ───────────────────── +# A session wedged against an unresponsive provider hits the stale detector +# on every call and loops forever (observed: 494 consecutive failures over +# 3+ days, each burning the full stale timeout × retries with no response). +# The agent carries ``_consecutive_stale_streams``: incremented on every +# stale kill, reset only when a call actually completes (or when the +# provider is swapped — switch_model / try_activate_fallback / +# restore_primary_runtime — since the streak measured the OLD provider). +# Past the give-up threshold, calls abort immediately with an actionable +# error instead of re-waiting out the stale timeout. + +def _stale_streak(agent) -> int: + try: + return int(getattr(agent, "_consecutive_stale_streams", 0) or 0) + except Exception: + return 0 + + +def _bump_stale_streak(agent) -> None: + try: + agent._consecutive_stale_streams = _stale_streak(agent) + 1 + except Exception: + pass + + +def _reset_stale_streak(agent) -> None: + try: + agent._consecutive_stale_streams = 0 + except Exception: + pass + + +def _check_stale_giveup(agent) -> None: + """Raise immediately when the consecutive-stale streak is past the + give-up threshold — no network attempt, no stale-timeout wait.""" + _giveup = env_int("HERMES_STREAM_STALE_GIVEUP", 5) + _streak = _stale_streak(agent) + if _giveup > 0 and _streak >= _giveup: + raise RuntimeError( + "Provider has been unresponsive (no response received) for " + f"{_streak} consecutive stale attempts — aborting this call to " + "avoid an indefinite stall. Switch models or start a new " + "session, then retry." + ) + + def interruptible_api_call(agent, api_kwargs: dict): """ Run the API call in a background thread so the main conversation loop @@ -186,6 +232,13 @@ def interruptible_api_call(agent, api_kwargs: dict): provider fallback. """ result = {"response": None, "error": None} + + # Cross-turn stale-call circuit breaker (#58962) — non-streaming sibling + # of the guard in interruptible_streaming_api_call. Quiet-mode / + # subagent / no-stream-consumer sessions take THIS path, and a wedged + # unattended session here has the same infinite stale-retry class. + _check_stale_giveup(agent) + request_client_holder = {"client": None, "owner_tid": None} request_client_lock = threading.Lock() # Request-local cancellation flag. Distinct from agent._interrupt_requested @@ -557,6 +610,9 @@ def interruptible_api_call(agent, api_kwargs: dict): _close_request_client_once("stale_call_kill") except Exception: pass + # Circuit breaker (#58962): count the stale kill. See the + # canonical comment block above ``_stale_streak()``. + _bump_stale_streak(agent) agent._touch_activity( f"stale non-streaming call killed after {int(_elapsed)}s" ) @@ -599,6 +655,10 @@ def interruptible_api_call(agent, api_kwargs: dict): raise InterruptedError("Agent interrupted during API call") if result["error"] is not None: raise result["error"] + # Success — clear the circuit breaker (#58962): the provider proved + # responsive. See the canonical comment block above ``_stale_streak()``. + if result["response"] is not None: + _reset_stale_streak(agent) return result["response"] @@ -1488,11 +1548,11 @@ def try_activate_fallback(agent, reason: "FailoverReason | None" = None) -> bool "Fallback activated: %s → %s (%s)", old_model, fb_model, fb_provider, ) - # Reset the cross-turn stream-stale circuit breaker (#58962): the - # streak measured the OLD provider's unresponsiveness. Carrying it - # over would short-circuit the freshly activated fallback before it - # gets a single stream attempt. - agent._consecutive_stale_streams = 0 + # Reset the stale-call circuit breaker (#58962): the streak measured + # the OLD provider's unresponsiveness. Carrying it over would + # short-circuit the freshly activated fallback before it gets a + # single stream attempt. + _reset_stale_streak(agent) return True except Exception as e: if fb_provider == "nous": @@ -1902,29 +1962,10 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta= result = {"response": None, "error": None, "partial_tool_names": []} - # ── Cross-turn stream-stale circuit breaker (#58962) ─────────────── - # A session wedged against an unresponsive provider hits the stale-stream - # detector on every turn and loops forever (observed: 494 consecutive - # failures over 3+ days, every turn burning the full 180s×retries with - # no response). After N consecutive turns end in a stale-stream kill, - # stop spending 180s×retries each turn and surface a clear, actionable - # error instead. The streak resets only when a stream actually - # completes (see the success return below), so a recoverable provider - # resumes normally while a permanently-broken session stops silently - # consuming the gateway. - _stale_giveup = env_int("HERMES_STREAM_STALE_GIVEUP", 5) - _stale_streak = 0 - try: - _stale_streak = int(getattr(agent, "_consecutive_stale_streams", 0) or 0) - except Exception: - pass - if _stale_giveup > 0 and _stale_streak >= _stale_giveup: - raise RuntimeError( - "Provider has been unresponsive (no stream chunks received) for " - f"{_stale_streak} consecutive turns — aborting this turn to avoid " - "an indefinite stall. Switch models or start a new session, then " - "retry." - ) + # Cross-turn stale-stream circuit breaker (#58962) — see the canonical + # comment block above ``_stale_streak()``. Raises past the give-up + # threshold instead of burning another stale-timeout×retries cycle. + _check_stale_giveup(agent) request_client_holder = {"client": None, "diag": None, "owner_tid": None} request_client_lock = threading.Lock() @@ -2903,17 +2944,9 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta= _close_request_client_once("stale_stream_kill") except Exception: pass - # Cross-turn circuit breaker (#58962): count consecutive - # stale-stream kills so a wedged session eventually stops - # re-attempting every turn. Reset only on a successful stream - # (see the success return). Wrapped so a weird agent object - # can never break the stale detector. - try: - agent._consecutive_stale_streams = ( - int(getattr(agent, "_consecutive_stale_streams", 0) or 0) + 1 - ) - except Exception: - pass + # Circuit breaker (#58962): count the stale kill. See the + # canonical comment block above ``_stale_streak()``. + _bump_stale_streak(agent) # Rebuild the primary client too — its connection pool # may hold dead sockets from the same provider outage. if agent.api_mode == "anthropic_messages": @@ -3043,17 +3076,16 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta= ) if _content_filter_terminated: _stub._content_filter_terminated = True + # Partial-stream stub: chunks WERE received (deltas fired), so + # the provider is demonstrably responsive — clear the circuit + # breaker (#58962) just like the full-success return below. + _reset_stale_streak(agent) return _stub raise result["error"] - # Success — clear the cross-turn stream-stale circuit breaker (#58962). - # Only a stream that actually completed resets the streak, so a wedged - # session keeps short-circuiting (clear error each turn) until a real - # response arrives, then resumes normally. - try: - if result["response"] is not None: - agent._consecutive_stale_streams = 0 - except Exception: - pass + # Success — clear the circuit breaker (#58962): the provider proved + # responsive. See the canonical comment block above ``_stale_streak()``. + if result["response"] is not None: + _reset_stale_streak(agent) return result["response"] # ── Provider fallback ────────────────────────────────────────────────── diff --git a/tests/run_agent/test_stream_stale_breaker_reset.py b/tests/run_agent/test_stream_stale_breaker_reset.py index 4df83e8ac9f..379821808df 100644 --- a/tests/run_agent/test_stream_stale_breaker_reset.py +++ b/tests/run_agent/test_stream_stale_breaker_reset.py @@ -10,10 +10,18 @@ provider would keep short-circuiting forever: - ``switch_model()`` (user-initiated /model swap) - ``try_activate_fallback()`` (automatic provider fallback) +- ``restore_primary_runtime()`` (turn-start restore back to the primary) + +The non-streaming sibling ``interruptible_api_call`` shares the same +breaker (guard at entry, bump on stale_call_kill, reset on success) — +quiet-mode / subagent sessions take that path and had the identical +infinite stale-retry class. """ from unittest.mock import MagicMock, patch +import pytest + from run_agent import AIAgent @@ -145,3 +153,64 @@ def test_fallback_exhaustion_keeps_stale_streak(): assert agent._try_activate_fallback() is False assert agent._consecutive_stale_streams == 7 + + +def test_restore_primary_runtime_resets_stale_streak(): + """Turn-start restore back to the primary is the third provider-swap + path: the streak measured the fallback we're leaving, so the restored + primary must get a fresh attempt instead of an instant short-circuit.""" + fbs = [{"provider": "openai", "model": "gpt-4o"}] + agent = _make_fallback_agent(fallback_model=fbs) + + with patch( + "agent.auxiliary_client.resolve_provider_client", + return_value=(_mock_client(), "resolved"), + ): + assert agent._try_activate_fallback() is True + + # Streak accumulated while wedged on the FALLBACK provider. + agent._consecutive_stale_streams = 7 + + with patch("run_agent.OpenAI", return_value=MagicMock()): + assert agent._restore_primary_runtime() is True + + assert agent._consecutive_stale_streams == 0 + + +def test_no_fallback_restore_noop_keeps_stale_streak(): + """When no fallback was activated, restore is a no-op (returns False) + and must NOT clear the streak — the session never left the wedged + primary, so the breaker's cross-turn latch has to survive turn starts.""" + agent = _make_fallback_agent(fallback_model=[]) + agent._consecutive_stale_streams = 7 + + assert agent._restore_primary_runtime() is False + assert agent._consecutive_stale_streams == 7 + + +class TestNonStreamingSibling: + """interruptible_api_call carries the same breaker (#58962).""" + + def test_non_streaming_short_circuits_at_threshold(self, monkeypatch): + monkeypatch.setenv("HERMES_STREAM_STALE_GIVEUP", "3") + agent = _make_fallback_agent(fallback_model=[]) + agent._consecutive_stale_streams = 3 + + with pytest.raises(RuntimeError, match="unresponsive"): + agent._interruptible_api_call({}) + + # The client is never touched on the short-circuit path. + agent.client.chat.completions.create.assert_not_called() + assert agent._consecutive_stale_streams == 3 + + def test_non_streaming_success_resets_streak(self, monkeypatch): + monkeypatch.setenv("HERMES_STREAM_STALE_GIVEUP", "3") + agent = _make_fallback_agent(fallback_model=[]) + agent._consecutive_stale_streams = 2 # below threshold + agent.client.chat.completions.create.return_value = MagicMock( + name="resp", choices=[MagicMock()] + ) + + resp = agent._interruptible_api_call({"model": "m", "messages": []}) + assert resp is not None + assert agent._consecutive_stale_streams == 0 diff --git a/website/docs/reference/environment-variables.md b/website/docs/reference/environment-variables.md index c2846ea38b1..b7df6823837 100644 --- a/website/docs/reference/environment-variables.md +++ b/website/docs/reference/environment-variables.md @@ -731,7 +731,7 @@ Advanced per-platform knobs for throttling the outbound message batcher. Most us | `HERMES_STREAM_READ_TIMEOUT` | Streaming socket read timeout in seconds (default: `120`). Auto-increased to `HERMES_API_TIMEOUT` for local providers. Increase if local LLMs time out during long code generation. | | `HERMES_STREAM_STALE_TIMEOUT` | Stale stream detection timeout in seconds (default: `180`). Auto-disabled for local providers. Triggers connection kill if no chunks arrive within this window. | | `HERMES_STREAM_RETRIES` | Number of mid-stream reconnect attempts on transient network errors (default: `3`). | -| `HERMES_STREAM_STALE_GIVEUP` | Cross-turn circuit breaker: after this many consecutive stale-stream kills with no completed stream, abort each turn immediately with an actionable error instead of re-waiting out the stale timeout (default: `5`, `0` disables). Resets on a completed stream, `/model` switch, or fallback activation. | +| `HERMES_STREAM_STALE_GIVEUP` | Cross-turn circuit breaker: after this many consecutive stale kills (streaming or non-streaming) with no completed response, abort each call immediately with an actionable error instead of re-waiting out the stale timeout (default: `5`, `0` disables). Resets on any completed response, `/model` switch, fallback activation, or turn-start primary restore. | | `HERMES_AGENT_TIMEOUT` | Gateway inactivity timeout for a running agent in seconds (default: `1800`, 30 minutes). Resets on every tool call and streamed token. Set to `0` to disable. | | `HERMES_AGENT_TIMEOUT_WARNING` | Gateway: send a warning message after this many seconds of inactivity (default: 75% of `HERMES_AGENT_TIMEOUT`). | | `HERMES_AGENT_NOTIFY_INTERVAL` | Gateway: interval in seconds between progress notifications on long-running agent turns. |