fix(codex): reap subprocesses on early stream close

src/agents/extensions/experimental/codex/exec.py

@@ -207,9 +207,12 @@ async def _read_stdout_line() -> bytes:
         finally:
             if cancel_task is not None and not cancel_task.done():
                 cancel_task.cancel()
-            await stderr_task
             if process.returncode is None:
                 process.kill()
+            try:
+                await stderr_task
+            finally:
+                await process.wait()
 
     def _build_env(self, args: CodexExecArgs) -> dict[str, str]:
         # Respect env overrides when provided; otherwise copy from os.environ.

tests/extensions/experiemental/codex/test_codex_exec_thread.py

@@ -422,6 +422,59 @@ async def fake_create_subprocess_exec(*_args: Any, **kwargs: Any) -> StreamReade
     assert captured["kwargs"]["limit"] == exec_module._DEFAULT_SUBPROCESS_STREAM_LIMIT_BYTES
 
 
+@pytest.mark.asyncio
+async def test_codex_exec_run_closes_live_process_before_draining_stderr(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    stderr_eof = asyncio.Event()
+
+    class BlockingStderr:
+        async def read(self, _size: int) -> bytes:
+            await stderr_eof.wait()
+            return b""
+
+    class LiveProcess:
+        def __init__(self) -> None:
+            self.stdin = FakeStdin()
+            self.stdout = FakeStdout(["line\n"])
+            self.stderr = BlockingStderr()
+            self.returncode: int | None = None
+            self.killed = False
+            self.wait_called = False
+
+        async def wait(self) -> None:
+            self.wait_called = True
+            await stderr_eof.wait()
+            self.returncode = -9
+
+        def kill(self) -> None:
+            self.killed = True
+            stderr_eof.set()
+
+        def terminate(self) -> None:
+            raise AssertionError("terminate() should not be used when the stream is closed")
+
+    process = LiveProcess()
+
+    async def fake_create_subprocess_exec(*_args: Any, **_kwargs: Any) -> LiveProcess:
+        return process
+
+    monkeypatch.setattr(exec_module.asyncio, "create_subprocess_exec", fake_create_subprocess_exec)
+
+    exec_client = exec_module.CodexExec(executable_path="/bin/codex")
+    stream = exec_client.run(exec_module.CodexExecArgs(input="hello"))
+
+    assert await anext(stream) == "line"
+    close_task = asyncio.create_task(stream.aclose())
+    await asyncio.sleep(0)
+    killed_before_stderr_eof = process.killed
+    stderr_eof.set()
+    await close_task
+
+    assert killed_before_stderr_eof is True
+    assert process.wait_called is True
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     ("enabled", "expected_config"),