fix(agent): close codex app-server sessions

agent/transports/codex_app_server_session.py

@@ -397,8 +397,10 @@ def run_turn(
             # turn re-spawns cleanly.
             result.should_retire = True
             return result
-        assert self._client is not None and self._thread_id is not None
-        result.thread_id = self._thread_id
+        client = self._client
+        thread_id = self._thread_id
+        assert client is not None and thread_id is not None
+        result.thread_id = thread_id
 
         self._interrupt_event.clear()
         projector = CodexEventProjector()
@@ -408,18 +410,18 @@ def run_turn(
         # Send turn/start with the user input. Text-only for now (codex
         # supports rich content but Hermes' text path is the common case).
         try:
-            ts = self._client.request(
+            ts = client.request(
                 "turn/start",
                 {
-                    "threadId": self._thread_id,
+                    "threadId": thread_id,
                     "input": [{"type": "text", "text": user_input_text}],
                 },
                 timeout=10,
             )
         except CodexAppServerError as exc:
             # Classify auth/refresh failures so the user gets a clear
             # `codex login` pointer instead of a raw RPC error string.
-            stderr_blob = "\n".join(self._client.stderr_tail(40))
+            stderr_blob = "\n".join(client.stderr_tail(40))
             hint = _classify_oauth_failure(exc.message, stderr_blob)
             if hint is not None:
                 result.error = hint
@@ -435,7 +437,7 @@ def run_turn(
             return result
         except TimeoutError as exc:
             # turn/start hanging is a strong signal the subprocess is wedged.
-            stderr_blob = "\n".join(self._client.stderr_tail(40))
+            stderr_blob = "\n".join(client.stderr_tail(40))
             hint = _classify_oauth_failure(stderr_blob)
             result.error = hint or self._format_error_with_stderr(
                 "turn/start timed out", exc
@@ -462,8 +464,8 @@ def run_turn(
             # (e.g. crashed, segfaulted, or its auth refresh thread killed
             # the process), we won't get any more notifications — bail out
             # rather than waiting for the full turn deadline.
-            if not self._client.is_alive():
-                stderr_blob = "\n".join(self._client.stderr_tail(60))
+            if not client.is_alive():
+                stderr_blob = "\n".join(client.stderr_tail(60))
                 hint = _classify_oauth_failure(stderr_blob)
                 if hint is not None:
                     result.error = hint
@@ -495,14 +497,14 @@ def run_turn(
 
             # Drain any server-initiated requests (approvals) before
             # reading notifications, so the codex side isn't blocked.
-            sreq = self._client.take_server_request(timeout=0)
+            sreq = client.take_server_request(timeout=0)
             if sreq is not None:
                 # Drain any pending notifications first so per-turn state
                 # (e.g. _pending_file_changes for fileChange approvals) is
                 # up to date when we make the approval decision. Bounded
                 # to avoid starving the server-request response.
                 for _ in range(8):
-                    pending = self._client.take_notification(timeout=0)
+                    pending = client.take_notification(timeout=0)
                     if pending is None:
                         break
                     _apply_token_usage_notification(result, pending)
@@ -529,7 +531,7 @@ def run_turn(
                 last_tool_completion_at = None
                 continue
 
-            note = self._client.take_notification(
+            note = client.take_notification(
                 timeout=notification_poll_timeout
             )
             if note is None:
@@ -596,7 +598,7 @@ def run_turn(
                         # rewrite the error into a re-auth hint AND mark
                         # the session for retirement.
                         stderr_blob = "\n".join(
-                            self._client.stderr_tail(40)
+                            client.stderr_tail(40)
                         )
                         hint = _classify_oauth_failure(err_msg, stderr_blob)
                         if hint is not None:
@@ -797,6 +799,8 @@ def _issue_interrupt(self, turn_id: Optional[str]) -> None:
         except CodexAppServerError as exc:
             # "no active turn to interrupt" is fine — already done.
             logger.debug("turn/interrupt non-fatal: %s", exc)
+        except RuntimeError as exc:
+            logger.debug("turn/interrupt skipped; transport closed: %s", exc)
         except TimeoutError:
             logger.warning("turn/interrupt timed out")
 

run_agent.py

@@ -3536,6 +3536,14 @@ def release_clients(self) -> None:
         except Exception:
             pass
 
+        try:
+            codex_session = getattr(self, "_codex_session", None)
+            if codex_session is not None:
+                codex_session.close()
+                self._codex_session = None
+        except Exception:
+            pass
+
         # Close the OpenAI/httpx client to release sockets immediately.
         try:
             client = getattr(self, "client", None)
@@ -3592,6 +3600,14 @@ def close(self) -> None:
         except Exception:
             pass
 
+        try:
+            codex_session = getattr(self, "_codex_session", None)
+            if codex_session is not None:
+                codex_session.close()
+                self._codex_session = None
+        except Exception:
+            pass
+
         # 5. Close the OpenAI/httpx client
         try:
             client = getattr(self, "client", None)

tests/agent/transports/test_codex_app_server_session.py

@@ -417,6 +417,31 @@ def test_deadline_exceeded_records_error(self):
         assert r.interrupted is True
         assert r.error and "timed out" in r.error
 
+    def test_deadline_interrupt_broken_pipe_still_retires_turn(self):
+        client = FakeClient()
+        s = make_session(client)
+        s.ensure_started()
+
+        def request(method: str, params: dict):
+            if method == "turn/start":
+                return {"turn": {"id": "turn-fake-001"}}
+            if method == "turn/interrupt":
+                raise RuntimeError(
+                    "codex app-server stdin closed unexpectedly: [Errno 32] Broken pipe"
+                )
+            return {}
+
+        client._request_handler = request
+        r = s.run_turn(
+            "never finishes",
+            turn_timeout=0.05,
+            notification_poll_timeout=0.01,
+        )
+
+        assert r.interrupted is True
+        assert r.should_retire is True
+        assert r.error and "timed out" in r.error
+
     def test_deadline_uses_monotonic_clock(self):
         client = FakeClient()
         s = make_session(client)
@@ -1110,6 +1135,23 @@ 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_close_during_turn_does_not_clear_client_under_loop(self):
+        client = FakeClient()
+        s = make_session(client)
+        s.ensure_started()
+
+        def request(method: str, params: dict):
+            if method == "turn/start":
+                s.close()
+                return {"turn": {"id": "turn-fake-001"}}
+            return {}
+
+        client._request_handler = request
+        r = s.run_turn("x", turn_timeout=2.0, notification_poll_timeout=0.01)
+
+        assert r.should_retire is True
+        assert r.error and "subprocess exited unexpectedly" in r.error
+
 
 # ---- thread/start cross-fill ----
 

tests/run_agent/test_codex_app_server_integration.py

@@ -70,6 +70,26 @@ def test_api_mode_is_codex_app_server(self):
         agent = _make_codex_agent()
         assert agent.api_mode == "codex_app_server"
 
+    def test_agent_close_closes_attached_codex_app_server_session(self):
+        agent = _make_codex_agent()
+        codex_session = MagicMock()
+        agent._codex_session = codex_session
+
+        agent.close()
+
+        codex_session.close.assert_called_once()
+        assert agent._codex_session is None
+
+    def test_agent_release_clients_closes_attached_codex_app_server_session(self):
+        agent = _make_codex_agent()
+        codex_session = MagicMock()
+        agent._codex_session = codex_session
+
+        agent.release_clients()
+
+        codex_session.close.assert_called_once()
+        assert agent._codex_session is None
+
 
 class TestRunConversationCodexPath:
     def test_run_conversation_returns_codex_shape(self, fake_session):