test(agent): cover mount-pool interrupt abort and zero-socket warning

tests/run_agent/test_create_openai_client_reuse.py

@@ -224,3 +224,68 @@ def test_force_close_tcp_sockets_descends_httpcore_1_connection_wrapper():
     # #29507: close() must NOT be called from this helper — the owning
     # httpx worker thread releases the FD, not us.
     assert sock.close_calls == 0
+
+
+def test_force_close_tcp_sockets_finds_sockets_on_httpx_mounts():
+    """HTTP(S)_PROXY / keepalive mounts put live sockets on ``_mounts``.
+
+    #72975: walking only ``_transport`` returned tcp_force_closed=0 while the
+    stream was still mid-recv on a mounted proxy pool, so interrupt logged
+    success and the provider kept the slot for minutes.
+    """
+    from agent.agent_runtime_helpers import force_close_tcp_sockets
+
+    class FakeSocket:
+        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
+
+    sock = FakeSocket()
+    stream = SimpleNamespace(_sock=sock)
+    # TunnelHTTPConnection-shaped: outer proxy wrapper → HTTP11 stream.
+    http11 = SimpleNamespace(_network_stream=stream)
+    tunnel = SimpleNamespace(_connection=http11)
+    mount_pool = SimpleNamespace(_connections=[tunnel])
+    mount_transport = SimpleNamespace(_pool=mount_pool)
+    # Default transport is empty — the failing layout from #72975.
+    empty_pool = SimpleNamespace(_connections=[])
+    default_transport = SimpleNamespace(_pool=empty_pool)
+    http_client = SimpleNamespace(
+        _transport=default_transport,
+        _mounts={"https://": mount_transport},
+    )
+    openai_client = SimpleNamespace(_client=http_client)
+
+    assert force_close_tcp_sockets(openai_client) == 1
+    assert sock.shutdown_calls == 1
+    assert sock.close_calls == 0
+
+
+def test_force_close_tcp_sockets_walks_nested_connection_wrappers():
+    """Forward/tunnel proxies nest HTTPConnection under another ``_connection``."""
+    from agent.agent_runtime_helpers import force_close_tcp_sockets
+
+    class FakeSocket:
+        def __init__(self):
+            self.shutdown_calls = 0
+
+        def shutdown(self, _how):
+            self.shutdown_calls += 1
+
+    sock = FakeSocket()
+    stream = SimpleNamespace(_sock=sock)
+    http11 = SimpleNamespace(_network_stream=stream)
+    http_conn = SimpleNamespace(_connection=http11)  # HTTPConnection wrapper
+    forward = SimpleNamespace(_connection=http_conn)  # ForwardHTTPConnection
+    pool = SimpleNamespace(_connections=[forward])
+    transport = SimpleNamespace(_pool=pool)
+    openai_client = SimpleNamespace(_client=SimpleNamespace(_transport=transport, _mounts={}))
+
+    assert force_close_tcp_sockets(openai_client) == 1
+    assert sock.shutdown_calls == 1

tests/run_agent/test_tls_fd_recycle_corruption.py

@@ -392,6 +392,32 @@ def test_agent_abort_request_openai_client_does_not_call_client_close(caplog):
     ), f"missing abort log line; got: {msgs!r}"
 
 
+def test_agent_abort_request_openai_client_warns_when_no_sockets(caplog):
+    """tcp_force_closed=0 must not look like a successful abort (#72975)."""
+    from run_agent import AIAgent
+
+    # Client with an empty pool — abort finds nothing to shut down.
+    empty_pool = SimpleNamespace(_connections=[])
+    transport = SimpleNamespace(_pool=empty_pool)
+    http_client = SimpleNamespace(_transport=transport, _mounts={})
+    client = SimpleNamespace(_client=http_client, close=MagicMock())
+
+    agent = AIAgent.__new__(AIAgent)
+    agent._client_log_context = lambda: "provider=test"
+
+    with caplog.at_level(logging.WARNING, logger="run_agent"):
+        agent._abort_request_openai_client(client, reason="stream_interrupt_abort")
+
+    client.close.assert_not_called()
+    msgs = [r.getMessage() for r in caplog.records]
+    assert any(
+        "OpenAI client aborted (stream_interrupt_abort" in m
+        and "tcp_force_closed=0" in m
+        and "no sockets found" in m
+        for m in msgs
+    ), f"missing ineffective-abort WARNING; 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