fix(transport): unblock read_messages when the CLI dies but orphans hold its stdout pipe

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
 

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"})]

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)