Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 46 additions & 4 deletions agent/auxiliary_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -1138,18 +1138,47 @@ def create(self, **kwargs) -> Any:
deadline = time.monotonic() + float(total_timeout) if total_timeout else None
timed_out = threading.Event()
timeout_timer: Optional[threading.Timer] = None
# The thread driving this request owns its transport's file
# descriptors — see the ownership note in _close_client_on_timeout.
owner_tid = threading.get_ident()

def _timeout_message() -> str:
return f"Codex auxiliary Responses stream exceeded {float(total_timeout):.1f}s total timeout"

def _close_client_on_timeout() -> None:
timed_out.set()
close = getattr(self._client, "close", None)
if callable(close):
# FD-ownership contract (#29507 / #67142 / #70773): only the thread
# driving the request may RELEASE this client's file descriptors.
# This callback has two callers — ``_check_cancelled()`` on the
# owning thread, and the daemon ``threading.Timer`` below, which is
# a stranger thread. From a stranger thread we may only
# ``shutdown()`` the pooled sockets: ``close()`` releases the raw
# TLS fd while the owner's OpenSSL BIO still caches that integer,
# the kernel recycles it into the next ``open()`` in this process
# (a SessionDB / kanban.db handle), and the owner's unwinding TLS
# flush writes an application-data record into that database file.
# ``shutdown()`` from any thread is FD-safe; ``close()`` is not.
# The owning thread performs the real close in the ``finally``
# below, which is where the FD release belongs.
if threading.get_ident() == owner_tid:
close = getattr(self._client, "close", None)
if callable(close):
try:
close()
except Exception:
logger.debug("Codex auxiliary: client close during timeout failed", exc_info=True)
else:
try:
close()
from agent.agent_runtime_helpers import force_close_tcp_sockets

shutdown_count = force_close_tcp_sockets(self._client)
logger.info(
"Codex auxiliary client aborted (timeout, tcp_force_closed=%d, "
"deferred_close=stranger_thread)",
shutdown_count,
)
except Exception:
logger.debug("Codex auxiliary: client close during timeout failed", exc_info=True)
logger.debug("Codex auxiliary: client abort during timeout failed", exc_info=True)
# The cached auxiliary client wraps this same ``self._client``
# (or *is* a ``CodexAuxiliaryClient`` whose ``_real_client`` is
# this instance). After we close the httpx transport above, the
Expand Down Expand Up @@ -1274,6 +1303,19 @@ def _item_get(obj: Any, key: str, default: Any = None) -> Any:
finally:
if timeout_timer is not None:
timeout_timer.cancel()
# A stranger-thread timeout only shut the sockets down; the FDs are
# still open and this — the owning thread, now unwound — is the one
# context that may release them (#29507).
if timed_out.is_set():

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

self._client is cache-shared (agent/auxiliary_client.py:6681-6700), so this request's owner thread may not own another worker's active pooled connection. Closing here can still release that other worker's FD; use a shared-client lease/lifecycle or defer release after all borrowers unwind.

close = getattr(self._client, "close", None)
if callable(close):
try:
close()
except Exception:
logger.debug(
"Codex auxiliary: owner-thread close after timeout failed",
exc_info=True,
)

content = "".join(text_parts).strip() or None

Expand Down
114 changes: 114 additions & 0 deletions tests/agent/test_auxiliary_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import base64
import json
import logging
import threading
import time
from types import SimpleNamespace
from unittest.mock import patch, MagicMock, AsyncMock
Expand Down Expand Up @@ -5279,6 +5280,119 @@ def create(self, **kwargs):
assert time.monotonic() - started < 0.14


class TestCodexAuxiliaryTimeoutFdOwnership:
"""Regression: the timeout Timer must not release the client's FDs.

``close()`` from a thread that does not own the in-flight httpx connection
releases the raw TLS fd while the owner's OpenSSL BIO still caches that
integer. The kernel can hand the same integer to the next ``open()`` in the
process — a SessionDB or kanban.db handle — and the owner's unwinding TLS
flush then writes an application-data record into that database file
(#29507 / #67142 / #70773). ``shutdown()`` is FD-safe from any thread;
``close()`` is not, so it belongs to the owning thread.
"""

@staticmethod
def _adapter_with_recording_client(stream):
"""Build an adapter whose client records (action, thread) events.

The nested ``_client._transport._pool._connections`` shape is what
``force_close_tcp_sockets`` traverses.
"""
events = []

class _Sock:
def shutdown(self, how):
events.append(("shutdown", threading.get_ident()))

def close(self):
events.append(("sock.close", threading.get_ident()))

sock = _Sock()

class _Conn:
def __init__(self):
self._connection = self
self._network_stream = self

def get_extra_info(self, name):
return sock if name == "socket" else None

class _Client:
def __init__(self):
self._transport = SimpleNamespace(
_pool=SimpleNamespace(_connections=[_Conn()])
)

class _LeafClient:
def __init__(self):
self._client = _Client()
self.responses = SimpleNamespace(create=lambda **kw: stream)

def close(self):
events.append(("client.close", threading.get_ident()))

return _CodexCompletionsAdapter(_LeafClient(), "gpt-5.5"), events

def test_stalled_stream_timeout_does_not_release_fds_from_timer_thread(self):
"""The Timer fires on a stalled stream — it may only shutdown()."""

class _StalledStream:
def __iter__(self):
# No event ever arrives, so the owner-thread ``_check_cancelled``
# hook never runs and the Timer is the only thing that can fire.
time.sleep(1.5)
return
yield # pragma: no cover

def close(self):
pass

owner_tid = threading.get_ident()
adapter, events = self._adapter_with_recording_client(_StalledStream())

with pytest.raises(TimeoutError):
adapter.create(
messages=[{"role": "user", "content": "summarize this"}],
timeout=0.05,
)

stranger = [action for action, tid in events if tid != owner_tid]
assert stranger, "the timeout callback never ran on the Timer thread"
# The stranger thread may shut sockets down, but must never release a FD.
assert all(action == "shutdown" for action in stranger), events
assert "client.close" not in stranger
assert "sock.close" not in stranger

def test_owner_thread_releases_fds_after_a_stranger_thread_timeout(self):
"""Deferring the close still has to actually happen — on the owner."""

class _StalledStream:
def __iter__(self):
time.sleep(1.5)
return
yield # pragma: no cover

def close(self):
pass

owner_tid = threading.get_ident()
adapter, events = self._adapter_with_recording_client(_StalledStream())

with pytest.raises(TimeoutError):
adapter.create(
messages=[{"role": "user", "content": "summarize this"}],
timeout=0.05,
)

owner_actions = [action for action, tid in events if tid == owner_tid]
assert "client.close" in owner_actions, events
# Ordering matters: shutdown unblocks the owner, then the owner closes.
assert [a for a, _ in events].index("shutdown") < [
a for a, _ in events
].index("client.close")


class TestCodexAuxiliaryToolMessageConversion:
"""Regression for issue #5709.

Expand Down
Loading