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

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="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
+
+
+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(