TLS FD-recycling corruption of SQLite databases

Fixes on record

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

• #30858 fix(api-call): close FD-recycling race that wrote TLS bytes into kanban.db (#29507)

  merged to main

  files: agent/agent_runtime_helpers.py, agent/chat_completion_helpers.py, run_agent.py, tests/run_agent/test_create_openai_client_reuse.py, tests/run_agent/test_tls_fd_recycle_corruption.py

  Show the diff

  • agent/agent_runtime_helpers.py

    @@ -2132,33 +2132,56 @@ def apply_pending_steer_to_tool_results(agent, messages: list, num_tool_msgs: in
     
     
     def force_close_tcp_sockets(client: Any) -> int:
    -    """Force-close underlying TCP sockets to prevent CLOSE-WAIT accumulation.
    -
    -    When a provider drops a connection mid-stream, httpx's ``client.close()``
    -    performs a graceful shutdown which leaves sockets in CLOSE-WAIT until the
    -    OS times them out (often minutes).  This method walks the httpx transport
    -    pool and issues ``socket.shutdown(SHUT_RDWR)`` + ``socket.close()`` to
    -    force an immediate TCP RST, freeing the file descriptors.
    -
    -    Returns the number of sockets force-closed.
    +    """Abort in-flight TCP I/O by shutting down sockets WITHOUT closing FDs.
    +
    +    When a provider drops a connection mid-stream — or the user issues an
    +    interrupt — we want to unblock httpx's reader/writer immediately rather
    +    than waiting for the kernel's per-connection timeout. ``shutdown(SHUT_RDWR)``
    +    achieves that: it sends FIN, breaks any pending ``recv``/``send`` with EOF
    +    or ``EPIPE``, but does NOT release the file descriptor.
    +
    +    Historically this helper also called ``socket.close()`` so the FD got
    +    released immediately, but that's unsafe when (as is the case for both the
    +    interrupt-abort path and stale-call kill path) the helper runs on a
    +    different thread than the one driving the request:
    +
    +      * The Python ``socket.socket`` we close here is the SAME object held by
    +        httpx's pool, so closing it via Python sets its ``_fd`` to -1 and
    +        future operations on that Python object fail safely.
    +      * BUT the SSL wrapper (``ssl.SSLSocket``'s underlying OpenSSL ``BIO``)
    +        caches the raw integer FD. Once ``os.close(fd)`` runs, the kernel may
    +        immediately recycle that integer to the next ``open()`` call — e.g.
    +        the kanban dispatcher opening ``kanban.db``.
    +      * The owning worker thread then unwinds httpx, the SSL layer flushes a
    +        pending TLS record, and the encrypted bytes get written into the
    +        wrong file (issue #29507: 24-byte TLS application-data record
    +        clobbering SQLite header bytes 5..28).
    +
    +    The fix is to let the owning thread own the close. ``shutdown()`` from any
    +    thread is FD-safe; ``close()`` is not. The httpx connection's own close
    +    path — which runs from the worker thread when it unwinds — will release
    +    the FD via the same ``socket.socket`` object, and because Python's socket
    +    close atomically swaps ``_fd`` to -1 *before* issuing ``os.close``, there
    +    is no FD-aliasing window when only one thread closes.
    +
    +    Returns the number of sockets shut down. (Field kept as
    +    ``tcp_force_closed=N`` in the log line for backwards-compatible parsing.)
         """
         import socket as _socket
     
    -    closed = 0
    +    shutdown_count = 0
         try:
             for sock in _iter_pool_sockets(client):
                 try:
                     sock.shutdown(_socket.SHUT_RDWR)
                 except OSError:
    +                # Already shut down / not connected / FD invalid — all benign.
                     pass
    -            try:
    -                sock.close()
    -            except OSError:
    -                pass
    -            closed += 1
    +            # IMPORTANT (#29507): do NOT call sock.close() here. See docstring.
    +            shutdown_count += 1
         except Exception as exc:
             _ra().logger.debug("Force-close TCP sockets sweep error: %s", exc)
    -    return closed
    +    return shutdown_count
     
     
     

  • agent/chat_completion_helpers.py

    @@ -91,23 +91,55 @@ def interruptible_api_call(agent, api_kwargs: dict):
         provider fallback.
         """
         result = {"response": None, "error": None}
    -    request_client_holder = {"client": None}
    +    request_client_holder = {"client": None, "owner_tid": None}
         request_client_lock = threading.Lock()
     
         def _set_request_client(client):
             with request_client_lock:
                 request_client_holder["client"] = client
    +            # #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()``.
    +            request_client_holder["owner_tid"] = threading.get_ident()
             return client
     
         def _take_request_client():
             with request_client_lock:
                 client = request_client_holder.get("client")
                 request_client_holder["client"] = None
    +            request_client_holder["owner_tid"] = None
                 return client
     
         def _close_request_client_once(reason: str) -> None:
    -        request_client = _take_request_client()
    -        if request_client is not None:
    +        # #29507: dispatch on the calling thread.
    +        #
    +        # When ``_call`` (the worker) reaches its ``finally`` it owns the
    +        # close and we pop + fully close as before. When a *stranger* thread
    +        # (the interrupt-check loop, the stale-call detector) drives the
    +        # close, only shut the sockets down so the worker's blocked
    +        # ``recv``/``send`` unwinds with an ``EPIPE`` / EOF — and let the
    +        # worker close ``client`` from its own thread on its way out. That
    +        # avoids the FD-recycling race where the kernel reassigned a
    +        # just-closed TLS socket FD to ``kanban.db``, and the still-live SSL
    +        # BIO on the worker thread then wrote a 24-byte TLS application-data
    +        # record into the SQLite header (#29507).
    +        with request_client_lock:
    +            request_client = request_client_holder.get("client")
    +            owner_tid = request_client_holder.get("owner_tid")
    +            stranger_thread = (
    +                request_client is not None
    +                and owner_tid is not None
    +                and owner_tid != threading.get_ident()
    +            )
    +            if not stranger_thread:
    +                # Owning thread (or no recorded owner) → pop and fully close.
    +                request_client_holder["client"] = None
    +                request_client_holder["owner_tid"] = None
    +        if request_client is None:
    +            return
    +        if stranger_thread:
    +            agent._abort_request_openai_client(request_client, reason=reason)
    +        else:
                 agent._close_request_openai_client(request_client, reason=reason)
     
         def _call():
    @@ -1271,23 +1303,44 @@ def _on_reasoning(text):
             return result["response"]
     
         result = {"response": None, "error": None, "partial_tool_names": []}
    -    request_client_holder = {"client": None, "diag": None}
    +    request_client_holder = {"client": None, "diag": None, "owner_tid": None}
         request_client_lock = threading.Lock()
     
         def _set_request_client(client):
             with request_client_lock:
                 request_client_holder["client"] = client
    +            # See #29507 explanation in the non-streaming variant above.
    +            request_client_holder["owner_tid"] = threading.get_ident()
             return client
     
         def _take_request_client():
             with request_client_lock:
                 client = request_client_holder.get("client")
                 request_client_holder["client"] = None
    +            request_client_holder["owner_tid"] = None
                 return client
     
         def _close_request_client_once(reason: str) -> None:
    -        request_client = _take_request_client()
    -        if request_client is not None:
    +        # See #29507 explanation in the non-streaming variant above. A
    +        # stranger thread (the interrupt-check / stale-stream detector loop)
    +        # only aborts sockets — never pops, never calls ``client.close()`` —
    +        # so the worker thread retains ownership of the FD release.
    +        with request_client_lock:
    +            request_client = request_client_holder.get("client")
    +            owner_tid = request_client_holder.get("owner_tid")
    +            stranger_thread = (
    +                request_client is not None
    +                and owner_tid is not None
    +                and owner_tid != threading.get_ident()
    +            )
    +            if not stranger_thread:
    +                request_client_holder["client"] = None
    +                request_client_holder["owner_tid"] = None
    +        if request_client is None:
    +            return
    +        if stranger_thread:
    +            agent._abort_request_openai_client(request_client, reason=reason)
    +        else:
                 agent._close_request_openai_client(request_client, reason=reason)
     
         first_delta_fired = {"done": False}

  • run_agent.py

    @@ -2563,6 +2563,39 @@ def _create_request_openai_client(self, *, reason: str, api_kwargs: Optional[dic
         def _close_request_openai_client(self, client: Any, *, reason: str) -> None:
             self._close_openai_client(client, reason=reason, shared=False)
     
    +    def _abort_request_openai_client(self, client: Any, *, reason: str) -> None:
    +        """Cross-thread abort: shut sockets down without releasing FDs.
    +
    +        Companion to :meth:`_close_request_openai_client` for stranger-thread
    +        callers (interrupt-check loop, stale-call detector). Calling
    +        ``client.close()`` from a thread that does not own the active httpx
    +        connection raced the still-live SSL BIO and corrupted unrelated file
    +        descriptors when the kernel recycled the just-freed TCP FD (#29507).
    +
    +        Here we only ``shutdown(SHUT_RDWR)`` the pool's sockets. That unblocks
    +        the owning worker thread's pending ``recv``/``send`` with an EOF or
    +        ``EPIPE`` so it can unwind and close ``client`` from its own context
    +        — which is where the FD release belongs.
    +        """
    +        if client is None:
    +            return
    +        try:
    +            shutdown_count = self._force_close_tcp_sockets(client)
    +            logger.info(
    +                "OpenAI client aborted (%s, shared=False, tcp_force_closed=%d, "
    +                "deferred_close=stranger_thread) %s",
    +                reason,
    +                shutdown_count,
    +                self._client_log_context(),
    +            )
    +        except Exception as exc:
    +            logger.debug(
    +                "OpenAI client abort failed (%s, shared=False) %s error=%s",
    +                reason,
    +                self._client_log_context(),
    +                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

  • tests/run_agent/test_create_openai_client_reuse.py

    @@ -190,7 +190,13 @@ def test_replace_primary_openai_client_survives_repeated_rebuilds():
     
     
     def test_force_close_tcp_sockets_descends_httpcore_1_connection_wrapper():
    -    """httpcore 1.x stores the real stream below conn._connection."""
    +    """httpcore 1.x stores the real stream below conn._connection.
    +
    +    Post-#29507: the helper must shut sockets down but must NOT release the
    +    FD via ``sock.close()`` — that race recycled FDs into unrelated file
    +    descriptors (kanban.db) and let TLS bytes overwrite SQLite headers. The
    +    owning httpx thread is responsible for closing FDs on its own unwind.
    +    """
         from agent.agent_runtime_helpers import force_close_tcp_sockets
     
         class FakeSocket:
    @@ -215,4 +221,6 @@ def close(self):
     
         assert force_close_tcp_sockets(openai_client) == 1
         assert sock.shutdown_calls == 1
    -    assert sock.close_calls == 1
    +    # #29507: close() must NOT be called from this helper — the owning
    +    # httpx worker thread releases the FD, not us.
    +    assert sock.close_calls == 0

  • tests/run_agent/test_tls_fd_recycle_corruption.py

    @@ -0,0 +1,454 @@
    +"""Regressions for issue #29507 — cross-thread close of the per-request OpenAI
    +client could release a TLS socket FD whose integer was still cached in the
    +owning httpx worker's SSL BIO. The kernel then recycled the FD into the next
    +``open()`` (e.g. the kanban dispatcher's ``kanban.db``), and the worker's
    +delayed TLS flush wrote a 24-byte TLS application-data record on top of the
    +SQLite header.
    +
    +The fix has two prongs:
    +
    +1. ``force_close_tcp_sockets`` no longer calls ``sock.close()`` — only
    +   ``shutdown(SHUT_RDWR)``. Shutdown unblocks the worker's pending
    +   ``recv``/``send`` without releasing the FD.
    +
    +2. ``_close_request_client_once`` is thread-aware: a stranger thread (the
    +   interrupt-check / stale-call loop) only aborts the sockets and leaves
    +   the client in the holder; the worker's own ``finally`` performs the
    +   actual ``client.close()`` from its own thread context.
    +
    +Both prongs together close the FD-recycling window. The tests below pin
    +each prong individually and one end-to-end test simulates the reporter's
    +timeline at object granularity (no network, no real sockets).
    +"""
    +from __future__ import annotations
    +
    +import logging
    +import socket as _socket
    +import threading
    +from types import SimpleNamespace
    +from unittest.mock import MagicMock, patch
    +
    +import pytest
    +
    +
    +# ---------------------------------------------------------------------------
    +# Prong 1: force_close_tcp_sockets must NOT release file descriptors.
    +# ---------------------------------------------------------------------------
    +
    +
    +class _FakeSocket:
    +    """Records shutdown/close calls without touching real FDs."""
    +
    +    def __init__(self):
    +        self.shutdown_calls = 0
    +        self.close_calls = 0
    +
    +    def shutdown(self, _how):
    +        self.shutdown_calls += 1
    +
    +    def close(self):
    +        self.close_calls += 1
    +
    +
    +def _build_fake_client(sock):
    +    """Mimic the httpcore-1 layout that ``_iter_pool_sockets`` walks."""
    +    stream = SimpleNamespace(_sock=sock)
    +    http11 = SimpleNamespace(_network_stream=stream)
    +    pool_entry = SimpleNamespace(_connection=http11)
    +    pool = SimpleNamespace(_connections=[pool_entry])
    +    transport = SimpleNamespace(_pool=pool)
    +    http_client = SimpleNamespace(_transport=transport)
    +    return SimpleNamespace(_client=http_client)
    +
    +
    +def test_force_close_tcp_sockets_shutdown_only_no_close():
    +    """The smoking-gun guarantee: shutdown is called, close is NOT.
    +
    +    If a future refactor reintroduces ``sock.close()`` here, the
    +    FD-recycling race that corrupted ``kanban.db`` (issue #29507) will
    +    re-open. Pin the contract explicitly.
    +    """
    +    from agent.agent_runtime_helpers import force_close_tcp_sockets
    +
    +    sock = _FakeSocket()
    +    client = _build_fake_client(sock)
    +
    +    n = force_close_tcp_sockets(client)
    +
    +    assert n == 1
    +    assert sock.shutdown_calls == 1, "shutdown() must run — it's how we unblock the worker"
    +    assert sock.close_calls == 0, (
    +        "close() must NOT run from this helper — releasing the FD here is the "
    +        "race that wrote TLS bytes into kanban.db (#29507)"
    +    )
    +
    +
    +def test_force_close_tcp_sockets_uses_shut_rdwr():
    +    """Both directions must be shut down so the SSL state machine fully unwinds.
    +
    +    Half-close (e.g. SHUT_WR only) wouldn't unblock a worker blocked in
    +    ``recv``, defeating the whole point of the helper.
    +    """
    +    from agent.agent_runtime_helpers import force_close_tcp_sockets
    +
    +    captured = []
    +
    +    class _ProbingSocket:
    +        def shutdown(self, how):
    +            captured.append(how)
    +
    +        def close(self):  # pragma: no cover — must not run, asserted below
    +            captured.append("CLOSE_CALLED")
    +
    +    sock = _ProbingSocket()
    +    client = _build_fake_client(sock)
    +
    +    force_close_tcp_sockets(client)
    +
    +    assert captured == [_socket.SHUT_RDWR]
    +
    +
    +def test_force_close_tcp_sockets_swallows_oserror_on_shutdown():
    +    """A socket already shut down / not connected raises ``OSError`` — benign."""
    +    from agent.agent_runtime_helpers import force_close_tcp_sockets
    +
    +    class _AlreadyShut:
    +        def shutdown(self, _how):
    +            raise OSError("not connected")
    +
    +        def close(self):  # pragma: no cover — must not run
    +            raise AssertionError("close() must not be called")
    +
    +    client = _build_fake_client(_AlreadyShut())
    +
    +    # No exception escapes; the helper still counts the socket as handled.
    +    assert force_close_tcp_sockets(client) == 1
    +
    +
    +def test_force_close_tcp_sockets_handles_multiple_pool_entries():
    +    """Walk every pool connection — the bug equally applies to all of them."""
    +    from agent.agent_runtime_helpers import force_close_tcp_sockets
    +
    +    socks = [_FakeSocket(), _FakeSocket(), _FakeSocket()]
    +    entries = [
    +        SimpleNamespace(_connection=SimpleNamespace(_network_stream=SimpleNamespace(_sock=s)))
    +        for s in socks
    +    ]
    +    pool = SimpleNamespace(_connections=entries)
    +    transport = SimpleNamespace(_pool=pool)
    +    http_client = SimpleNamespace(_transport=transport)
    +    client = SimpleNamespace(_client=http_client)
    +
    +    assert force_close_tcp_sockets(client) == 3
    +    for s in socks:
    +        assert s.shutdown_calls == 1
    +        assert s.close_calls == 0
    +
    +
    +# ---------------------------------------------------------------------------
    +# Prong 2: _close_request_client_once is thread-aware.
    +# ---------------------------------------------------------------------------
    +
    +
    +def _make_agent_mock():
    +    """Minimal agent with the two close primitives stubbed for spy-style checks."""
    +    agent = MagicMock()
    +    agent._interrupt_requested = False
    +    agent._close_request_openai_client = MagicMock()
    +    agent._abort_request_openai_client = MagicMock()
    +    return agent
    +
    +
    +def _call_inside_owner_thread(callable_):
    +    """Run callable_ on a separate thread so its ``threading.get_ident()``
    +    differs from the test thread."""
    +    result = {"value": None, "exc": None}
    +
    +    def runner():
    +        try:
    +            result["value"] = callable_()
    +        except BaseException as e:  # noqa: BLE001 — propagate test failures faithfully
    +            result["exc"] = e
    +
    +    t = threading.Thread(target=runner)
    +    t.start()
    +    t.join(timeout=5.0)
    +    if result["exc"] is not None:
    +        raise result["exc"]
    +    return result["value"]
    +
    +
    +def test_close_from_stranger_thread_aborts_only_no_close():
    +    """Stranger-thread close → ``_abort_request_openai_client``, holder NOT popped.
    +
    +    Reproduces the asyncio_0 → Thread-1616 interrupt path. After this call
    +    the worker's eventual ``finally`` must still see the client in the
    +    holder so IT can be the one releasing the FD.
    +    """
    +    from agent.chat_completion_helpers import interruptible_api_call
    +
    +    # We can't easily invoke just `_close_request_client_once` because it's
    +    # a closure local to ``interruptible_api_call``. Re-extract the same
    +    # logic by exercising it through a fake worker that lets us drive the
    +    # holder state manually.
    +    agent = _make_agent_mock()
    +    # Pretend ``_call`` ran far enough to set the client on the holder
    +    # from the owner thread.
    +    sentinel = object()
    +    owner_tid_holder = {"tid": None, "client_present_after_stranger_close": False}
    +
    +    def _owner_workload(holder, lock):
    +        # Owner-thread set
    +        with lock:
    +            holder["client"] = sentinel
    +            holder["owner_tid"] = threading.get_ident()
    +        owner_tid_holder["tid"] = threading.get_ident()
    +
    +    holder = {"client": None, "owner_tid": None}
    +    lock = threading.Lock()
    +    _call_inside_owner_thread(lambda: _owner_workload(holder, lock))
    +
    +    # Now drive the exact body of the post-#29507 ``_close_request_client_once``
    +    # from the test thread (stranger) and from the owner thread.
    +    def close_once(holder, lock, reason):
    +        with lock:
    +            request_client = holder.get("client")
    +            owner_tid = holder.get("owner_tid")
    +            stranger = (
    +                request_client is not None
    +                and owner_tid is not None
    +                and owner_tid != threading.get_ident()
    +            )
    +            if not stranger:
    +                holder["client"] = None
    +                holder["owner_tid"] = None
    +        if request_client is None:
    +            return None
    +        if stranger:
    +            agent._abort_request_openai_client(request_client, reason=reason)
    +            return "aborted"
    +        agent._close_request_openai_client(request_client, reason=reason)
    +        return "closed"
    +
    +    outcome = close_once(holder, lock, "interrupt_abort")
    +
    +    assert outcome == "aborted"
    +    agent._abort_request_openai_client.assert_called_once()
    +    agent._close_request_openai_client.assert_not_called()
    +    # Holder is still populated — the worker thread will pick this up in
    +    # its ``finally`` and own the actual ``client.close()``.
    +    assert holder["client"] is sentinel
    +    assert holder["owner_tid"] == owner_tid_holder["tid"]
    +
    +
    +def test_close_from_owner_thread_pops_and_full_close():
    +    """Worker-thread close → ``_close_request_openai_client``, holder popped."""
    +    agent = _make_agent_mock()
    +    sentinel = object()
    +    holder = {"client": None, "owner_tid": None}
    +    lock = threading.Lock()
    +
    +    def workload():
    +        with lock:
    +            holder["client"] = sentinel
    +            holder["owner_tid"] = threading.get_ident()
    +
    +        # Same body inlined here so the test thread and the closing thread
    +        # are identical (owner == self).
    +        with lock:
    +            request_client = holder.get("client")
    +            owner_tid = holder.get("owner_tid")
    +            stranger = (
    +                request_client is not None
    +                and owner_tid is not None
    +                and owner_tid != threading.get_ident()
    +            )
    +            if not stranger:
    +                holder["client"] = None
    +                holder["owner_tid"] = None
    +        if request_client is None:
    +            return None
    +        if stranger:
    +            agent._abort_request_openai_client(request_client, reason="request_complete")
    +            return "aborted"
    +        agent._close_request_openai_client(request_client, reason="request_complete")
    +        return "closed"
    +
    +    outcome = _call_inside_owner_thread(workload)
    +
    +    assert outcome == "closed"
    +    agent._close_request_openai_client.assert_called_once()
    +    agent._abort_request_openai_client.assert_not_called()
    +    assert holder["client"] is None
    +    assert holder["owner_tid"] is None
    +
    +
    +def test_stranger_then_owner_close_sequence_runs_full_close_exactly_once():
    +    """Stranger abort followed by owner close → full close runs once.
    +
    +    This mirrors the reporter's timeline: asyncio_0 fires interrupt_abort
    +    (stranger → abort only), then Thread-1616 unwinds and its finally
    +    fires request_complete (owner → full close). Net result must be one
    +    abort + one full close, with the holder ending empty.
    +    """
    +    agent = _make_agent_mock()
    +    sentinel = object()
    +    holder = {"client": None, "owner_tid": None}
    +    lock = threading.Lock()
    +
    +    def close_once(reason):
    +        with lock:
    +            request_client = holder.get("client")
    +            owner_tid = holder.get("owner_tid")
    +            stranger = (
    +                request_client is not None
    +                and owner_tid is not None
    +                and owner_tid != threading.get_ident()
    +            )
    +            if not stranger:
    +                holder["client"] = None
    +                holder["owner_tid"] = None
    +        if request_client is None:
    +            return
    +        if stranger:
    +            agent._abort_request_openai_client(request_client, reason=reason)
    +        else:
    +            agent._close_request_openai_client(request_client, reason=reason)
    +
    +    def owner_workload():
    +        # Set client from owner thread.
    +        with lock:
    +            holder["client"] = sentinel
    +            holder["owner_tid"] = threading.get_ident()
    +        # Simulate work being interrupted by a stranger from outside.
    +        nonlocal_stranger_event.wait(timeout=2.0)
    +        # Worker unwinds — its finally calls close once.
    +        close_once("request_complete")
    +
    +    nonlocal_stranger_event = threading.Event()
    +    owner = threading.Thread(target=owner_workload)
    +    owner.start()
    +
    +    # Test thread plays the stranger.
    +    # Give the owner a moment to set the holder.
    +    import time as _t
    +    _t.sleep(0.05)
    +    close_once("interrupt_abort")
    +    nonlocal_stranger_event.set()
    +    owner.join(timeout=5.0)
    +
    +    assert not owner.is_alive(), "owner thread hung past join timeout"
    +
    +    # The fix's intended outcome: abort once, close once, holder empty.
    +    assert agent._abort_request_openai_client.call_count == 1
    +    assert agent._close_request_openai_client.call_count == 1
    +    assert holder["client"] is None
    +    assert holder["owner_tid"] is None
    +
    +
    +# ---------------------------------------------------------------------------
    +# End-to-end: the agent's ``_abort_request_openai_client`` shuts sockets and
    +# logs deferred_close=stranger_thread without ever calling client.close().
    +# ---------------------------------------------------------------------------
    +
    +
    +def test_agent_abort_request_openai_client_does_not_call_client_close(caplog):
    +    """``_abort_request_openai_client`` must shutdown sockets but NEVER close().
    +
    +    This is the actual entry point used by the stranger-thread path. If a
    +    future refactor accidentally wires it back to ``_close_openai_client``
    +    the FD race is back. Pin both the shutdown side-effect AND the absence
    +    of any ``client.close()`` call.
    +    """
    +    from run_agent import AIAgent
    +
    +    sock = _FakeSocket()
    +    client = _build_fake_client(sock)
    +
    +    # ``client.close()`` would mutate the holder if invoked — give it a
    +    # MagicMock spy so we can assert no call.
    +    client.close = MagicMock()
    +
    +    agent = AIAgent.__new__(AIAgent)
    +    agent._client_log_context = lambda: "provider=test"
    +
    +    with caplog.at_level(logging.INFO, logger="run_agent"):
    +        agent._abort_request_openai_client(client, reason="interrupt_abort")
    +
    +    # Sockets shut down (one in our fake pool).
    +    assert sock.shutdown_calls == 1
    +    assert sock.close_calls == 0
    +    # And critically: client.close() never ran here.
    +    client.close.assert_not_called()
    +
    +    # The log line is parseable: same ``tcp_force_closed=N`` field shape as
    +    # the existing ``close`` log so dashboards keep working, plus a
    +    # ``deferred_close=stranger_thread`` marker to make the new path
    +    # observable in production triage.
    +    msgs = [r.getMessage() for r in caplog.records]
    +    assert any(
    +        "OpenAI client aborted (interrupt_abort" in m
    +        and "tcp_force_closed=1" in m
    +        and "deferred_close=stranger_thread" in m
    +        for m in msgs
    +    ), f"missing abort log line; got: {msgs!r}"
    +
    +
    +def test_agent_abort_request_openai_client_null_client_is_noop():
    +    """A ``None`` client must short-circuit cleanly (defensive)."""
    +    from run_agent import AIAgent
    +
    +    agent = AIAgent.__new__(AIAgent)
    +    agent._client_log_context = lambda: "provider=test"
    +
    +    # No exception, no side effect.
    +    agent._abort_request_openai_client(None, reason="interrupt_abort")
    +
    +
    +# ---------------------------------------------------------------------------
    +# FD-recycling proof: when shutdown-only is honored, a stranger-thread abort
    +# CANNOT release an FD that the owning thread still references.
    +# ---------------------------------------------------------------------------
    +
    +
    +def test_fd_recycle_window_closed_by_shutdown_only():
    +    """Construct the exact race the reporter saw — abort from a stranger
    +    thread, then have the (simulated) kernel recycle the FD into a new file.
    +    With the fix, the worker's surviving socket reference cannot be
    +    confused with the recycled file descriptor.
    +    """
    +    from agent.agent_runtime_helpers import force_close_tcp_sockets
    +
    +    # Tracks "was the FD released by the abort path?" — that is the only
    +    # signal the kernel needs to recycle the integer to a new ``open()``.
    +    fd_released = {"yes": False}
    +
    +    class _OwnedSocket:
    +        """Simulates a socket whose FD is shared with the owner's SSL BIO.
    +
    +        ``close`` flips ``fd_released`` so the test can assert that with
    +        the fix the abort path NEVER releases the FD (and therefore the
    +        kernel never recycles it under the owner's still-active reference).
    +        """
    +
    +        def __init__(self):
    +            self.shutdowns = 0
    +
    +        def shutdown(self, _how):
    +            self.shutdowns += 1
    +
    +        def close(self):
    +            fd_released["yes"] = True
    +
    +    sock = _OwnedSocket()
    +    client = _build_fake_client(sock)
    +
    +    # Stranger thread runs the abort sweep (== what asyncio_0 did in the
    +    # reporter's session).
    +    _call_inside_owner_thread(lambda: force_close_tcp_sockets(client))
    +
    +    assert sock.shutdowns == 1, "shutdown must wake the worker"
    +    assert fd_released["yes"] is False, (
    +        "force_close_tcp_sockets released the FD from a stranger thread — "
    +        "this is exactly the #29507 race. The owner thread must own close()."
    +    )

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

  merged to main

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

  Show the diff

  • 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 _abort_active_request(reason: str) -> None:
             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 _close_request_client_once(reason: str) -> None:
                     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 _call():
                         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 _call():
                 )
                 # 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 _on_reasoning(text):
         _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 _on_reasoning(text):
         # 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 _close_request_client_once(reason: str) -> None:
                     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 _call_chat_completions(stream_attempt_id: int):
                 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 _call_anthropic():
                 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 _call():
                         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 _call():
                             )
                             _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 _call():
                                 # 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 _call():
                 # 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 _call():
                 )
                 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")

  • 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 = "[redacted:secret-assignment]"
    +    # #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

  • tests/run_agent/test_run_agent.py

    @@ -7648,13 +7648,60 @@ def test_interruptible_has_anthropic_branch(self):
             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="[redacted:secret-assignment]",
    +            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."""

  • 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 = "[redacted:secret-assignment]"
    +    # #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 @@ def _stream_side_effect(*args, **kwargs):
     
             # 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({})

  • tests/run_agent/test_streaming.py

    @@ -1196,6 +1196,9 @@ def test_anthropic_stream_refreshes_activity_on_every_event(self):
     
             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 @@ def __iter__(self):
                 _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 @@ def test_generic_anthropic_valueerror_still_propagates_without_stream_retry(
             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 @@ def test_anthropic_zero_event_stream_retried_as_transient(
     
             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 @@ def test_anthropic_eventless_sdk_assertion_normalized_to_empty_stream(
     
             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: