diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py
index 2912628b9882..c3632bffc41e 100644
--- a/agent/chat_completion_helpers.py
+++ b/agent/chat_completion_helpers.py
@@ -372,13 +372,14 @@ def _dispatch_nonstreaming_api_request(agent, api_kwargs: dict, *, make_client):
     inline path (``direct_api_call``) so the per-api_mode dispatch — codex /
     anthropic / bedrock / MoA / OpenAI-compatible — lives in exactly one place.
 
-    ``make_client(reason)`` builds the per-request OpenAI client for the codex
-    and OpenAI-compatible branches; the worker path uses it to register the
-    client with its stranger-thread abort machinery, the inline path uses it to
-    capture the client for its own ``finally`` close. The anthropic / bedrock /
-    MoA branches manage their own clients and never call it. All interrupt,
-    abort, cancellation, and close semantics stay in the callers — this helper
-    only issues the request.
+    ``make_client(reason, kind=...)`` builds the per-request client for the
+    codex / OpenAI-compatible (``kind="openai"``) and anthropic
+    (``kind="anthropic_messages"``) branches; the worker path uses it to
+    register the client with its stranger-thread abort machinery, the inline
+    path uses it to capture the client for its own ``finally`` close. The
+    bedrock / MoA branches manage their own clients and never call it. All
+    interrupt, abort, cancellation, and close semantics stay in the callers —
+    this helper only issues the request.
     """
     if agent.api_mode == "codex_responses":
         request_client = make_client("codex_stream_request")
@@ -388,7 +389,13 @@ def _dispatch_nonstreaming_api_request(agent, api_kwargs: dict, *, make_client):
             on_first_delta=getattr(agent, "_codex_on_first_delta", None),
         )
     if agent.api_mode == "anthropic_messages":
-        return agent._anthropic_messages_create(api_kwargs)
+        # #67142: use a request-local Anthropic client so the stale/interrupt
+        # watchdog aborts sockets from the stranger thread while the worker
+        # owns the SDK close — never closing the shared client mid-flight.
+        request_client = make_client(
+            "anthropic_messages_request", kind="anthropic_messages"
+        )
+        return agent._anthropic_messages_create(api_kwargs, client=request_client)
     if agent.api_mode == "bedrock_converse":
         # Bedrock uses boto3 directly — no OpenAI client needed.
         # normalize_converse_response produces an OpenAI-compatible
@@ -458,7 +465,11 @@ def direct_api_call(agent, api_kwargs: dict):
         if request_client is not None:
             agent._abort_request_openai_client(request_client, reason=reason)
 
-    def _make_client(reason: str):
+    def _make_client(reason: str, kind: str = "openai"):
+        # direct_api_call only runs for OpenAI-wire chat_completions cron
+        # requests (see should_use_direct_api_call), so the anthropic branch of
+        # the dispatch — the only caller that passes kind — is never reached
+        # here; the ``kind`` parameter exists purely for signature parity.
         client = agent._create_request_openai_client(reason=reason, api_kwargs=api_kwargs)
         with request_client_lock:
             request_client_holder["client"] = client
@@ -517,6 +528,10 @@ def interruptible_api_call(agent, api_kwargs: dict):
     _check_stale_giveup(agent)
 
     request_client_holder = {"client": None, "owner_tid": None}
+    # Transport kind of the registered request client ("openai" or
+    # "anthropic_messages") so _close_request_client_once routes to the right
+    # abort/close helpers (#67142).
+    request_client_kind = {"value": "openai"}
     request_client_lock = threading.Lock()
     # Request-local cancellation flag. Distinct from agent._interrupt_requested
     # because that flag is cleared at run_conversation() turn boundaries, but
@@ -528,9 +543,10 @@ def interruptible_api_call(agent, api_kwargs: dict):
     # hang.)
     _request_cancelled = {"value": False}
 
-    def _set_request_client(client):
+    def _set_request_client(client, *, kind: str = "openai"):
         with request_client_lock:
             request_client_holder["client"] = client
+            request_client_kind["value"] = kind
             # #29507: stamp the owning thread so a stranger-thread interrupt
             # only shuts the connection down rather than racing the worker
             # for FD ownership during ``client.close()``.
@@ -564,24 +580,34 @@ def interruptible_api_call(agent, api_kwargs: dict):
                 request_client_holder["owner_tid"] = None
         if request_client is None:
             return
-        if stranger_thread:
+        kind = request_client_kind.get("value", "openai")
+        if kind == "anthropic_messages":
+            if stranger_thread:
+                agent._abort_request_anthropic_client(request_client, reason=reason)
+            else:
+                agent._close_request_anthropic_client(request_client, reason=reason)
+        elif stranger_thread:
             agent._abort_request_openai_client(request_client, reason=reason)
         else:
             agent._close_request_openai_client(request_client, reason=reason)
 
     def _call():
         try:
-            # _set_request_client registers each per-request OpenAI client with
-            # the stranger-thread abort machinery above; the shared dispatch
-            # helper builds it via this callback so the interrupt / stale-call
-            # detectors can force-close the worker's connection.
+            # _set_request_client registers each per-request client with the
+            # stranger-thread abort machinery above; the shared dispatch helper
+            # builds it via this callback (openai- or anthropic-kind) so the
+            # interrupt / stale-call detectors can force-close the worker's
+            # connection without touching the shared client (#67142).
             result["response"] = _dispatch_nonstreaming_api_request(
                 agent,
                 api_kwargs,
-                make_client=lambda reason: _set_request_client(
-                    agent._create_request_openai_client(
+                make_client=lambda reason, kind="openai": _set_request_client(
+                    agent._create_request_anthropic_client(reason=reason)
+                    if kind == "anthropic_messages"
+                    else agent._create_request_openai_client(
                         reason=reason, api_kwargs=api_kwargs
-                    )
+                    ),
+                    kind=kind,
                 ),
             )
         except Exception as e:
@@ -893,11 +919,10 @@ def interruptible_api_call(agent, api_kwargs: dict):
                     f"Aborting call."
                 )
             try:
-                if agent.api_mode == "anthropic_messages":
-                    agent._anthropic_client.close()
-                    agent._rebuild_anthropic_client()
-                else:
-                    _close_request_client_once("stale_call_kill")
+                # #67142: routes by client kind — anthropic now aborts the
+                # request-local client's sockets from this poll (stranger)
+                # thread instead of closing the shared _anthropic_client.
+                _close_request_client_once("stale_call_kill")
             except Exception:
                 pass
             # Circuit breaker (#58962): count the stale kill.  See the
@@ -933,13 +958,12 @@ def interruptible_api_call(agent, api_kwargs: dict):
             )
             # Force-close the in-flight worker-local HTTP connection to stop
             # token generation without poisoning the shared client used to
-            # seed future retries.
+            # seed future retries. #67142: for anthropic this aborts the
+            # request-local client's sockets from this poll (stranger) thread
+            # rather than closing the shared _anthropic_client, which could
+            # release a TLS FD mid-SSL-BIO and corrupt an unrelated SQLite DB.
             try:
-                if agent.api_mode == "anthropic_messages":
-                    agent._anthropic_client.close()
-                    agent._rebuild_anthropic_client()
-                else:
-                    _close_request_client_once("interrupt_abort")
+                _close_request_client_once("interrupt_abort")
             except Exception:
                 pass
             raise InterruptedError("Agent interrupted during API call")
@@ -2393,6 +2417,10 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
     _check_stale_giveup(agent)
 
     request_client_holder = {"client": None, "diag": None, "owner_tid": None}
+    # Transport kind of the registered request client — see the non-streaming
+    # variant. Routes _close_request_client_once to anthropic vs openai abort/
+    # close helpers (#67142).
+    request_client_kind = {"value": "openai"}
     request_client_lock = threading.Lock()
     # Request-local cancellation flag — see interruptible_api_call for the full
     # rationale. The streaming retry loop is where the 7-minute cascading-
@@ -2403,9 +2431,10 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
     # exit immediately instead of retrying. (PR #6600.)
     _request_cancelled = {"value": False}
 
-    def _set_request_client(client):
+    def _set_request_client(client, *, kind: str = "openai"):
         with request_client_lock:
             request_client_holder["client"] = client
+            request_client_kind["value"] = kind
             # See #29507 explanation in the non-streaming variant above.
             request_client_holder["owner_tid"] = threading.get_ident()
         return client
@@ -2428,7 +2457,13 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
                 request_client_holder["owner_tid"] = None
         if request_client is None:
             return
-        if stranger_thread:
+        kind = request_client_kind.get("value", "openai")
+        if kind == "anthropic_messages":
+            if stranger_thread:
+                agent._abort_request_anthropic_client(request_client, reason=reason)
+            else:
+                agent._close_request_anthropic_client(request_client, reason=reason)
+        elif stranger_thread:
             agent._abort_request_openai_client(request_client, reason=reason)
         else:
             agent._close_request_openai_client(request_client, reason=reason)
@@ -2976,13 +3011,18 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
             usage=usage_obj,
         )
 
-    def _call_anthropic():
+    def _call_anthropic(request_client):
         """Stream an Anthropic Messages API response.
 
         Fires delta callbacks for real-time token delivery, but returns
         the native Anthropic Message object from get_final_message() so
         the rest of the agent loop (validation, tool extraction, etc.)
         works unchanged.
+
+        Uses ``request_client`` (a per-request Anthropic client registered with
+        the stranger-thread abort machinery) rather than the shared
+        ``_anthropic_client``, so the stale/interrupt watchdog can abort this
+        stream's socket without closing the shared client mid-flight (#67142).
         """
         has_tool_use = False
         # Zero-event guard parity with the chat_completions path: track
@@ -3011,7 +3051,7 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
             api_kwargs, log_prefix=getattr(agent, "log_prefix", "")
         )
         # Use the Anthropic SDK's streaming context manager
-        with agent._anthropic_client.messages.stream(**api_kwargs) as stream:
+        with request_client.messages.stream(**api_kwargs) as stream:
             # The Anthropic SDK exposes the raw httpx response on
             # ``stream.response``.  Snapshot diagnostic headers
             # immediately so they survive a stream that dies before the
@@ -3149,8 +3189,16 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
                     raise InterruptedError("Agent interrupted before stream retry")
                 try:
                     if agent.api_mode == "anthropic_messages":
-                        agent._try_refresh_anthropic_client_credentials()
-                        result["response"] = _call_anthropic()
+                        # #67142: per-request client (credential refresh happens
+                        # inside _create_request_anthropic_client) registered so
+                        # the watchdog aborts its socket, not the shared client.
+                        request_client = _set_request_client(
+                            agent._create_request_anthropic_client(
+                                reason="anthropic_stream_request"
+                            ),
+                            kind="anthropic_messages",
+                        )
+                        result["response"] = _call_anthropic(request_client)
                     else:
                         result["response"] = _call_chat_completions(stream_attempt_id)
                     return  # success
@@ -3276,13 +3324,11 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
                         )
                         _cancel_current_stream_attempt("stream_mid_tool_retry_cleanup")
                         _close_request_client_once("stream_mid_tool_retry_cleanup")
-                        if agent.api_mode == "anthropic_messages":
-                            try:
-                                agent._anthropic_client.close()
-                                agent._rebuild_anthropic_client()
-                            except Exception:
-                                pass
-                        else:
+                        # #67142: anthropic streams on a request-local client,
+                        # already worker-owned-closed by _close_request_client_once
+                        # above; the next attempt builds a fresh one. The shared
+                        # _anthropic_client is never closed from inside a request.
+                        if agent.api_mode != "anthropic_messages":
                             try:
                                 agent._replace_primary_openai_client(
                                     reason="stream_mid_tool_retry_pool_cleanup"
@@ -3341,15 +3387,13 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
                             # Close the stale request client before retry
                             _cancel_current_stream_attempt("stream_retry_cleanup")
                             _close_request_client_once("stream_retry_cleanup")
-                            # Also rebuild the primary client to purge
-                            # any dead connections from the pool.
-                            if agent.api_mode == "anthropic_messages":
-                                try:
-                                    agent._anthropic_client.close()
-                                    agent._rebuild_anthropic_client()
-                                except Exception:
-                                    pass
-                            else:
+                            # Also rebuild the primary client to purge any dead
+                            # connections from the pool. #67142: anthropic uses a
+                            # request-local client (already worker-owned-closed
+                            # above; next attempt builds fresh), so the shared
+                            # _anthropic_client is never closed from inside a
+                            # request — only the OpenAI-wire primary is refreshed.
+                            if agent.api_mode != "anthropic_messages":
                                 try:
                                     agent._replace_primary_openai_client(
                                         reason="stream_retry_pool_cleanup"
@@ -3589,11 +3633,14 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
             # Rebuild the primary client too — its connection pool
             # may hold dead sockets from the same provider outage.
             if agent.api_mode == "anthropic_messages":
-                try:
-                    agent._anthropic_client.close()
-                    agent._rebuild_anthropic_client()
-                except Exception:
-                    pass
+                # #67142: the stale stream ran on a request-local anthropic
+                # client, already socket-aborted above via
+                # _close_request_client_once (which unblocks the worker and
+                # preserves the #28161 no-hang guarantee). The shared
+                # _anthropic_client is NOT the in-flight transport, so we must
+                # not close it from this poll (stranger) thread — that was the
+                # FD-recycle corruption vector. Nothing further is needed.
+                pass
             else:
                 try:
                     agent._replace_primary_openai_client(reason="stale_stream_pool_cleanup")
@@ -3622,11 +3669,10 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
             )
             try:
                 _cancel_current_stream_attempt("stream_interrupt_abort")
-                if agent.api_mode == "anthropic_messages":
-                    agent._anthropic_client.close()
-                    agent._rebuild_anthropic_client()
-                else:
-                    _close_request_client_once("stream_interrupt_abort")
+                # #67142: kind-aware — anthropic aborts the request-local
+                # client's socket from this poll thread; the shared
+                # _anthropic_client is never closed here.
+                _close_request_client_once("stream_interrupt_abort")
             except Exception:
                 pass
             raise InterruptedError("Agent interrupted during streaming API call")
diff --git a/run_agent.py b/run_agent.py
index 8a92981e1dd2..ccb61a572aed 100644
--- a/run_agent.py
+++ b/run_agent.py
@@ -4259,6 +4259,103 @@ class AIAgent:
                 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 @@ class AIAgent:
             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)),
diff --git a/tests/agent/test_cascading_interrupt_6600.py b/tests/agent/test_cascading_interrupt_6600.py
index b335c6a6b17a..ce748f8dc196 100644
--- a/tests/agent/test_cascading_interrupt_6600.py
+++ b/tests/agent/test_cascading_interrupt_6600.py
@@ -132,3 +132,93 @@ def test_request_cancelled_token_is_request_local():
 
     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"
+    )
diff --git a/tests/run_agent/test_28161_anthropic_stream_pool_cleanup.py b/tests/run_agent/test_28161_anthropic_stream_pool_cleanup.py
index b0205ae9a29d..6d7f62408bbe 100644
--- a/tests/run_agent/test_28161_anthropic_stream_pool_cleanup.py
+++ b/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.
+"""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).
 
-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.
+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 @@ class TestAnthropicStreamPoolCleanup:
                 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 @@ class TestAnthropicStreamPoolCleanup:
         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 @@ class TestAnthropicStreamPoolCleanup:
             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 @@ class TestAnthropicStreamPoolCleanup:
                 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
diff --git a/tests/run_agent/test_run_agent.py b/tests/run_agent/test_run_agent.py
index 24c1030dc089..539dbdd20a40 100644
--- a/tests/run_agent/test_run_agent.py
+++ b/tests/run_agent/test_run_agent.py
@@ -7648,13 +7648,60 @@ class TestAnthropicInterruptHandler:
         assert "anthropic_messages" in source, \
             "interruptible_api_call must handle Anthropic interrupt (api_mode check)"
 
-    def test_interruptible_rebuilds_anthropic_client(self):
-        """After interrupting, the Anthropic client should be rebuilt."""
-        import inspect
+    def test_interruptible_anthropic_interrupt_never_closes_shared_client(self):
+        """#67142: a non-streaming Anthropic interrupt must abort the
+        request-local client from the poll thread, never close/rebuild the
+        shared _anthropic_client (which raced a live SSL BIO and corrupted an
+        unrelated SQLite DB via TLS-FD recycling).
+
+        Replaces the former source-reading assertion (which asserted the old,
+        now-removed rebuild-on-interrupt behavior) with a behavior test.
+        """
+        import threading
+        import time
+        from unittest.mock import MagicMock
+        from run_agent import AIAgent
         from agent.chat_completion_helpers import interruptible_api_call
-        source = inspect.getsource(interruptible_api_call)
-        assert "build_anthropic_client" in source, \
-            "interruptible_api_call must rebuild Anthropic client after interrupt"
+
+        agent = AIAgent(
+            api_key="test-key",
+            base_url="https://api.anthropic.com",
+            provider="anthropic",
+            model="claude-test",
+            quiet_mode=True,
+            skip_context_files=True,
+            skip_memory=True,
+        )
+        agent.api_mode = "anthropic_messages"
+        agent._interrupt_requested = False
+        agent._anthropic_client = MagicMock()
+        agent._rebuild_anthropic_client = MagicMock()
+        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 RuntimeError("forced close would have happened")
+
+        agent._anthropic_messages_create = MagicMock(side_effect=_create)
+
+        t0 = time.time()
+        with pytest.raises(InterruptedError):
+            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"
+        # The shared client is never closed/rebuilt from the poll thread.
+        agent._anthropic_client.close.assert_not_called()
+        agent._rebuild_anthropic_client.assert_not_called()
+        # The poll (stranger) thread aborts the request-local client's socket.
+        agent._abort_request_anthropic_client.assert_called_once_with(
+            request_client, reason="interrupt_abort"
+        )
 
     def test_streaming_has_anthropic_branch(self):
         """_streaming_api_call must also handle Anthropic interrupt."""
diff --git a/tests/run_agent/test_stream_stale_circuit_breaker.py b/tests/run_agent/test_stream_stale_circuit_breaker.py
index 4a90b27b7b82..19207e1f27aa 100644
--- a/tests/run_agent/test_stream_stale_circuit_breaker.py
+++ b/tests/run_agent/test_stream_stale_circuit_breaker.py
@@ -39,6 +39,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.
+    agent._create_request_anthropic_client = lambda *a, **k: agent._anthropic_client
     return agent
 
 
@@ -116,7 +119,10 @@ class TestStreamStaleCircuitBreaker:
 
         # Every attempt blocks, trips the stale detector, and fails.
         agent._anthropic_client.messages.stream.side_effect = _stream_side_effect
-        agent._anthropic_client.close.side_effect = unblock.set
+        # #67142: the stale detector now aborts the request-local client's
+        # sockets from the poll thread (not close() on the shared client), so
+        # unblock on the abort to simulate the socket shutdown waking the read.
+        agent._abort_request_anthropic_client = lambda *a, **k: unblock.set()
 
         with pytest.raises(Exception):
             agent._interruptible_streaming_api_call({})
diff --git a/tests/run_agent/test_streaming.py b/tests/run_agent/test_streaming.py
index 5f763f004495..55a0b9c008e7 100644
--- a/tests/run_agent/test_streaming.py
+++ b/tests/run_agent/test_streaming.py
@@ -1196,6 +1196,9 @@ class TestAnthropicStreamCallbacks:
 
         agent._anthropic_client = MagicMock()
         agent._anthropic_client.messages.stream.return_value = mock_stream
+        # #67142: streaming now runs on a request-local anthropic client; route
+        # it to the test mock so .messages.stream is exercised.
+        agent._create_request_anthropic_client = lambda *a, **k: agent._anthropic_client
 
         agent._interruptible_streaming_api_call({})
 
@@ -1251,16 +1254,18 @@ class TestAnthropicStreamCallbacks:
             _BadStream(),
             good_stream,
         ]
+        agent._create_request_anthropic_client = lambda *a, **k: agent._anthropic_client
 
         response = agent._interruptible_streaming_api_call({})
 
         assert response is final_message
         assert agent._anthropic_client.messages.stream.call_count == 2
-        # Anthropic-native cleanup: close + rebuild the Anthropic client, never
-        # the OpenAI primary client.
+        # #67142: cleanup runs on the request-local anthropic client (closed,
+        # worker-owned, via _close_request_client_once), never rebuilding the
+        # shared client and never touching the OpenAI primary client.
         assert mock_replace.call_count == 0
-        assert mock_rebuild.call_count == 1
-        assert agent._anthropic_client.close.call_count == 1
+        assert mock_rebuild.call_count == 0
+        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(
@@ -1286,6 +1291,7 @@ class TestAnthropicStreamCallbacks:
         agent._anthropic_client.messages.stream.side_effect = ValueError(
             "invalid local request shape"
         )
+        agent._create_request_anthropic_client = lambda *a, **k: agent._anthropic_client
 
         with pytest.raises(ValueError, match="invalid local request shape"):
             agent._interruptible_streaming_api_call({})
@@ -1326,15 +1332,17 @@ class TestAnthropicStreamCallbacks:
 
         agent._anthropic_client = MagicMock()
         agent._anthropic_client.messages.stream.return_value = empty_stream
+        agent._create_request_anthropic_client = lambda *a, **k: agent._anthropic_client
 
         with pytest.raises(EmptyStreamError):
             agent._interruptible_streaming_api_call({})
 
         assert agent._anthropic_client.messages.stream.call_count == 3
-        # Anthropic-native cleanup between attempts: rebuild the Anthropic
-        # client, never the OpenAI primary client.
+        # #67142: cleanup between attempts runs on the request-local anthropic
+        # client (fresh one built per attempt); the shared client is never
+        # rebuilt and the OpenAI primary client is never touched.
         assert mock_replace.call_count == 0
-        assert mock_rebuild.call_count == 2
+        assert mock_rebuild.call_count == 0
 
     @patch("run_agent.AIAgent._try_refresh_anthropic_client_credentials")
     @patch("run_agent.AIAgent._rebuild_anthropic_client")
@@ -1369,13 +1377,14 @@ class TestAnthropicStreamCallbacks:
 
         agent._anthropic_client = MagicMock()
         agent._anthropic_client.messages.stream.return_value = empty_stream
+        agent._create_request_anthropic_client = lambda *a, **k: agent._anthropic_client
 
         with pytest.raises(EmptyStreamError):
             agent._interruptible_streaming_api_call({})
 
         assert agent._anthropic_client.messages.stream.call_count == 3
         assert mock_replace.call_count == 0
-        assert mock_rebuild.call_count == 2
+        assert mock_rebuild.call_count == 0
 
 
 class TestPartialToolCallWarning: