From 985e19c110f19e1131b52f78c6505e2a06837127 Mon Sep 17 00:00:00 2001 From: dsad Date: Tue, 7 Jul 2026 18:16:41 +0300 Subject: [PATCH] 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 | 45 +++++++ .../test_stream_stale_circuit_breaker.py | 125 ++++++++++++++++++ 2 files changed, 170 insertions(+) create mode 100644 tests/run_agent/test_stream_stale_circuit_breaker.py diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index 72e6920cdba..b4d0d7dbad4 100644 --- a/agent/chat_completion_helpers.py +++ b/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 ────────────────────────────────────────────────── diff --git a/tests/run_agent/test_stream_stale_circuit_breaker.py b/tests/run_agent/test_stream_stale_circuit_breaker.py new file mode 100644 index 00000000000..4a90b27b7b8 --- /dev/null +++ b/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