diff --git a/agent/transports/codex_app_server.py b/agent/transports/codex_app_server.py index c23ff836ed89..b96228bfca8d 100644 --- a/agent/transports/codex_app_server.py +++ b/agent/transports/codex_app_server.py @@ -44,6 +44,10 @@ def __str__(self) -> str: # pragma: no cover - trivial return f"codex app-server error {self.code}: {self.message}" +class CodexAppServerTransportError(RuntimeError): + """Raised when a JSON-RPC message cannot be written to the child.""" + + @dataclass class _Pending: queue: queue.Queue @@ -222,7 +226,12 @@ def request( q: queue.Queue = queue.Queue(maxsize=1) with self._pending_lock: self._pending[rid] = _Pending(queue=q, method=method) - self._send({"id": rid, "method": method, "params": params or {}}) + try: + self._send({"id": rid, "method": method, "params": params or {}}) + except CodexAppServerTransportError: + with self._pending_lock: + self._pending.pop(rid, None) + raise try: msg = q.get(timeout=timeout) except queue.Empty: @@ -300,14 +309,16 @@ def _take_id(self) -> int: def _send(self, obj: dict) -> None: if self._closed: - raise RuntimeError("codex app-server client is closed") + raise CodexAppServerTransportError("codex app-server client is closed") if self._proc.stdin is None: - raise RuntimeError("codex app-server stdin not available") + raise CodexAppServerTransportError( + "codex app-server stdin not available" + ) try: self._proc.stdin.write((json.dumps(obj) + "\n").encode("utf-8")) self._proc.stdin.flush() except (BrokenPipeError, ValueError) as exc: - raise RuntimeError( + raise CodexAppServerTransportError( f"codex app-server stdin closed unexpectedly: {exc}" ) from exc diff --git a/agent/transports/codex_app_server_session.py b/agent/transports/codex_app_server_session.py index 384a7de8ab81..93e84118be90 100644 --- a/agent/transports/codex_app_server_session.py +++ b/agent/transports/codex_app_server_session.py @@ -36,6 +36,7 @@ from agent.transports.codex_app_server import ( CodexAppServerClient, CodexAppServerError, + CodexAppServerTransportError, ) from agent.transports.codex_event_projector import CodexEventProjector @@ -420,7 +421,11 @@ def request_steer(self, text: str) -> bool: }, timeout=10, ) - except (CodexAppServerError, TimeoutError): + except ( + CodexAppServerError, + CodexAppServerTransportError, + TimeoutError, + ): logger.debug("turn/steer rejected for active Codex turn", exc_info=True) return False accepted_turn_id = response.get("turnId") if isinstance(response, dict) else None @@ -493,7 +498,11 @@ def run_turn( result = TurnResult() try: self.ensure_started() - except (CodexAppServerError, TimeoutError) as exc: + except ( + CodexAppServerError, + CodexAppServerTransportError, + TimeoutError, + ) as exc: result.error = self._format_error_with_stderr( "codex app-server startup failed", exc ) @@ -555,6 +564,13 @@ def run_turn( result.should_retire = True self._interrupt_event.clear() return result + except CodexAppServerTransportError as exc: + result.error = self._format_error_with_stderr( + "turn/start transport failed", exc + ) + result.should_retire = True + self._interrupt_event.clear() + return result result.turn_id = (ts.get("turn") or {}).get("id") with self._active_turn_lock: @@ -663,7 +679,14 @@ def run_turn( result.error or "codex reported turn_aborted" ) - self._handle_server_request(sreq) + try: + self._handle_server_request(sreq) + except CodexAppServerTransportError as exc: + result.error = self._format_error_with_stderr( + "server request response transport failed", exc + ) + result.should_retire = True + break # Activity counts as live signal — reset the post-tool # quiet timer so an approval round-trip doesn't trip it. last_tool_completion_at = None @@ -804,7 +827,11 @@ def compact_thread( result = TurnResult() try: self.ensure_started() - except (CodexAppServerError, TimeoutError) as exc: + except ( + CodexAppServerError, + CodexAppServerTransportError, + TimeoutError, + ) as exc: result.error = self._format_error_with_stderr( "codex app-server startup failed", exc ) @@ -841,6 +868,12 @@ def compact_thread( ) result.should_retire = True return result + except CodexAppServerTransportError as exc: + result.error = self._format_error_with_stderr( + "thread/compact/start transport failed", exc + ) + result.should_retire = True + return result deadline = time.monotonic() + turn_timeout turn_complete = False @@ -866,7 +899,14 @@ def compact_thread( sreq = self._client.take_server_request(timeout=0) if sreq is not None: - self._handle_server_request(sreq) + try: + self._handle_server_request(sreq) + except CodexAppServerTransportError as exc: + result.error = self._format_error_with_stderr( + "server request response transport failed", exc + ) + result.should_retire = True + break continue note = self._client.take_notification( @@ -994,6 +1034,8 @@ def _issue_interrupt(self, turn_id: Optional[str]) -> None: logger.debug("turn/interrupt non-fatal: %s", exc) except TimeoutError: logger.warning("turn/interrupt timed out") + except CodexAppServerTransportError: + logger.debug("turn/interrupt transport unavailable", exc_info=True) def _handle_server_request(self, req: dict) -> None: """Translate a codex server request (approval) into Hermes' approval diff --git a/tests/agent/transports/test_codex_app_server_runtime.py b/tests/agent/transports/test_codex_app_server_runtime.py index aa3199df94be..41e3283467c7 100644 --- a/tests/agent/transports/test_codex_app_server_runtime.py +++ b/tests/agent/transports/test_codex_app_server_runtime.py @@ -7,6 +7,8 @@ from __future__ import annotations +import threading + import pytest from hermes_cli.runtime_provider import ( @@ -110,6 +112,28 @@ def test_codex_error_class_is_runtimeerror(self) -> None: assert "boom" in str(err) assert "-32600" in str(err) + def test_request_write_failure_is_typed_and_clears_pending(self) -> None: + from agent.transports.codex_app_server import ( + CodexAppServerClient, + CodexAppServerTransportError, + ) + + client = object.__new__(CodexAppServerClient) + client._next_id = 1 + client._pending = {} + client._pending_lock = threading.Lock() + + def fail_send(_payload: dict) -> None: + raise CodexAppServerTransportError("stdin closed") + + client._send = fail_send + + with pytest.raises(CodexAppServerTransportError, match="stdin closed") as exc: + client.request("turn/start", {}) + + assert isinstance(exc.value, RuntimeError) + assert client._pending == {} + class TestSpawnEnvIsolation: """The codex spawn must NOT rewrite HOME — codex's shell tool spawns diff --git a/tests/agent/transports/test_codex_app_server_session.py b/tests/agent/transports/test_codex_app_server_session.py index 5a82567ba656..a1a4b2f8a82f 100644 --- a/tests/agent/transports/test_codex_app_server_session.py +++ b/tests/agent/transports/test_codex_app_server_session.py @@ -366,6 +366,67 @@ def stall(method, params): assert "sk-stalled-secret-abc123" not in r.error assert r.should_retire is True + def test_turn_start_transport_failure_returns_retiring_result(self): + from agent.transports.codex_app_server import CodexAppServerTransportError + + client = FakeClient() + client.set_stderr_tail(["provider token sk-transport-secret"]) + + def fail_write(method, params): + if method == "turn/start": + raise CodexAppServerTransportError("stdin closed unexpectedly") + return { + "thread": {"id": "thread-fake-001"}, + "activePermissionProfile": {"id": "x"}, + } + + client._request_handler = fail_write + + result = make_session(client).run_turn("hi", turn_timeout=2.0) + + assert result.thread_id == "thread-fake-001" + assert result.error and "turn/start transport failed" in result.error + assert "stdin closed unexpectedly" in result.error + assert "sk-transport-secret" not in result.error + assert result.should_retire is True + + def test_turn_start_unrelated_runtime_error_still_escapes(self): + client = FakeClient() + + def programming_error(method, params): + if method == "turn/start": + raise RuntimeError("unexpected adapter defect") + return { + "thread": {"id": "thread-fake-001"}, + "activePermissionProfile": {"id": "x"}, + } + + client._request_handler = programming_error + + with pytest.raises(RuntimeError, match="unexpected adapter defect"): + make_session(client).run_turn("hi", turn_timeout=2.0) + + def test_approval_response_transport_failure_returns_retiring_result(self): + from agent.transports.codex_app_server import CodexAppServerTransportError + + client = FakeClient() + client.queue_server_request( + "item/commandExecution/requestApproval", + command="pwd", + cwd="/tmp", + ) + + def fail_write(_request_id, _result): + raise CodexAppServerTransportError("stdin closed") + + client.respond = fail_write + + result = make_session(client).run_turn("hi", turn_timeout=2.0) + + assert result.error and "response transport failed" in result.error + assert "stdin closed" in result.error + assert result.should_retire is True + @@ -385,6 +446,22 @@ def test_steer_appends_input_to_active_turn(self): "expectedTurnId": "turn-live-123", } + def test_steer_transport_failure_is_non_fatal(self): + from agent.transports.codex_app_server import CodexAppServerTransportError + + client = FakeClient() + session = make_session(client) + session.ensure_started() + with session._active_turn_lock: + session._active_turn_id = "turn-live-123" + + def fail_write(method, params): + raise CodexAppServerTransportError("stdin closed") + + client._request_handler = fail_write + + assert session.request_steer("Use Postgres instead") is False + @@ -438,6 +515,48 @@ def test_compact_thread_sends_rpc_and_waits_for_completion(self): assert r.token_usage_last["totalTokens"] == 12 assert r.model_context_window == 200000 + def test_compact_start_transport_failure_returns_retiring_result(self): + from agent.transports.codex_app_server import CodexAppServerTransportError + + client = FakeClient() + + def fail_write(method, params): + if method == "thread/compact/start": + raise CodexAppServerTransportError("stdin closed unexpectedly") + return { + "thread": {"id": "thread-fake-001"}, + "activePermissionProfile": {"id": "x"}, + } + + client._request_handler = fail_write + + result = make_session(client).compact_thread(turn_timeout=2.0) + + assert result.thread_id == "thread-fake-001" + assert result.error and "thread/compact/start transport failed" in result.error + assert result.should_retire is True + + def test_compact_approval_response_transport_failure_retires(self): + from agent.transports.codex_app_server import CodexAppServerTransportError + + client = FakeClient() + client.queue_server_request( + "item/commandExecution/requestApproval", + command="pwd", + cwd="/tmp", + ) + + def fail_write(_request_id, _result): + raise CodexAppServerTransportError("stdin closed") + + client.respond = fail_write + + result = make_session(client).compact_thread(turn_timeout=2.0) + + assert result.error and "response transport failed" in result.error + assert "stdin closed" in result.error + assert result.should_retire is True + def test_compact_thread_ignores_foreign_child_completion(self): client = FakeClient() client.queue_notification( @@ -823,6 +942,20 @@ def test_dead_subprocess_detected_between_iterations(self): # Stderr-derived auth hint takes precedence over generic message assert r.error and "codex login" in r.error + def test_interrupt_transport_failure_is_non_fatal(self): + from agent.transports.codex_app_server import CodexAppServerTransportError + + client = FakeClient() + session = make_session(client) + session.ensure_started() + + def fail_write(method, params): + raise CodexAppServerTransportError("stdin closed") + + client._request_handler = fail_write + + session._issue_interrupt("turn-fake-001") + # ---- thread/start cross-fill ----