fix(stream): rebuild Anthropic client on stream retry, not the OpenAI client

agent/chat_completion_helpers.py

@@ -2333,10 +2333,21 @@ def _call():
                             diag=request_client_holder.get("diag"),
                         )
                         _close_request_client_once("stream_mid_tool_retry_cleanup")
+                        # Rebuild the client that actually carried the
+                        # stream. In anthropic_messages mode that is the
+                        # shared Anthropic client — the OpenAI rebuild
+                        # would fail (``_client_kwargs`` is empty in this
+                        # mode) while leaving the dead connection in the
+                        # Anthropic pool for the retry to hit. (#44006)
                         try:
-                            agent._replace_primary_openai_client(
-                                reason="stream_mid_tool_retry_pool_cleanup"
-                            )
+                            if agent.api_mode == "anthropic_messages":
+                                agent._replace_primary_anthropic_client(
+                                    reason="stream_mid_tool_retry_pool_cleanup"
+                                )
+                            else:
+                                agent._replace_primary_openai_client(
+                                    reason="stream_mid_tool_retry_pool_cleanup"
+                                )
                         except Exception:
                             pass
                         continue
@@ -2385,11 +2396,18 @@ def _call():
                             # Close the stale request client before retry
                             _close_request_client_once("stream_retry_cleanup")
                             # Also rebuild the primary client to purge
-                            # any dead connections from the pool.
+                            # any dead connections from the pool — the
+                            # Anthropic client in anthropic_messages mode,
+                            # the shared OpenAI client otherwise. (#44006)
                             try:
-                                agent._replace_primary_openai_client(
-                                    reason="stream_retry_pool_cleanup"
-                                )
+                                if agent.api_mode == "anthropic_messages":
+                                    agent._replace_primary_anthropic_client(
+                                        reason="stream_retry_pool_cleanup"
+                                    )
+                                else:
+                                    agent._replace_primary_openai_client(
+                                        reason="stream_retry_pool_cleanup"
+                                    )
                             except Exception:
                                 pass
                             continue

run_agent.py

@@ -3479,6 +3479,56 @@ def _replace_primary_openai_client(self, *, reason: str) -> bool:
         self._close_openai_client(old_client, reason=f"replace:{reason}", shared=True)
         return True
 
+    def _replace_primary_anthropic_client(self, *, reason: str) -> bool:
+        """Rebuild the shared Anthropic client with the same credentials.
+
+        Anthropic-protocol analog of ``_replace_primary_openai_client``:
+        after a stream drop the dead connection lives in
+        ``_anthropic_client``'s httpx pool, where a retry can pick it up
+        and die again. Rebuilding swaps in a fresh pool.
+        ``_try_refresh_anthropic_client_credentials`` doesn't cover this —
+        it only rebuilds for the native Anthropic provider when the token
+        actually rotated, so third-party anthropic_messages providers
+        (MiniMax, Alibaba, ...) keep the poisoned pool across retries.
+        """
+        if self.api_mode != "anthropic_messages":
+            return False
+        old_client = getattr(self, "_anthropic_client", None)
+        if old_client is None:
+            return False
+        from agent.anthropic_adapter import build_anthropic_client
+
+        try:
+            new_client = build_anthropic_client(
+                self._anthropic_api_key,
+                getattr(self, "_anthropic_base_url", None),
+                timeout=get_provider_request_timeout(self.provider, self.model),
+            )
+        except Exception as exc:
+            logger.warning(
+                "Failed to rebuild shared Anthropic client (%s) %s error=%s",
+                reason,
+                self._client_log_context(),
+                exc,
+            )
+            return False
+        self._anthropic_client = new_client
+        try:
+            old_client.close()
+            logger.info(
+                "Anthropic client replaced (%s) %s",
+                reason,
+                self._client_log_context(),
+            )
+        except Exception as exc:
+            logger.debug(
+                "Anthropic client close failed (replace:%s) %s error=%s",
+                reason,
+                self._client_log_context(),
+                exc,
+            )
+        return True
+
     def _ensure_primary_openai_client(self, *, reason: str) -> Any:
         with self._openai_client_lock():
             client = getattr(self, "client", None)

tests/run_agent/test_streaming.py

@@ -986,9 +986,10 @@ def test_anthropic_stream_refreshes_activity_on_every_event(self):
 
         assert touch_calls.count("receiving stream response") == len(events)
 
+    @patch("run_agent.AIAgent._replace_primary_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_openai, mock_replace_anthropic, monkeypatch,
     ):
         """Malformed Anthropic event-stream frames retry instead of surfacing HTTP None."""
         from run_agent import AIAgent
@@ -1035,7 +1036,11 @@ def __iter__(self):
 
         assert response is final_message
         assert agent._anthropic_client.messages.stream.call_count == 2
-        assert mock_replace.call_count == 1
+        # The retry must purge the Anthropic client's pool — the OpenAI
+        # rebuild is a guaranteed failure in anthropic_messages mode
+        # (``_client_kwargs`` is empty) and must not be attempted. (#44006)
+        assert mock_replace_anthropic.call_count == 1
+        assert mock_replace_openai.call_count == 0
 
     @patch("run_agent.AIAgent._replace_primary_openai_client")
     def test_generic_anthropic_valueerror_still_propagates_without_stream_retry(
@@ -1069,6 +1074,75 @@ def test_generic_anthropic_valueerror_still_propagates_without_stream_retry(
         assert mock_replace.call_count == 0
 
 
+class TestReplacePrimaryAnthropicClient:
+    """``_replace_primary_anthropic_client`` swaps in a fresh client (and
+    httpx pool) with the SAME credentials, so the stream-retry path can
+    purge dead connections for anthropic_messages providers. (#44006)"""
+
+    def _agent(self):
+        from run_agent import AIAgent
+
+        agent = AIAgent(
+            api_key="test-key",
+            base_url="https://api.minimax.io/anthropic",
+            provider="minimax",
+            model="MiniMax-M2.7",
+            quiet_mode=True,
+            skip_context_files=True,
+            skip_memory=True,
+        )
+        agent.api_mode = "anthropic_messages"
+        return agent
+
+    @patch("agent.anthropic_adapter.build_anthropic_client")
+    def test_rebuilds_with_same_credentials_and_closes_old_client(
+        self, mock_build,
+    ):
+        agent = self._agent()
+        old_client = MagicMock()
+        agent._anthropic_client = old_client
+        agent._anthropic_api_key = "mm-key"
+        agent._anthropic_base_url = "https://api.minimax.io/anthropic"
+        new_client = MagicMock()
+        mock_build.return_value = new_client
+
+        assert agent._replace_primary_anthropic_client(
+            reason="stream_retry_pool_cleanup"
+        ) is True
+        assert agent._anthropic_client is new_client
+        old_client.close.assert_called_once()
+        args, _kwargs = mock_build.call_args
+        assert args[0] == "mm-key"
+        assert args[1] == "https://api.minimax.io/anthropic"
+
+    @patch("agent.anthropic_adapter.build_anthropic_client")
+    def test_keeps_old_client_when_rebuild_fails(self, mock_build):
+        agent = self._agent()
+        old_client = MagicMock()
+        agent._anthropic_client = old_client
+        agent._anthropic_api_key = "mm-key"
+        mock_build.side_effect = RuntimeError("boom")
+
+        assert agent._replace_primary_anthropic_client(
+            reason="stream_retry_pool_cleanup"
+        ) is False
+        assert agent._anthropic_client is old_client
+        old_client.close.assert_not_called()
+
+    def test_noop_outside_anthropic_mode(self):
+        agent = self._agent()
+        agent.api_mode = "chat_completions"
+        agent._anthropic_client = MagicMock()
+
+        assert agent._replace_primary_anthropic_client(reason="x") is False
+
+    def test_noop_without_anthropic_client(self):
+        agent = self._agent()
+        agent._anthropic_client = None
+
+        assert agent._replace_primary_anthropic_client(reason="x") is False
+
+
 class TestPartialToolCallWarning:
     """Regression: when a stream dies mid tool-call argument generation after
     text was already delivered, the partial-stream stub at run_agent.py