From 2985d16be06ea5021f3d3a7d8e0aff724f11c530 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Wed, 8 Jul 2026 01:18:14 +0530 Subject: [PATCH] fix: reset stream-stale breaker on model switch and fallback activation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up for the salvaged #60332 circuit breaker. The breaker latches: once the streak trips, interruptible_streaming_api_call raises before any stream is attempted, so the on-success reset can never run again. The error text tells the user to switch models and retry — but neither switch_model() nor try_activate_fallback() cleared the streak, so a freshly selected healthy provider kept short-circuiting forever (only /new recovered), and the automatic fallback chain was wedged the same way. Reset the streak at both swap sites (after a successful rebuild only; rollback/exhaustion paths keep the latch). 4 tests. --- agent/agent_runtime_helpers.py | 7 + agent/chat_completion_helpers.py | 5 + .../test_stream_stale_breaker_reset.py | 147 ++++++++++++++++++ 3 files changed, 159 insertions(+) create mode 100644 tests/run_agent/test_stream_stale_breaker_reset.py diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py index b3a9cf5321c..875bc17d5c8 100644 --- a/agent/agent_runtime_helpers.py +++ b/agent/agent_runtime_helpers.py @@ -1992,6 +1992,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) ── + # 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 + # ── 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._primary_runtime = { diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index b4d0d7dbad4..1820f32caed 100644 --- a/agent/chat_completion_helpers.py +++ b/agent/chat_completion_helpers.py @@ -1488,6 +1488,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 return True except Exception as e: if fb_provider == "nous": diff --git a/tests/run_agent/test_stream_stale_breaker_reset.py b/tests/run_agent/test_stream_stale_breaker_reset.py new file mode 100644 index 00000000000..4df83e8ac9f --- /dev/null +++ b/tests/run_agent/test_stream_stale_breaker_reset.py @@ -0,0 +1,147 @@ +"""Follow-up for the cross-turn stream-stale circuit breaker (#58962). + +The breaker latches: once ``_consecutive_stale_streams`` reaches the give-up +threshold, ``interruptible_streaming_api_call`` raises BEFORE any stream is +attempted — so the "reset on successful stream" path can never run again on +its own. The breaker's error message tells the user to "switch models … +then retry", and the provider-fallback chain swaps providers on the same +agent object, so BOTH swap paths must clear the streak or a healthy new +provider would keep short-circuiting forever: + +- ``switch_model()`` (user-initiated /model swap) +- ``try_activate_fallback()`` (automatic provider fallback) +""" + +from unittest.mock import MagicMock, patch + +from run_agent import AIAgent + + +def _make_agent_openrouter(): + """Minimal openrouter agent (skips __init__), mirroring + tests/run_agent/test_switch_model_rollback.py.""" + agent = AIAgent.__new__(AIAgent) + + agent.provider = "openrouter" + agent.model = "x-ai/grok-4" + agent.base_url = "https://openrouter.ai/api/v1" + agent.api_key = "or-key-original" + agent.api_mode = "chat_completions" + agent.client = MagicMock(name="OriginalClient") + agent._client_kwargs = { + "api_key": "or-key-original", + "base_url": "https://openrouter.ai/api/v1", + } + agent.context_compressor = None + agent._anthropic_api_key = "" + agent._anthropic_base_url = None + agent._anthropic_client = None + agent._is_anthropic_oauth = False + agent._cached_system_prompt = "cached" + agent._primary_runtime = {} + agent._fallback_activated = False + agent._fallback_index = 0 + agent._fallback_chain = [] + agent._fallback_model = None + agent._config_context_length = None + + return agent + + +def _make_fallback_agent(fallback_model): + """Full-constructor agent for the fallback path, mirroring + tests/run_agent/test_24996_fallback_exhaustion_cooldown.py.""" + with ( + patch("run_agent.get_tool_definitions", return_value=[]), + patch("run_agent.check_toolset_requirements", return_value={}), + patch("run_agent.OpenAI"), + ): + agent = AIAgent( + api_key="test-key", + base_url="https://openrouter.ai/api/v1", + quiet_mode=True, + skip_context_files=True, + skip_memory=True, + fallback_model=fallback_model, + ) + agent.client = MagicMock() + return agent + + +def _mock_client(base_url="https://openrouter.ai/api/v1", api_key="fb-key"): + mock = MagicMock() + mock.base_url = base_url + mock.api_key = api_key + return mock + + +def test_switch_model_resets_stale_streak(): + """A user-initiated /model swap must clear the latched streak so the new + provider gets a real stream attempt instead of an instant short-circuit.""" + agent = _make_agent_openrouter() + agent._consecutive_stale_streams = 7 # past any reasonable threshold + + agent._create_openai_client = MagicMock(return_value=MagicMock(name="NewClient")) + + with patch("hermes_cli.timeouts.get_provider_request_timeout", return_value=None): + agent.switch_model( + new_model="openai/gpt-5", + new_provider="openrouter", + api_key="or-key-new", + base_url="https://openrouter.ai/api/v1", + api_mode="chat_completions", + ) + + assert agent._consecutive_stale_streams == 0 + + +def test_switch_model_failure_does_not_reset_streak(): + """A failed swap rolls back — the agent is still on the wedged provider, + so the breaker must stay latched (reset happens after the rebuild).""" + agent = _make_agent_openrouter() + agent._consecutive_stale_streams = 7 + + def boom(*_a, **_kw): + raise RuntimeError("simulated client build failure") + + agent._create_openai_client = boom + + with patch("hermes_cli.timeouts.get_provider_request_timeout", return_value=None): + try: + agent.switch_model( + new_model="openai/gpt-5", + new_provider="openrouter", + api_key="or-key-new", + base_url="https://openrouter.ai/api/v1", + api_mode="chat_completions", + ) + except RuntimeError: + pass + + assert agent._consecutive_stale_streams == 7 + + +def test_fallback_activation_resets_stale_streak(): + """Automatic provider fallback swaps to a different backend; the streak + measured the OLD provider and must not wedge the new one.""" + fbs = [{"provider": "openai", "model": "gpt-4o"}] + agent = _make_fallback_agent(fallback_model=fbs) + agent._consecutive_stale_streams = 7 + + with patch( + "agent.auxiliary_client.resolve_provider_client", + return_value=(_mock_client(), "resolved"), + ): + assert agent._try_activate_fallback() is True + + assert agent._consecutive_stale_streams == 0 + + +def test_fallback_exhaustion_keeps_stale_streak(): + """When the chain is exhausted (no swap happened), the streak stays + latched — the session is still wedged on the same provider.""" + agent = _make_fallback_agent(fallback_model=[]) + agent._consecutive_stale_streams = 7 + + assert agent._try_activate_fallback() is False + assert agent._consecutive_stale_streams == 7