fix(agent): request-local Anthropic clients so the stale/interrupt watchdog never corrupts SQLite (#67142, supersedes #51688)

run_agent.py

@@ -4259,6 +4259,103 @@ def _abort_request_openai_client(self, client: Any, *, reason: str) -> None:
                 exc,
             )
 
+    def _create_request_anthropic_client(self, *, reason: str) -> Any:
+        """Build a request-local Anthropic client for one in-flight call.
+
+        The shared ``_anthropic_client`` stays the long-lived primary, but the
+        stale/interrupt watchdog runs on the poll thread and must never call
+        ``close()`` on the client whose TLS socket a worker thread is still
+        reading: releasing that FD from a stranger thread lets the kernel
+        recycle it under a still-live SSL BIO, which then writes a TLS record
+        into an unrelated SQLite header (#29507 / #67142). A per-request client
+        lets the stranger thread ``shutdown()`` the socket while the owning
+        worker performs the SDK-level close from its own context — the same
+        ownership contract the OpenAI-wire path already uses.
+
+        Mirrors ``_rebuild_anthropic_client`` construction (direct + Bedrock,
+        1M-beta drop) but returns a fresh client instead of swapping the shared
+        one.
+        """
+        if self.api_mode == "anthropic_messages":
+            self._try_refresh_anthropic_client_credentials()
+        _drop_1m = bool(getattr(self, "_oauth_1m_beta_disabled", False))
+        if getattr(self, "provider", None) == "bedrock":
+            from agent.anthropic_adapter import build_anthropic_bedrock_client
+            region = getattr(self, "_bedrock_region", "us-east-1") or "us-east-1"
+            client = build_anthropic_bedrock_client(region)
+        else:
+            from agent.anthropic_adapter import build_anthropic_client
+            client = build_anthropic_client(
+                self._anthropic_api_key,
+                getattr(self, "_anthropic_base_url", None),
+                timeout=get_provider_request_timeout(self.provider, self.model),
+                drop_context_1m_beta=_drop_1m,
+            )
+        logger.debug(
+            "Anthropic request client created (%s, shared=False) provider=%s model=%s",
+            reason,
+            getattr(self, "provider", None),
+            getattr(self, "model", None),
+        )
+        return client
+
+    def _close_request_anthropic_client(self, client: Any, *, reason: str) -> None:
+        """Owner-thread full close of a request-local Anthropic client.
+
+        Force-closes the pool's TCP sockets first (CLOSE-WAIT hygiene, parity
+        with ``_close_openai_client``), then does the graceful SDK close. Safe
+        because the caller owns the connection.
+        """
+        if client is None:
+            return
+        try:
+            self._force_close_tcp_sockets(client)
+            client.close()
+            logger.info(
+                "Anthropic client closed (%s, shared=False) provider=%s model=%s",
+                reason,
+                getattr(self, "provider", None),
+                getattr(self, "model", None),
+            )
+        except Exception as exc:
+            logger.debug(
+                "Anthropic client close failed (%s, shared=False) provider=%s model=%s error=%s",
+                reason,
+                getattr(self, "provider", None),
+                getattr(self, "model", None),
+                exc,
+            )
+
+    def _abort_request_anthropic_client(self, client: Any, *, reason: str) -> None:
+        """Cross-thread abort for request-local Anthropic clients.
+
+        Stranger threads (the interrupt-check / stale-stream detector loop)
+        must not call the SDK ``close()`` — that races the owning worker's live
+        SSL BIO and can recycle a TLS FD into a SQLite header (#29507 /
+        #67142). Only ``shutdown(SHUT_RDWR)`` the pool's sockets so the worker
+        unblocks and releases the FD from its own thread.
+        """
+        if client is None:
+            return
+        try:
+            shutdown_count = self._force_close_tcp_sockets(client)
+            logger.info(
+                "Anthropic client aborted (%s, shared=False, tcp_force_closed=%d, "
+                "deferred_close=stranger_thread) provider=%s model=%s",
+                reason,
+                shutdown_count,
+                getattr(self, "provider", None),
+                getattr(self, "model", None),
+            )
+        except Exception as exc:
+            logger.debug(
+                "Anthropic client abort failed (%s, shared=False) provider=%s model=%s error=%s",
+                reason,
+                getattr(self, "provider", None),
+                getattr(self, "model", None),
+                exc,
+            )
+
     def _run_codex_stream(self, api_kwargs: dict, client: Any = None, on_first_delta: callable = None):
         """Forwarder — see ``agent.codex_runtime.run_codex_stream``."""
         from agent.codex_runtime import run_codex_stream
@@ -4644,15 +4741,18 @@ def _credential_pool_may_recover_rate_limit(self) -> bool:
             return False
         return pool.has_available()
 
-    def _anthropic_messages_create(self, api_kwargs: dict):
-        if self.api_mode == "anthropic_messages":
+    def _anthropic_messages_create(self, api_kwargs: dict, *, client: Any = None):
+        # When a request-local client is supplied it was already credential-
+        # refreshed in ``_create_request_anthropic_client``; only the shared
+        # fallback path refreshes here.
+        if client is None and self.api_mode == "anthropic_messages":
             self._try_refresh_anthropic_client_credentials()
         # Defensive: strip Responses-only kwargs that can leak in under an
         # api_mode-flip race (the Anthropic SDK raises a non-retryable
         # TypeError on them). See #31673.
         from agent.anthropic_adapter import create_anthropic_message
         return create_anthropic_message(
-            self._anthropic_client,
+            client or self._anthropic_client,
             api_kwargs,
             log_prefix=getattr(self, "log_prefix", ""),
             prefer_stream=not bool(getattr(self, "_disable_streaming", False)),

tests/agent/test_cascading_interrupt_6600.py

@@ -132,3 +132,93 @@ def _create_1(**kwargs):
 
     with pytest.raises(httpx.RemoteProtocolError):
         cch.interruptible_api_call(agent, {"model": "x", "messages": []})
+
+
+# ---------------------------------------------------------------------------
+# #67142: direct-Anthropic stale/interrupt watchdog must abort the request-local
+# client from the poll (stranger) thread and NEVER close/rebuild the shared
+# _anthropic_client — closing it there released a live TLS FD that the kernel
+# recycled into a SQLite handle, writing a TLS record over a DB header.
+# ---------------------------------------------------------------------------
+
+
+def _make_anthropic_agent():
+    agent = _make_agent()
+    agent.api_mode = "anthropic_messages"
+    return agent
+
+
+def _wait_for_mock_call(mock, timeout=3.0):
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        if mock.called:
+            return
+        time.sleep(0.02)
+    raise AssertionError(f"{mock!r} was not called within {timeout}s")
+
+
+def test_anthropic_non_streaming_stale_aborts_request_client_not_shared():
+    """Stale non-streaming Anthropic call: the poll thread aborts the
+    request-local client's socket; the shared client is never closed/rebuilt,
+    and the worker still unblocks and closes its own client (no #28161 hang)."""
+    agent = _make_anthropic_agent()
+    agent._compute_non_stream_stale_timeout.return_value = 0.05
+    agent._codex_silent_hang_hint = MagicMock(return_value=None)
+
+    request_client = MagicMock()
+    agent._create_request_anthropic_client = MagicMock(return_value=request_client)
+    agent._abort_request_anthropic_client = MagicMock()
+    agent._close_request_anthropic_client = MagicMock()
+
+    def _create(_api_kwargs, *, client):
+        assert client is request_client
+        # Outlive the 0.05s stale timeout AND the worker join (2.0s) so the
+        # stale detector surfaces its TimeoutError.
+        time.sleep(2.5)
+        return object()
+
+    agent._anthropic_messages_create = MagicMock(side_effect=_create)
+
+    with pytest.raises(TimeoutError):
+        cch.interruptible_api_call(agent, {"model": "x", "messages": []})
+
+    # Shared client untouched from the poll thread.
+    agent._anthropic_client.close.assert_not_called()
+    agent._rebuild_anthropic_client.assert_not_called()
+    # Poll (stranger) thread aborts the request-local client's socket only.
+    agent._abort_request_anthropic_client.assert_called_once_with(
+        request_client, reason="stale_call_kill"
+    )
+    # Worker unblocks and closes its own request client from its own thread.
+    _wait_for_mock_call(agent._close_request_anthropic_client)
+
+
+def test_anthropic_non_streaming_interrupt_aborts_request_client_not_shared():
+    """Interrupted non-streaming Anthropic call: near-instant InterruptedError,
+    request-local client aborted from the poll thread, shared client untouched."""
+    agent = _make_anthropic_agent()
+
+    request_client = MagicMock()
+    agent._create_request_anthropic_client = MagicMock(return_value=request_client)
+    agent._abort_request_anthropic_client = MagicMock()
+    agent._close_request_anthropic_client = MagicMock()
+
+    def _create(_api_kwargs, *, client):
+        assert client is request_client
+        agent._interrupt_requested = True
+        time.sleep(1.0)
+        raise httpx.RemoteProtocolError("forced close would have happened")
+
+    agent._anthropic_messages_create = MagicMock(side_effect=_create)
+
+    t0 = time.time()
+    with pytest.raises(InterruptedError):
+        cch.interruptible_api_call(agent, {"model": "x", "messages": []})
+    elapsed = time.time() - t0
+
+    assert elapsed < 3.0, f"interrupt took {elapsed:.1f}s — should be near-instant"
+    agent._anthropic_client.close.assert_not_called()
+    agent._rebuild_anthropic_client.assert_not_called()
+    agent._abort_request_anthropic_client.assert_called_once_with(
+        request_client, reason="interrupt_abort"
+    )

tests/run_agent/test_28161_anthropic_stream_pool_cleanup.py

@@ -1,16 +1,24 @@
-"""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.
+"""Anthropic stream cleanup must not call _replace_primary_openai_client() and
+must not hang on Anthropic-native configs (#28161), now via the request-local
+client model (#67142).
+
+Originally three cleanup sites in interruptible_streaming_api_call() called
+_replace_primary_openai_client() unconditionally; for api_mode=anthropic_messages
+that silently failed (no OPENAI_API_KEY) and left the in-flight httpx stream
+unclosed, blocking the worker until the 900s read-timeout fired.
+
+Since #67142, anthropic streams run on a per-request client: the stale/retry
+cleanup closes the *request-local* client (worker-owned) and builds a fresh one
+next attempt — the shared _anthropic_client is never closed/rebuilt from inside
+a request (that poll-thread close was the TLS-FD→SQLite corruption vector). The
+no-hang guarantee is preserved because the poll thread aborts the request
+client's sockets, which unblocks the worker.
 
 Tests cover:
-- stream_retry_pool_cleanup  (connection error on fresh stream, L1836)
-- stale_stream_pool_cleanup  (outer poll loop detects stale stream, L1987)
+- stream_retry cleanup  (connection error on fresh stream)
+- stale_stream cleanup  (outer poll loop detects stale stream)
 
-Fixes #28161
+Fixes #28161. Extends #67142.
 """
 import threading
 from types import SimpleNamespace
@@ -41,6 +49,9 @@ def _make_anthropic_agent(**kwargs):
     agent.api_mode = "anthropic_messages"
     agent._anthropic_client = MagicMock()
     agent._anthropic_api_key = "test-anthropic-key"
+    # #67142: anthropic streams now run on a request-local client; route it to
+    # the test mock so .messages.stream is exercised and its cleanup observed.
+    agent._create_request_anthropic_client = lambda *a, **k: agent._anthropic_client
     return agent
 
 
@@ -74,13 +85,16 @@ def _failing_stream_cm():
 
 
 class TestAnthropicStreamPoolCleanup:
-    """_replace_primary_openai_client must not be called for api_mode=anthropic_messages."""
+    """Anthropic cleanup must never touch the OpenAI primary or the shared
+    Anthropic client, and must not hang (#28161 / #67142)."""
 
     @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."""
+    def test_stream_retry_closes_request_client_not_openai(self):
+        """Connection error during stream retry → close the request-local
+        Anthropic client (worker-owned) and retry; never rebuild the shared
+        Anthropic client, never touch the OpenAI primary."""
         agent = _make_anthropic_agent()
 
         attempt_count = [0]
@@ -100,14 +114,19 @@ def _stream_side_effect(*args, **kwargs):
                 agent._interruptible_streaming_api_call({})
 
         mock_replace.assert_not_called()
-        mock_rebuild.assert_called_once()
-        agent._anthropic_client.close.assert_called_once()
+        # #67142: the shared client is never rebuilt from inside a request; the
+        # request-local client (routed to this mock) is closed instead.
+        mock_rebuild.assert_not_called()
+        agent._anthropic_client.close.assert_called()
+        assert attempt_count[0] == 2  # retried once, then succeeded
 
     @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."""
+    def test_stale_stream_aborts_request_client_not_openai(self, monkeypatch):
+        """Stale-stream outer-poll detector → abort the request-local client's
+        socket (unblocking the worker) and retry; never _replace_primary_openai
+        and never rebuild the shared Anthropic client."""
         monkeypatch.setenv("HERMES_STREAM_STALE_TIMEOUT", "0.1")
 
         agent = _make_anthropic_agent()
@@ -117,14 +136,15 @@ def test_stale_stream_calls_anthropic_rebuild_not_openai(self, monkeypatch):
         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.
+                # First attempt: stream that yields nothing (triggers stale
+                # detector), then raises ConnectError once the poll thread
+                # aborts the request client's socket.
                 cm = MagicMock()
                 stream = MagicMock()
 
                 def _blocking_gen():
                     unblock.wait(timeout=5.0)
-                    raise httpx.ConnectError("connection dropped after close()")
+                    raise httpx.ConnectError("connection dropped after abort")
                     yield  # make this a generator so next() triggers the wait
 
                 stream.__iter__ = MagicMock(return_value=_blocking_gen())
@@ -135,8 +155,10 @@ def _blocking_gen():
             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
+        # #67142: the stale detector aborts the request-local client's sockets
+        # from the poll thread (not close() on the shared client); simulate the
+        # socket shutdown waking the blocked read.
+        agent._abort_request_anthropic_client = lambda *a, **k: unblock.set()
 
         with patch.object(agent, "_rebuild_anthropic_client") as mock_rebuild:
             with patch.object(
@@ -145,6 +167,6 @@ def _blocking_gen():
                 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
+        # The shared Anthropic client is never rebuilt from inside a request.
+        mock_rebuild.assert_not_called()
+        assert attempt_count[0] >= 2  # stale-killed once, then retried