fix(query): suppress redundant ProcessError after error result

When the CLI emits a result message with is_error=True (e.g.
subtype=error_max_turns, error_during_execution) it then exits with a
non-zero code. The transport raised ProcessError on that exit code,
which _read_messages converted to a {type: error} stream message and
receive_messages re-raised as a bare Exception — after the consumer
had already received the structured ResultMessage.

Track whether an error result was delivered; if so, treat the
subsequent ProcessError as expected termination and let the stream
end cleanly. ProcessError before any result, or after a success
result, still propagates.

Fixes #913.

Author: qing-ant

src/claude_agent_sdk/_internal/query.py

@@ -15,6 +15,7 @@
     ListToolsRequest,
 )
 
+from .._errors import ProcessError
 from ..types import (
     PermissionMode,
     PermissionResultAllow,
@@ -128,6 +129,7 @@ def __init__(
 
         # Track first result for proper stream closure with SDK MCP servers
         self._first_result_event = anyio.Event()
+        self._got_error_result = False
 
         # SessionStore mirroring (set via set_transcript_mirror_batcher)
         self._transcript_mirror_batcher: TranscriptMirrorBatcher | None = None
@@ -294,6 +296,8 @@ async def _read_messages(self) -> None:
                     if self._transcript_mirror_batcher is not None:
                         await self._transcript_mirror_batcher.flush()
                     self._first_result_event.set()
+                    if message.get("is_error"):
+                        self._got_error_result = True
 
                 # Regular SDK messages go to the stream
                 await self._message_send.send(message)
@@ -303,14 +307,25 @@ async def _read_messages(self) -> None:
             logger.debug("Read task cancelled")
             raise  # Re-raise to properly handle cancellation
         except Exception as e:
-            logger.error(f"Fatal error in message reader: {e}")
-            # Signal all pending control requests so they fail fast instead of timing out
-            for request_id, event in list(self.pending_control_responses.items()):
-                if request_id not in self.pending_control_results:
-                    self.pending_control_results[request_id] = e
-                    event.set()
-            # Put error in stream so iterators can handle it
-            await self._message_send.send({"type": "error", "error": str(e)})
+            if isinstance(e, ProcessError) and self._got_error_result:
+                # CLI exits non-zero after emitting an error result
+                # (error_max_turns, error_during_execution, ...). The consumer
+                # already received the structured ResultMessage; don't follow
+                # it with a redundant bare Exception.
+                logger.debug(
+                    "CLI exited with code %s after error result; "
+                    "treating as clean termination",
+                    e.exit_code,
+                )
+            else:
+                logger.error(f"Fatal error in message reader: {e}")
+                # Signal all pending control requests so they fail fast instead of timing out
+                for request_id, event in list(self.pending_control_responses.items()):
+                    if request_id not in self.pending_control_results:
+                        self.pending_control_results[request_id] = e
+                        event.set()
+                # Put error in stream so iterators can handle it
+                await self._message_send.send({"type": "error", "error": str(e)})
         finally:
             # Flush any remaining transcript mirror entries before closing so
             # an early stdout EOF or transport error doesn't drop entries

tests/test_query.py

@@ -22,6 +22,7 @@
     query,
     tool,
 )
+from claude_agent_sdk._errors import ProcessError
 from claude_agent_sdk._internal.query import Query
 from claude_agent_sdk.types import HookMatcher
 
@@ -949,3 +950,110 @@ async def _test():
             assert "fast_1" not in q._inflight_requests
 
         asyncio.run(_test())
+
+
+class TestProcessExitAfterErrorResult:
+    """Regression tests for #913: when the CLI emits a result message with
+    is_error=True (e.g. subtype=error_max_turns) and then exits non-zero,
+    the SDK should treat that as clean termination — the consumer already
+    received the structured ResultMessage and shouldn't see a redundant
+    bare Exception."""
+
+    def _make_transport_then_raise(self, messages, exc):
+        mock_transport = AsyncMock()
+
+        async def mock_receive():
+            for msg in messages:
+                yield msg
+            raise exc
+
+        mock_transport.read_messages = mock_receive
+        mock_transport.connect = AsyncMock()
+        mock_transport.close = AsyncMock()
+        mock_transport.end_input = AsyncMock()
+        mock_transport.write = AsyncMock()
+        mock_transport.is_ready = Mock(return_value=True)
+        return mock_transport
+
+    def test_process_error_after_error_result_is_suppressed(self):
+        async def _test():
+            transport = self._make_transport_then_raise(
+                messages=[
+                    {
+                        "type": "result",
+                        "subtype": "error_max_turns",
+                        "is_error": True,
+                        "num_turns": 60,
+                        "session_id": "s",
+                        "duration_ms": 1,
+                        "duration_api_ms": 1,
+                        "total_cost_usd": 0.0,
+                    }
+                ],
+                exc=ProcessError(
+                    "Command failed with exit code 1", exit_code=1, stderr=""
+                ),
+            )
+            q = Query(transport=transport, is_streaming_mode=True)
+            await q.start()
+
+            received = []
+            async for msg in q.receive_messages():
+                received.append(msg)
+            await q.close()
+
+            assert len(received) == 1
+            assert received[0]["subtype"] == "error_max_turns"
+
+        anyio.run(_test)
+
+    def test_process_error_without_result_still_raises(self):
+        async def _test():
+            transport = self._make_transport_then_raise(
+                messages=[],
+                exc=ProcessError(
+                    "Command failed with exit code 1", exit_code=1, stderr=""
+                ),
+            )
+            q = Query(transport=transport, is_streaming_mode=True)
+            await q.start()
+
+            with pytest.raises(Exception, match="Command failed"):
+                async for _ in q.receive_messages():
+                    pass
+            await q.close()
+
+        anyio.run(_test)
+
+    def test_process_error_after_success_result_still_raises(self):
+        async def _test():
+            transport = self._make_transport_then_raise(
+                messages=[
+                    {
+                        "type": "result",
+                        "subtype": "success",
+                        "is_error": False,
+                        "num_turns": 1,
+                        "session_id": "s",
+                        "duration_ms": 1,
+                        "duration_api_ms": 1,
+                        "total_cost_usd": 0.0,
+                    }
+                ],
+                exc=ProcessError(
+                    "Command failed with exit code 1", exit_code=1, stderr=""
+                ),
+            )
+            q = Query(transport=transport, is_streaming_mode=True)
+            await q.start()
+
+            received = []
+            with pytest.raises(Exception, match="Command failed"):
+                async for msg in q.receive_messages():
+                    received.append(msg)
+            await q.close()
+
+            assert len(received) == 1
+            assert received[0]["subtype"] == "success"
+
+        anyio.run(_test)