From e625602a6772ea40cc149a7c619081c7d7258326 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Tue, 15 Sep 2026 11:57:49 +0530 Subject: [PATCH] test(agent): trim stale-kill unwedge tests to the two invariants Keep the real httpx 0.28 wrapper-shape test (proves the shutdown reaches the socket through BoundSyncStream/ResponseStream/PoolByteStream) and the loopback E2E (a parked reader unwinds within its stale budget and the retry lands). The other four were narrower restatements of the same paths. Also treat httpx.ReadError as a transport error in codex_runtime: it is the same abort-induced-read class the streaming retry loop now recovers from. --- agent/codex_runtime.py | 2 +- tests/agent/test_stream_stale_kill_unwedge.py | 81 +------------------ 2 files changed, 2 insertions(+), 81 deletions(-) diff --git a/agent/codex_runtime.py b/agent/codex_runtime.py index 48a6c1a673..d3263b8f99 100644 --- a/agent/codex_runtime.py +++ b/agent/codex_runtime.py @@ -869,7 +869,7 @@ def run_codex_stream(agent, api_kwargs: dict, client: Any = None, on_first_delta import httpx as _httpx from openai import APIConnectionError as _APIConnectionError from agent import relay_llm - transport_errors = (_httpx.RemoteProtocolError, _httpx.ReadTimeout, _httpx.ConnectError, ConnectionError) + transport_errors = (_httpx.RemoteProtocolError, _httpx.ReadTimeout, _httpx.ReadError, _httpx.ConnectError, ConnectionError) active_client = client or agent._ensure_primary_openai_client(reason="codex_stream_direct") max_stream_retries, model = 1, api_kwargs.get("model") # Accumulate streamed text so callers / compat shims can read it. diff --git a/tests/agent/test_stream_stale_kill_unwedge.py b/tests/agent/test_stream_stale_kill_unwedge.py index e60bc9cf1f..a56d3aa409 100644 --- a/tests/agent/test_stream_stale_kill_unwedge.py +++ b/tests/agent/test_stream_stale_kill_unwedge.py @@ -31,7 +31,7 @@ import run_agent from agent import chat_completion_helpers as helpers -# ── unit: the kill shuts down the killed attempt's socket, never closes it ── +# ── unit: the kill reaches the killed attempt's socket, never closes it ── def _agent(): @@ -47,64 +47,6 @@ def _call(agent, model="m"): return call -def _fake_tls_response(sock): - """httpx-shaped response: ``stream._httpcore_stream._connection`` -> socket.""" - def _no_close(): - raise AssertionError("monitor must never close") - conn = SimpleNamespace(_network_stream=SimpleNamespace(_sock=sock), _connection=None) - return SimpleNamespace(close=_no_close, - stream=SimpleNamespace(_httpcore_stream=SimpleNamespace(_connection=conn))) - - -def test_stale_kill_shuts_down_but_never_closes_the_killed_attempts_response(): - import socket as _socket - reader, writer = _socket.socketpair() - try: - call = _call(_agent()) - call._attempt_stream_response = _fake_tls_response(reader) - call._kill_stale_stream(7.0) # must not raise via the AssertionError close - reader.setblocking(False) - try: - data = reader.recv(1) - except (BlockingIOError, OSError): - data = b"" - assert data == b"", "the killed attempt's socket was not shut down" - finally: - reader.close() - writer.close() - - -def test_escalation_ignores_a_response_a_racing_retry_replaced(): - """A retry replaces ``_attempt_stream_response``; touching the *new* one would - disturb a healthy stream, so the escalation is identity-guarded.""" - import socket as _socket - reader, writer = _socket.socketpair() - try: - call = _call(_agent()) - live = _fake_tls_response(reader) - call._attempt_stream_response = live - call._shutdown_stale_attempt_socket(SimpleNamespace(close=lambda: None, stream=None)) - call._shutdown_stale_attempt_socket(None) - assert reader.getsockopt(_socket.SOL_SOCKET, _socket.SO_ERROR) == 0 - # ...and the socket it was handed for the killed attempt really is shut down. - call._shutdown_stale_attempt_socket(live) - reader.setblocking(False) - try: - data = reader.recv(1) - except (BlockingIOError, OSError): - data = b"" - assert data == b"" - finally: - reader.close() - writer.close() - - -def test_stale_kill_survives_a_response_without_socket(): - call = _call(_agent()) - call._attempt_stream_response = SimpleNamespace() # no stream/socket: must not raise - call._kill_stale_stream(7.0) - - def _recv_shutdown_proof(reader, writer): """Behavioral proof the helper reached *this* socket: the helper applies ``settimeout(0)`` then ``shutdown(SHUT_RDWR)`` (which emits FIN), so the @@ -148,27 +90,6 @@ def test_shutdown_reaches_socket_through_real_httpx_wrapper_shape(): reader.close() writer.close() - -def test_shutdown_prefers_response_extensions_network_stream(): - """``response.extensions["network_stream"]`` is the direct socket path - (reviewer's validated probe); it must work even with no stream wrappers.""" - import socket as _socket - reader, writer = _socket.socketpair() - try: - def _no_close(): - raise AssertionError("monitor must never close") - net = SimpleNamespace(get_extra_info=lambda _name: reader) - resp = SimpleNamespace(close=_no_close, stream=None, - extensions={"network_stream": net}) - call = _call(_agent()) - call._attempt_stream_response = resp - call._shutdown_stale_attempt_socket(resp) - _recv_shutdown_proof(reader, writer) - finally: - reader.close() - writer.close() - - # ── end to end: a reader parked on a silent provider must reconnect ──