Stale-stream watchdog rebuilds wrong client for non-OpenAI providers

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #53926 fix(streaming): route Anthropic stream cleanup to the Anthropic client, not OpenAI rebuild (#28161)

  merged to main

  files: agent/chat_completion_helpers.py, tests/run_agent/test_28161_anthropic_stream_pool_cleanup.py, tests/run_agent/test_streaming.py

  Show the diff

  • agent/chat_completion_helpers.py

    @@ -2459,12 +2459,19 @@ def _call():
                                 diag=request_client_holder.get("diag"),
                             )
                             _close_request_client_once("stream_mid_tool_retry_cleanup")
    -                        try:
    -                            agent._replace_primary_openai_client(
    -                                reason="stream_mid_tool_retry_pool_cleanup"
    -                            )
    -                        except Exception:
    -                            pass
    +                        if agent.api_mode == "anthropic_messages":
    +                            try:
    +                                agent._anthropic_client.close()
    +                                agent._rebuild_anthropic_client()
    +                            except Exception:
    +                                pass
    +                        else:
    +                            try:
    +                                agent._replace_primary_openai_client(
    +                                    reason="stream_mid_tool_retry_pool_cleanup"
    +                                )
    +                            except Exception:
    +                                pass
                             continue
     
                         # SSE error events from proxies (e.g. OpenRouter sends
    @@ -2512,12 +2519,19 @@ def _call():
                                 _close_request_client_once("stream_retry_cleanup")
                                 # Also rebuild the primary client to purge
                                 # any dead connections from the pool.
    -                            try:
    -                                agent._replace_primary_openai_client(
    -                                    reason="stream_retry_pool_cleanup"
    -                                )
    -                            except Exception:
    -                                pass
    +                            if agent.api_mode == "anthropic_messages":
    +                                try:
    +                                    agent._anthropic_client.close()
    +                                    agent._rebuild_anthropic_client()
    +                                except Exception:
    +                                    pass
    +                            else:
    +                                try:
    +                                    agent._replace_primary_openai_client(
    +                                        reason="stream_retry_pool_cleanup"
    +                                    )
    +                                except Exception:
    +                                    pass
                                 continue
                             # Retries exhausted. Log the final failure with
                             # full diagnostic detail (chain, headers,
    @@ -2688,10 +2702,17 @@ def _call():
                     pass
                 # Rebuild the primary client too — its connection pool
                 # may hold dead sockets from the same provider outage.
    -            try:
    -                agent._replace_primary_openai_client(reason="stale_stream_pool_cleanup")
    -            except Exception:
    -                pass
    +            if agent.api_mode == "anthropic_messages":
    +                try:
    +                    agent._anthropic_client.close()
    +                    agent._rebuild_anthropic_client()
    +                except Exception:
    +                    pass
    +            else:
    +                try:
    +                    agent._replace_primary_openai_client(reason="stale_stream_pool_cleanup")
    +                except Exception:
    +                    pass
                 # Reset the timer so we don't kill repeatedly while
                 # the inner thread processes the closure.
                 last_chunk_time["t"] = time.time()

  • tests/run_agent/test_28161_anthropic_stream_pool_cleanup.py

    @@ -0,0 +1,150 @@
    +"""Anthropic stream cleanup must call _anthropic_client.close() + _rebuild_anthropic_client(),
    +not _replace_primary_openai_client(), to avoid 15-minute hangs on Anthropic-native configs.
    +
    +Three cleanup sites in chat_completion_helpers.interruptible_streaming_api_call() were
    +calling _replace_primary_openai_client() unconditionally.  For api_mode=anthropic_messages
    +this silently fails (no OPENAI_API_KEY) and leaves the in-flight httpx stream unclosed,
    +blocking the worker thread until the 900s httpx read-timeout fires.
    +
    +Tests cover:
    +- stream_retry_pool_cleanup  (connection error on fresh stream, L1836)
    +- stale_stream_pool_cleanup  (outer poll loop detects stale stream, L1987)
    +
    +Fixes #28161
    +"""
    +import threading
    +from types import SimpleNamespace
    +from unittest.mock import MagicMock, patch
    +
    +import httpx
    +import pytest
    +
    +
    +# ---------------------------------------------------------------------------
    +# Helpers
    +# ---------------------------------------------------------------------------
    +
    +
    +def _make_anthropic_agent(**kwargs):
    +    from run_agent import AIAgent
    +
    +    defaults = dict(
    +        api_key="[redacted:secret-assignment]",
    +        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 = "[redacted:secret-assignment]"
    +    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
    +
    +
    +def _failing_stream_cm():
    +    """Context manager whose __enter__ raises ConnectError immediately."""
    +    cm = MagicMock()
    +    cm.__enter__ = MagicMock(
    +        side_effect=httpx.ConnectError("connection reset by peer")
    +    )
    +    return cm
    +
    +
    +# ---------------------------------------------------------------------------
    +# Tests
    +# ---------------------------------------------------------------------------
    +
    +
    +class TestAnthropicStreamPoolCleanup:
    +    """_replace_primary_openai_client must not be called for api_mode=anthropic_messages."""
    +
    +    @pytest.mark.filterwarnings(
    +        "ignore::pytest.PytestUnhandledThreadExceptionWarning"
    +    )
    +    def test_stream_retry_calls_anthropic_rebuild_not_openai(self):
    +        """Connection error during stream retry → close+rebuild Anthropic client, not OpenAI."""
    +        agent = _make_anthropic_agent()
    +
    +        attempt_count = [0]
    +
    +        def _stream_side_effect(*args, **kwargs):
    +            attempt_count[0] += 1
    +            if attempt_count[0] == 1:
    +                return _failing_stream_cm()
    +            return _good_stream_cm()
    +
    +        agent._anthropic_client.messages.stream.side_effect = _stream_side_effect
    +
    +        with patch.object(agent, "_rebuild_anthropic_client") as mock_rebuild:
    +            with patch.object(
    +                agent, "_replace_primary_openai_client"
    +            ) as mock_replace:
    +                agent._interruptible_streaming_api_call({})
    +
    +        mock_replace.assert_not_called()
    +        mock_rebuild.assert_called_once()
    +        agent._anthropic_client.close.assert_called_once()
    +
    +    @pytest.mark.filterwarnings(
    +        "ignore::pytest.PytestUnhandledThreadExceptionWarning"
    +    )
    +    def test_stale_stream_calls_anthropic_rebuild_not_openai(self, monkeypatch):
    +        """Stale-stream outer-poll detector → close+rebuild Anthropic client, not OpenAI."""
    +        monkeypatch.setenv("HERMES_STREAM_STALE_TIMEOUT", "0.1")
    +
    +        agent = _make_anthropic_agent()
    +        unblock = threading.Event()
    +        attempt_count = [0]
    +
    +        def _stream_side_effect(*args, **kwargs):
    +            attempt_count[0] += 1
    +            if attempt_count[0] == 1:
    +                # First attempt: stream that yields nothing (triggers stale detector),
    +                # then raises ConnectError once _anthropic_client.close() unblocks it.
    +                cm = MagicMock()
    +                stream = MagicMock()
    +
    +                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
    +
    +                stream.__iter__ = MagicMock(return_value=_blocking_gen())
    +                cm.__enter__ = MagicMock(return_value=stream)
    +                cm.__exit__ = MagicMock(return_value=False)
    +                return cm
    +            # Second attempt: succeed
    +            return _good_stream_cm()
    +
    +        agent._anthropic_client.messages.stream.side_effect = _stream_side_effect
    +        # close() on the mock Anthropic client unblocks the inner thread.
    +        agent._anthropic_client.close.side_effect = unblock.set
    +
    +        with patch.object(agent, "_rebuild_anthropic_client") as mock_rebuild:
    +            with patch.object(
    +                agent, "_replace_primary_openai_client"
    +            ) as mock_replace:
    +                agent._interruptible_streaming_api_call({})
    +
    +        mock_replace.assert_not_called()
    +        # close() and rebuild called at least once by the stale detector.
    +        agent._anthropic_client.close.assert_called()
    +        assert mock_rebuild.call_count >= 1

  • tests/run_agent/test_streaming.py

    @@ -986,11 +986,17 @@ def test_anthropic_stream_refreshes_activity_on_every_event(self):
     
             assert touch_calls.count("receiving stream response") == len(events)
     
    +    @patch("run_agent.AIAgent._rebuild_anthropic_client")
         @patch("run_agent.AIAgent._replace_primary_openai_client")
         def test_anthropic_stream_parser_valueerror_retries_before_delivery(
    -        self, mock_replace, monkeypatch,
    +        self, mock_replace, mock_rebuild, monkeypatch,
         ):
    -        """Malformed Anthropic event-stream frames retry instead of surfacing HTTP None."""
    +        """Malformed Anthropic event-stream frames retry instead of surfacing HTTP None.
    +
    +        On the Anthropic-native path the stream-retry cleanup must close + rebuild the
    +        Anthropic client, NOT the OpenAI primary client (which would fail with
    +        Missing-credentials and leave the wedged stream open). See #28161.
    +        """
             from run_agent import AIAgent
     
             agent = AIAgent(
    @@ -1035,7 +1041,11 @@ def __iter__(self):
     
             assert response is final_message
             assert agent._anthropic_client.messages.stream.call_count == 2
    -        assert mock_replace.call_count == 1
    +        # Anthropic-native cleanup: close + rebuild the Anthropic client, never
    +        # the OpenAI primary client.
    +        assert mock_replace.call_count == 0
    +        assert mock_rebuild.call_count == 1
    +        assert agent._anthropic_client.close.call_count == 1
     
         @patch("run_agent.AIAgent._replace_primary_openai_client")
         def test_generic_anthropic_valueerror_still_propagates_without_stream_retry(