fix(agent): release pool FDs on owning-thread client close (#61979)

force_close_tcp_sockets stayed shutdown-only after #29507 to avoid
cross-thread FD recycle. That left CLOSED sockets unreclaimed when
httpx.close() skipped already-shutdown sockets under long-lived
gateways (~1 CLOSED fd / 6 min via proxy).

Add release_fds= for the owning-thread dispose path only; abort still
defaults to shutdown-only.

agent/agent_runtime_helpers.py

@@ -3161,21 +3161,20 @@ def apply_pending_steer_to_tool_results(agent, messages: list, num_tool_msgs: in
 
 
 
-def force_close_tcp_sockets(client: Any) -> int:
-    """Abort in-flight TCP I/O by shutting down sockets WITHOUT closing FDs.
+def force_close_tcp_sockets(client: Any, *, release_fds: bool = False) -> int:
+    """Abort in-flight TCP I/O by shutting down pool sockets.
 
     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.
+    or ``EPIPE``.
 
-    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:
+    By default (``release_fds=False``) this helper does **not** call
+    ``socket.close()`` / release the FD. That default is load-bearing for
+    cross-thread abort paths (#29507):
 
-      * The Python ``socket.socket`` we close here is the SAME object held by
+      * The Python ``socket.socket`` we close 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``)
@@ -3187,15 +3186,20 @@ def force_close_tcp_sockets(client: Any) -> int:
         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.
+    ``shutdown()`` from any thread is FD-safe; ``close()`` is not when a
+    stranger thread still has the BIO holding the raw FD.
 
-    Returns the number of sockets shut down. (Field kept as
-    ``tcp_force_closed=N`` in the log line for backwards-compatible parsing.)
+    When the **owning** thread is disposing of a client that is no longer
+    shared (``_close_openai_client`` after replace / request-complete), pass
+    ``release_fds=True``. httpx's own ``client.close()`` does not reliably
+    ``os.close()`` sockets that were already ``shutdown()``'d, so without an
+    explicit ``sock.close()`` those FDs stay in kernel CLOSED state forever
+    and accumulate under long-lived gateways (issue #61979 — ~1 CLOSED fd
+    per ~6 minutes through a local proxy path).
+
+    Returns the number of sockets shut down (and optionally closed). Field
+    kept as ``tcp_force_closed=N`` in log lines for backwards-compatible
+    parsing.
     """
     import socket as _socket
 
@@ -3207,7 +3211,13 @@ def force_close_tcp_sockets(client: Any) -> int:
             except OSError:
                 # Already shut down / not connected / FD invalid — all benign.
                 pass
-            # IMPORTANT (#29507): do NOT call sock.close() here. See docstring.
+            # IMPORTANT (#29507): never release FDs from stranger-thread
+            # abort paths. Only the owning-thread close path may opt in.
+            if release_fds:
+                try:
+                    sock.close()
+                except OSError:
+                    pass
             shutdown_count += 1
     except Exception as exc:
         _ra().logger.debug("Force-close TCP sockets sweep error: %s", exc)

run_agent.py

@@ -3990,17 +3990,18 @@ class AIAgent:
         return create_openai_client(self, client_kwargs, reason=reason, shared=shared)
 
     @staticmethod
-    def _force_close_tcp_sockets(client: Any) -> int:
+    def _force_close_tcp_sockets(client: Any, *, release_fds: bool = False) -> int:
         """Forwarder — see ``agent.agent_runtime_helpers.force_close_tcp_sockets``."""
         from agent.agent_runtime_helpers import force_close_tcp_sockets
-        return force_close_tcp_sockets(client)
+        return force_close_tcp_sockets(client, release_fds=release_fds)
 
     def _close_openai_client(self, client: Any, *, reason: str, shared: bool) -> None:
         if client is None:
             return
-        # Force-close TCP sockets first to prevent CLOSE-WAIT accumulation,
-        # then do the graceful SDK-level close.
-        force_closed = self._force_close_tcp_sockets(client)
+        # Owning-thread dispose: shutdown + release FDs so already-shutdown
+        # sockets don't linger in kernel CLOSED state (#61979). Cross-thread
+        # abort uses _abort_request_openai_client (release_fds=False) instead.
+        force_closed = self._force_close_tcp_sockets(client, release_fds=True)
         try:
             client.close()
             logger.info(

tests/run_agent/test_tls_fd_recycle_corruption.py

@@ -7,9 +7,11 @@ 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.
+1. ``force_close_tcp_sockets`` defaults to ``shutdown(SHUT_RDWR)`` only —
+   no ``sock.close()``. Shutdown unblocks the worker's pending
+   ``recv``/``send`` without releasing the FD. Owning-thread dispose
+   opts into ``release_fds=True`` so already-shutdown sockets do not
+   accumulate as kernel CLOSED fds (#61979).
 
 2. ``_close_request_client_once`` is thread-aware: a stranger thread (the
    interrupt-check / stale-call loop) only aborts the sockets and leaves
@@ -61,11 +63,10 @@ def _build_fake_client(sock):
 
 
 def test_force_close_tcp_sockets_shutdown_only_no_close():
-    """The smoking-gun guarantee: shutdown is called, close is NOT.
+    """Default path: 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.
+    Stranger-thread abort (#29507) relies on this default. Releasing the
+    FD from a non-owning thread re-opens the TLS→SQLite corruption race.
     """
     from agent.agent_runtime_helpers import force_close_tcp_sockets
 
@@ -77,11 +78,30 @@ def test_force_close_tcp_sockets_shutdown_only_no_close():
     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 "
+        "close() must NOT run by default — releasing the FD here is the "
         "race that wrote TLS bytes into kanban.db (#29507)"
     )
 
 
+def test_force_close_tcp_sockets_release_fds_closes_after_shutdown():
+    """Owning-thread dispose path (#61979): shutdown then close.
+
+    httpx does not reliably ``os.close()`` sockets that were already
+    shutdown, so owner-thread close must release FDs explicitly or they
+    accumulate in kernel CLOSED state under long-lived gateways.
+    """
+    from agent.agent_runtime_helpers import force_close_tcp_sockets
+
+    sock = _FakeSocket()
+    client = _build_fake_client(sock)
+
+    n = force_close_tcp_sockets(client, release_fds=True)
+
+    assert n == 1
+    assert sock.shutdown_calls == 1
+    assert sock.close_calls == 1
+
+
 def test_force_close_tcp_sockets_uses_shut_rdwr():
     """Both directions must be shut down so the SSL state machine fully unwinds.
 
@@ -115,7 +135,7 @@ def test_force_close_tcp_sockets_swallows_oserror_on_shutdown():
         def shutdown(self, _how):
             raise OSError("not connected")
 
-        def close(self):  # pragma: no cover — must not run
+        def close(self):  # pragma: no cover — must not run on default path
             raise AssertionError("close() must not be called")
 
     client = _build_fake_client(_AlreadyShut())
@@ -124,6 +144,27 @@ def test_force_close_tcp_sockets_swallows_oserror_on_shutdown():
     assert force_close_tcp_sockets(client) == 1
 
 
+def test_force_close_tcp_sockets_release_fds_swallows_oserror_on_close():
+    """Already-closed sockets must not abort the owning-thread sweep."""
+    from agent.agent_runtime_helpers import force_close_tcp_sockets
+
+    class _CloseRaises:
+        def __init__(self):
+            self.shutdown_calls = 0
+
+        def shutdown(self, _how):
+            self.shutdown_calls += 1
+
+        def close(self):
+            raise OSError("Bad file descriptor")
+
+    sock = _CloseRaises()
+    client = _build_fake_client(sock)
+
+    assert force_close_tcp_sockets(client, release_fds=True) == 1
+    assert sock.shutdown_calls == 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
@@ -143,6 +184,20 @@ def test_force_close_tcp_sockets_handles_multiple_pool_entries():
         assert s.shutdown_calls == 1
         assert s.close_calls == 0
 
+    # Owning-thread dispose releases every pool entry's FD.
+    socks2 = [_FakeSocket(), _FakeSocket(), _FakeSocket()]
+    entries2 = [
+        SimpleNamespace(_connection=SimpleNamespace(_network_stream=SimpleNamespace(_sock=s)))
+        for s in socks2
+    ]
+    client2 = SimpleNamespace(
+        _client=SimpleNamespace(_transport=SimpleNamespace(_pool=SimpleNamespace(_connections=entries2)))
+    )
+    assert force_close_tcp_sockets(client2, release_fds=True) == 3
+    for s in socks2:
+        assert s.shutdown_calls == 1
+        assert s.close_calls == 1
+
 
 # ---------------------------------------------------------------------------
 # Prong 2: _close_request_client_once is thread-aware.
@@ -392,6 +447,38 @@ def test_agent_abort_request_openai_client_does_not_call_client_close(caplog):
     ), f"missing abort log line; got: {msgs!r}"
 
 
+def test_agent_close_openai_client_releases_fds(caplog):
+    """Owning-thread ``_close_openai_client`` must release pool FDs (#61979).
+
+    After #29507 the helper defaulted to shutdown-only; combined with
+    httpx not closing already-shutdown sockets, gateway processes leaked
+    CLOSED fds until rlimit. The owner dispose path must opt into
+    ``release_fds=True``.
+    """
+    from run_agent import AIAgent
+
+    sock = _FakeSocket()
+    client = _build_fake_client(sock)
+    client.close = MagicMock()
+
+    agent = AIAgent.__new__(AIAgent)
+    agent._client_log_context = lambda: "provider=test"
+
+    with caplog.at_level(logging.INFO, logger="run_agent"):
+        agent._close_openai_client(client, reason="replace:test", shared=True)
+
+    assert sock.shutdown_calls == 1
+    assert sock.close_calls == 1
+    client.close.assert_called_once()
+
+    msgs = [r.getMessage() for r in caplog.records]
+    assert any(
+        "OpenAI client closed (replace:test" in m
+        and "tcp_force_closed=1" in m
+        for m in msgs
+    ), f"missing close 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