fix(agent): return failed turns when Codex app-server exits during RPC writes

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
 

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

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

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 ----