From 3d076685533b46b228b65836d5d97918f81b5b4e Mon Sep 17 00:00:00 2001 From: Echolonius Date: Sun, 12 Jul 2026 00:03:15 -0500 Subject: [PATCH] fix(transport): unblock read_messages when the CLI dies but orphans hold its stdout pipe The stdout read loop treated pipe EOF as the only process-death signal. Helper processes spawned by the CLI inherit its stdout pipe, so when the CLI exits without reaping them (crash or deliberate nonzero exit) EOF never arrives and read_messages blocks forever: consumers hang with no ProcessError and no end-of-stream (#1110). On the asyncio backend process.wait() is gated on the same pipes, so the trailing exit-code check would hang identically. Add a process-exit watcher that polls returncode (visible regardless of pipe holders) and closes the stdout stream once the reader has been parked on a silent pipe for a full grace window, routing the read loop through its existing ClosedResourceError path into the exit-code check. Chunk-count and parked-on-read guards ensure a reader mid-drain or suspended at yield by consumer backpressure is never cut off, so output written before death is always delivered. Fixes #1110 --- .../_internal/transport/subprocess_cli.py | 96 +++++++++++- tests/test_subprocess_buffering.py | 4 + tests/test_transport.py | 148 ++++++++++++++++++ 3 files changed, 245 insertions(+), 3 deletions(-) diff --git a/src/claude_agent_sdk/_internal/transport/subprocess_cli.py b/src/claude_agent_sdk/_internal/transport/subprocess_cli.py index e9a522d13..0f1549cdb 100644 --- a/src/claude_agent_sdk/_internal/transport/subprocess_cli.py +++ b/src/claude_agent_sdk/_internal/transport/subprocess_cli.py @@ -30,6 +30,11 @@ _DEFAULT_MAX_BUFFER_SIZE = 1024 * 1024 # 1MB buffer limit MINIMUM_CLAUDE_CODE_VERSION = "2.0.0" +# After the CLI process exits, how long the stdout pipe must stay silent with +# the reader parked on it before the reader is forcibly unblocked. See +# _unblock_reader_on_exit. +_EXIT_STREAM_GRACE_SECONDS = 2.0 + # Track live CLI subprocesses so we can terminate them when the parent Python # process exits. This mirrors the TypeScript SDK's parent-exit cleanup and # prevents orphaned `claude` processes from leaking when callers crash or exit @@ -130,6 +135,12 @@ def __init__( self._stdin_stream: TextSendStream | None = None self._stderr_stream: TextReceiveStream | None = None self._stderr_task: TaskHandle | None = None + self._exit_watcher_task: TaskHandle | None = None + # Read-loop introspection for _unblock_reader_on_exit: whether the + # reader is currently parked awaiting a stdout chunk, and how many + # chunks it has consumed so far. + self._awaiting_stdout = False + self._stdout_chunks = 0 self._ready = False self._exit_error: Exception | None = None # Track process exit errors self._max_buffer_size = ( @@ -554,6 +565,10 @@ async def connect(self) -> None: # same pattern as Query._read_task. self._stderr_task = spawn_detached(self._handle_stderr()) + # Watch for process death so the read loop can't hang on a pipe + # kept open by orphaned CLI helper processes. + self._exit_watcher_task = spawn_detached(self._unblock_reader_on_exit()) + # Setup stdin for streaming (always used now) if self._process.stdin: self._stdin_stream = TextSendStream(self._process.stdin) @@ -621,6 +636,53 @@ def emit(line: str) -> None: # synchronous, so it is safe to run during cancellation unwind. emit(framer.flush()) + async def _unblock_reader_on_exit(self) -> None: + """Force the stdout read loop to finish when the process is dead but + stdout never reaches EOF. + + EOF is not a reliable process-death signal: helper processes spawned + by the CLI inherit its stdout pipe, and when the CLI exits without + reaping them (a crash, or the deliberate nonzero exit after a fatal + error) a surviving helper keeps the write end open. The read loop then + blocks forever and neither the trailing ProcessError nor end-of-stream + ever reaches consumers (#1110). + + After the process exits, close the stdout stream once the reader has + been parked on a silent pipe for a full grace window. Both conditions + matter: a reader that is mid-drain or suspended at ``yield`` by + consumer backpressure (``_awaiting_stdout`` False) is left alone, and + a chunk arriving during the window (``_stdout_chunks`` moved) restarts + it — so output the CLI wrote before dying is never cut off. Closing + the stream sends the read loop through its existing + ``ClosedResourceError`` path into the normal exit-code check. + """ + if self._process is None: + return + # Poll returncode rather than awaiting wait(): on the asyncio backend + # wait() itself is gated on the pipes disconnecting (the transport only + # finishes once every pipe protocol reports disconnected), so in + # exactly the scenario this watcher exists for, wait() blocks on the + # same orphan-held pipe as the read loop. returncode is set from the + # child-exit signal and becomes visible regardless of pipe holders. + while self._process.returncode is None: + await anyio.sleep(_EXIT_STREAM_GRACE_SECONDS) + prev_chunks = self._stdout_chunks + while True: + await anyio.sleep(_EXIT_STREAM_GRACE_SECONDS) + stream = self._stdout_stream + if stream is None: + return + if self._stdout_chunks == prev_chunks and self._awaiting_stdout: + break + prev_chunks = self._stdout_chunks + logger.debug( + "CLI process exited but stdout never reached EOF " + "(likely inherited by an orphaned child process); " + "closing the stream to unblock the reader" + ) + with suppress(Exception): + await stream.aclose() + async def close(self) -> None: """Close the transport and clean up resources. @@ -661,6 +723,16 @@ async def close(self) -> None: await self._stderr_task.wait() self._stderr_task = None + # Cancel the process-exit watcher (same pattern as stderr task) + if ( + self._exit_watcher_task is not None + and not self._exit_watcher_task.done() + ): + self._exit_watcher_task.cancel() + with suppress(Exception): + await self._exit_watcher_task.wait() + self._exit_watcher_task = None + # Close stdin stream (hold the write lock to prevent a race with # concurrent writes). Bounded: a writer blocked on a full stdin # pipe must not pin the shielded scope forever. @@ -780,7 +852,19 @@ def guard(length: int) -> None: ) try: - async for chunk in self._stdout_stream: + stream_iter = aiter(self._stdout_stream) + while True: + # Flag the park-on-read state (and count chunks) so + # _unblock_reader_on_exit can tell an idle pipe apart from a + # reader suspended at `yield` by consumer backpressure. + self._awaiting_stdout = True + try: + chunk = await anext(stream_iter) + except StopAsyncIteration: + break + finally: + self._awaiting_stdout = False + self._stdout_chunks += 1 for line in framer.push(chunk): guard(len(line)) data = _parse_stdout_line(line) @@ -808,9 +892,15 @@ def guard(length: int) -> None: if data is not None: yield data - # Check process completion and handle errors + # Check process completion and handle errors. Prefer the already-set + # returncode: when _unblock_reader_on_exit closed the stream, wait() + # would block right here on the asyncio backend (it is gated on the + # same orphan-held pipe the watcher just detected — see that method). try: - returncode = await self._process.wait() + if self._process.returncode is not None: + returncode = self._process.returncode + else: + returncode = await self._process.wait() except Exception: returncode = -1 diff --git a/tests/test_subprocess_buffering.py b/tests/test_subprocess_buffering.py index daf79dc53..0d7e09cfe 100644 --- a/tests/test_subprocess_buffering.py +++ b/tests/test_subprocess_buffering.py @@ -342,6 +342,7 @@ async def _test() -> None: transport = SubprocessCLITransport(prompt="test", options=make_options()) transport._stdout_stream = stream transport._process = MagicMock() + transport._process.returncode = None transport._process.wait = AsyncMock(return_value=0) messages: list[dict[str, Any]] = [] @@ -372,6 +373,7 @@ async def _test() -> None: transport = SubprocessCLITransport(prompt="test", options=make_options()) transport._stdout_stream = stream transport._process = MagicMock() + transport._process.returncode = None transport._process.wait = AsyncMock(return_value=0) messages: list[dict[str, Any]] = [] @@ -391,6 +393,7 @@ def _collect(self, chunks: list[str], **opts: object) -> list[Any]: async def _run() -> None: transport = SubprocessCLITransport(prompt="t", options=make_options(**opts)) transport._process = MagicMock() + transport._process.returncode = None transport._process.wait = AsyncMock(return_value=0) transport._stdout_stream = MockTextReceiveStream(chunks) transport._stderr_stream = MockTextReceiveStream([]) @@ -523,6 +526,7 @@ async def __anext__(self) -> str: transport = SubprocessCLITransport(prompt="t", options=make_options()) transport._process = MagicMock() + transport._process.returncode = None transport._process.wait = AsyncMock(return_value=0) transport._stdout_stream = ClosingStream( [json.dumps({"type": "result", "subtype": "success"})] diff --git a/tests/test_transport.py b/tests/test_transport.py index d55aa18b0..ab449a92a 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -2219,3 +2219,151 @@ async def _test() -> None: await proc.wait() anyio.run(_test) + + +class TestReaderUnblockOnProcessExit: + """Regression tests for #1110: process death must end read_messages even + when stdout never reaches EOF. + + Helper processes spawned by the CLI inherit its stdout pipe; if the CLI + dies without reaping them, EOF never arrives. On the asyncio backend + ``process.wait()`` is gated on the same pipes, so the mocks here enforce + the real contract: ``wait()`` never returns, only ``returncode`` is set. + """ + + class _PipeHeldOpenStream: + """Stdout stream whose EOF never arrives (orphan holds the pipe). + + Yields the given chunks, then parks forever — until aclose(), after + which the parked (or next) read raises ClosedResourceError, exactly + like a real TextReceiveStream. + """ + + def __init__(self, chunks: list[str]) -> None: + self._chunks = list(chunks) + self._closed = anyio.Event() + + def __aiter__(self) -> "TestReaderUnblockOnProcessExit._PipeHeldOpenStream": + return self + + async def __anext__(self) -> str: + if self._closed.is_set(): + raise anyio.ClosedResourceError + if self._chunks: + return self._chunks.pop(0) + await self._closed.wait() + raise anyio.ClosedResourceError + + async def aclose(self) -> None: + self._closed.set() + + class _ExitedProcessWithHeldPipes: + """Process that has exited, but whose wait() never completes. + + Mirrors asyncio's behavior when an inherited pipe outlives the child: + returncode is set from the exit signal while wait() stays blocked on + pipe disconnection. The fix must not rely on wait(). + """ + + def __init__(self, returncode: int) -> None: + self.returncode = returncode + + async def wait(self) -> int: + await anyio.sleep(3600) + return self.returncode + + def _make_transport(self, returncode: int, chunks: list[str]): + transport = SubprocessCLITransport(prompt="test", options=make_options()) + transport._process = self._ExitedProcessWithHeldPipes(returncode) # type: ignore[assignment] + transport._stdout_stream = self._PipeHeldOpenStream(chunks) # type: ignore[assignment] + return transport + + def test_crash_with_orphan_held_pipe_raises_process_error(self, monkeypatch): + """CLI died with exit 1, orphan holds stdout: must raise, not hang.""" + from claude_agent_sdk._errors import ProcessError + from claude_agent_sdk._internal._task_compat import spawn_detached + from claude_agent_sdk._internal.transport import subprocess_cli + + monkeypatch.setattr(subprocess_cli, "_EXIT_STREAM_GRACE_SECONDS", 0.05) + + async def _test(): + transport = self._make_transport( + returncode=1, + chunks=['{"type": "assistant", "message": {"content": []}}\n'], + ) + watcher = spawn_detached(transport._unblock_reader_on_exit()) + received = [] + try: + with anyio.fail_after(5): + with pytest.raises(ProcessError, match="exit code 1"): + async for message in transport.read_messages(): + received.append(message) + finally: + if not watcher.done(): + watcher.cancel() + # Output produced before death must still be delivered. + assert len(received) == 1 + assert received[0]["type"] == "assistant" + + anyio.run(_test) + + def test_clean_exit_with_orphan_held_pipe_ends_stream(self, monkeypatch): + """CLI exited 0 but orphan holds stdout: stream must end cleanly.""" + from claude_agent_sdk._internal._task_compat import spawn_detached + from claude_agent_sdk._internal.transport import subprocess_cli + + monkeypatch.setattr(subprocess_cli, "_EXIT_STREAM_GRACE_SECONDS", 0.05) + + async def _test(): + transport = self._make_transport( + returncode=0, + chunks=['{"type": "result", "subtype": "success"}\n'], + ) + watcher = spawn_detached(transport._unblock_reader_on_exit()) + received = [] + try: + with anyio.fail_after(5): + async for message in transport.read_messages(): + received.append(message) + finally: + if not watcher.done(): + watcher.cancel() + assert len(received) == 1 + assert received[0]["type"] == "result" + + anyio.run(_test) + + def test_backpressure_is_not_mistaken_for_idle_pipe(self, monkeypatch): + """A reader parked at yield (slow consumer) must not be force-closed. + + The consumer takes longer than the grace window to process each + message while more chunks are still arriving; every chunk must be + delivered before the trailing ProcessError. + """ + from claude_agent_sdk._errors import ProcessError + from claude_agent_sdk._internal._task_compat import spawn_detached + from claude_agent_sdk._internal.transport import subprocess_cli + + monkeypatch.setattr(subprocess_cli, "_EXIT_STREAM_GRACE_SECONDS", 0.05) + + async def _test(): + chunks = [ + f'{{"type": "assistant", "message": {{"content": [], "i": {i}}}}}\n' + for i in range(5) + ] + transport = self._make_transport(returncode=1, chunks=chunks) + watcher = spawn_detached(transport._unblock_reader_on_exit()) + received = [] + try: + with anyio.fail_after(10): + with pytest.raises(ProcessError, match="exit code 1"): + async for message in transport.read_messages(): + received.append(message) + # Slower than the grace window. + await anyio.sleep(0.15) + finally: + if not watcher.done(): + watcher.cancel() + assert len(received) == 5 + + anyio.run(_test)