fix: widen stale circuit breaker to non-streaming path + all provider-swap resets

Review findings on the salvaged #60332 breaker, fixed as follow-ups:

- restore_primary_runtime() now resets the streak (third provider-swap
  path; without it a recovered primary was short-circuited before a
  single attempt and could never be re-proven healthy except via /model).
- interruptible_api_call (non-streaming) now carries the same breaker
  (guard at entry, bump on stale_call_kill, reset on success). Quiet-mode
  / subagent / headless sessions — the profile most like #58962's
  unattended 494-failure session — take this path and had the identical
  infinite stale-retry class.
- Partial-stream stub return now resets the streak (chunks were received,
  provider demonstrably responsive).
- Consolidated the triple-duplicated counter arithmetic into shared
  helpers (_stale_streak/_bump_stale_streak/_reset_stale_streak/
  _check_stale_giveup) with one canonical comment block; error message
  now says 'consecutive stale attempts' (the counter counts kills, not
  turns — a single turn can produce several).

4 new tests (restore resets / no-op restore keeps latch / non-streaming
short-circuit / non-streaming success reset).

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

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 ──────────────────────────────────────────────────

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

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. |