fix(agent): add cross-turn stream-stale circuit breaker (#58962)

A session wedged against an unresponsive OpenAI-compatible provider can hit the stale-stream detector on every turn and loop forever, burning the full 180s x retries each turn with no response. Issue #58962 reports 494 consecutive failures over 3+ days on a single session.

The streaming retry path already caps retries WITHIN a turn (HERMES_STREAM_RETRIES, default 2) but has no cross-turn cap. Once a session's conversation state makes every turn stale, it retries indefinitely across turns and never notifies the user.

Add a per-session consecutive-stale-stream counter on the agent:
- incremented on every stale-stream kill in the outer poll loop;
- reset to 0 only when a stream actually completes;
- when it reaches HERMES_STREAM_STALE_GIVEUP (default 5), the next turn aborts immediately with a clear, actionable RuntimeError instead of spending 180s x retries again.

This is distinct from the existing stale-stream work (local-provider hard ceiling #44938, backoff/parse-error #60031): those bound a single hung stream, while this bounds repeated cross-turn staleness and surfaces a user-visible error.

Adds tests/run_agent/test_stream_stale_circuit_breaker.py covering the short-circuit, the success-reset, and the increment.

agent/chat_completion_helpers.py

@@ -1896,6 +1896,31 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
         return result["response"]
 
     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."
+        )
+
     request_client_holder = {"client": None, "diag": None, "owner_tid": None}
     request_client_lock = threading.Lock()
     # Request-local cancellation flag — see interruptible_api_call for the full
@@ -2873,6 +2898,17 @@ 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
             # Rebuild the primary client too — its connection pool
             # may hold dead sockets from the same provider outage.
             if agent.api_mode == "anthropic_messages":
@@ -3004,6 +3040,15 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
                 _stub._content_filter_terminated = True
             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
     return result["response"]
 
 # ── Provider fallback ──────────────────────────────────────────────────

tests/run_agent/test_stream_stale_circuit_breaker.py

@@ -0,0 +1,125 @@
+"""Cross-turn stream-stale circuit breaker (issue #58962).
+
+A session wedged against an unresponsive provider can hit the stale-stream
+detector on every turn and loop forever, burning the full 180s×retries each
+turn with no response (observed: 494 consecutive failures over 3+ days).
+
+These tests cover the guard added to ``interruptible_streaming_api_call``:
+
+- a session that has already tripped the consecutive-stale threshold short
+  circuits immediately (no network attempt, no 180s wait) with a clear error;
+- a successful stream resets the consecutive-stale streak;
+- a stale-stream kill increments the consecutive-stale streak.
+
+The harness mirrors tests/run_agent/test_28161_anthropic_stream_pool_cleanup.py.
+"""
+
+import threading
+
+import httpx
+import pytest
+from unittest.mock import MagicMock
+
+from types import SimpleNamespace
+
+
+def _make_anthropic_agent(**kwargs):
+    from run_agent import AIAgent
+
+    defaults = dict(
+        api_key="test-key",
+        base_url="https://example.com/v1",
+        model="claude-opus-4-7",
+        quiet_mode=True,
+        skip_context_files=True,
+        skip_memory=True,
+    )
+    defaults.update(kwargs)
+    agent = AIAgent(**defaults)
+    agent.api_mode = "anthropic_messages"
+    agent._anthropic_client = MagicMock()
+    agent._anthropic_api_key = "test-anthropic-key"
+    return agent
+
+
+def _good_stream_cm():
+    """Context manager whose stream yields no events and returns a valid message."""
+    cm = MagicMock()
+    stream = MagicMock()
+    stream.__iter__ = MagicMock(return_value=iter([]))
+    msg = MagicMock()
+    msg.content = []
+    msg.stop_reason = "end_turn"
+    msg.usage = SimpleNamespace(input_tokens=10, output_tokens=5)
+    stream.get_final_message = MagicMock(return_value=msg)
+    cm.__enter__ = MagicMock(return_value=stream)
+    cm.__exit__ = MagicMock(return_value=False)
+    return cm
+
+
+class TestStreamStaleCircuitBreaker:
+    @pytest.mark.filterwarnings("ignore::pytest.PytestUnhandledThreadExceptionWarning")
+    def test_short_circuits_when_streak_at_threshold(self, monkeypatch):
+        """A session already past the consecutive-stale threshold must abort
+        immediately without opening a stream or waiting out the stale timeout."""
+        monkeypatch.setenv("HERMES_STREAM_STALE_GIVEUP", "3")
+
+        agent = _make_anthropic_agent()
+        agent._consecutive_stale_streams = 3  # simulate prior wedged turns
+
+        # The stream must never be opened on the short-circuit path.
+        with pytest.raises(RuntimeError, match="unresponsive"):
+            agent._interruptible_streaming_api_call({})
+
+        agent._anthropic_client.messages.stream.assert_not_called()
+        # The streak is NOT reset on the short-circuit so subsequent turns
+        # keep failing fast instead of re-attempting forever.
+        assert agent._consecutive_stale_streams == 3
+
+    @pytest.mark.filterwarnings("ignore::pytest.PytestUnhandledThreadExceptionWarning")
+    def test_success_resets_streak(self, monkeypatch):
+        """A stream that completes successfully clears the consecutive-stale
+        streak so a recovered provider resumes normally."""
+        monkeypatch.setenv("HERMES_STREAM_STALE_GIVEUP", "3")
+
+        agent = _make_anthropic_agent()
+        agent._consecutive_stale_streams = 2  # below the giveup=3 threshold
+        agent._anthropic_client.messages.stream.return_value = _good_stream_cm()
+
+        resp = agent._interruptible_streaming_api_call({})
+        assert resp is not None
+        assert agent._consecutive_stale_streams == 0
+
+    @pytest.mark.filterwarnings("ignore::pytest.PytestUnhandledThreadExceptionWarning")
+    def test_stale_kill_increments_streak(self, monkeypatch):
+        """Each stale-stream kill increments the consecutive-stale streak so a
+        wedged session eventually trips the breaker."""
+        monkeypatch.setenv("HERMES_STREAM_STALE_TIMEOUT", "0.1")
+        monkeypatch.setenv("HERMES_STREAM_STALE_GIVEUP", "50")
+
+        agent = _make_anthropic_agent()
+        agent._consecutive_stale_streams = 0
+        unblock = threading.Event()
+
+        def _blocking_gen():
+            unblock.wait(timeout=5.0)
+            raise httpx.ConnectError("connection dropped after close()")
+            yield  # make this a generator so next() triggers the wait
+
+        def _stream_side_effect(*args, **kwargs):
+            cm = MagicMock()
+            stream = MagicMock()
+            stream.__iter__ = MagicMock(return_value=_blocking_gen())
+            cm.__enter__ = MagicMock(return_value=stream)
+            cm.__exit__ = MagicMock(return_value=False)
+            return cm
+
+        # Every attempt blocks, trips the stale detector, and fails.
+        agent._anthropic_client.messages.stream.side_effect = _stream_side_effect
+        agent._anthropic_client.close.side_effect = unblock.set
+
+        with pytest.raises(Exception):
+            agent._interruptible_streaming_api_call({})
+
+        # At least one stale kill happened; the streak must have advanced.
+        assert agent._consecutive_stale_streams >= 1