review: replace ProcessError with result error text instead of suppressing

Mirrors the TypeScript SDK (Query.ts readMessages): when the CLI exits
non-zero after emitting result(is_error=True), replace the generic
"Command failed with exit code 1" with the structured error text the CLI
already reported, instead of suppressing the exception entirely.

This preserves the try/except contract that existing consumers may rely
on (an exception is still raised), while fixing #913's "unactionable
message" complaint. Suppression silently changed iteration semantics for
consumers who detect failure via try/except around `query()`; replacing
the message keeps that contract intact and aligns the two SDKs.

Tracking flag `_got_error_result: bool` becomes
`_last_error_result_text: str | None`, holding the text from
`result.errors` (joined) or the subtype as fallback. Reset condition
broadened from `user`-only to "any non-result, non-session_state_changed
message" to match the TS SDK's `lastErrorResultText` reset.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: qing-ant

src/claude_agent_sdk/_internal/query.py

@@ -129,7 +129,11 @@ def __init__(
 
         # Track first result for proper stream closure with SDK MCP servers
         self._first_result_event = anyio.Event()
-        self._got_error_result = False
+        # Set to the result's error text when the most recent message is a
+        # result with is_error=True. Used to replace the generic "exit code 1"
+        # ProcessError with the structured error the CLI already reported.
+        # Mirrors the TypeScript SDK's `lastErrorResultText` (Query.ts).
+        self._last_error_result_text: str | None = None
 
         # SessionStore mirroring (set via set_transcript_mirror_batcher)
         self._transcript_mirror_batcher: TranscriptMirrorBatcher | None = None
@@ -296,11 +300,22 @@ 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()
-                    self._got_error_result = bool(message.get("is_error"))
-                elif msg_type == "user":
-                    # New turn observed — a ProcessError now is a fresh crash,
-                    # not the expected exit from a prior error result.
-                    self._got_error_result = False
+                    if message.get("is_error"):
+                        errors = message.get("errors") or []
+                        self._last_error_result_text = "; ".join(errors) or str(
+                            message.get("subtype", "unknown error")
+                        )
+                    else:
+                        self._last_error_result_text = None
+                elif not (
+                    msg_type == "system"
+                    and message.get("subtype") == "session_state_changed"
+                ):
+                    # Anything other than the post-turn session_state_changed
+                    # marker means the conversation moved on; a ProcessError
+                    # now is a fresh crash, not the expected exit from a prior
+                    # error result. Mirrors the TypeScript SDK's reset logic.
+                    self._last_error_result_text = None
 
                 # Regular SDK messages go to the stream
                 await self._message_send.send(message)
@@ -315,20 +330,26 @@ async def _read_messages(self) -> None:
                 if request_id not in self.pending_control_results:
                     self.pending_control_results[request_id] = e
                     event.set()
-            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.
+            # When the CLI emits a result with is_error=True (e.g.
+            # error_max_turns, error_during_execution) it then exits non-zero
+            # on purpose, for shell-script consumers. The trailing ProcessError
+            # carries no information beyond "exit code 1" — replace it with the
+            # structured error the CLI already reported so the exception is
+            # actionable. Mirrors the TypeScript SDK (Query.ts readMessages).
+            if isinstance(e, ProcessError) and self._last_error_result_text is not None:
+                error_text = (
+                    f"Claude Code returned an error result: "
+                    f"{self._last_error_result_text}"
+                )
                 logger.debug(
-                    "CLI exited with code %s after error result; "
-                    "treating as clean termination",
+                    "Replacing ProcessError (exit code %s) with result error text",
                     e.exit_code,
                 )
             else:
+                error_text = str(e)
                 logger.error(f"Fatal error in message reader: {e}")
-                # Put error in stream so iterators can handle it
-                await self._message_send.send({"type": "error", "error": str(e)})
+            # Put error in stream so iterators can handle it
+            await self._message_send.send({"type": "error", "error": error_text})
         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

@@ -955,9 +955,9 @@ async def _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."""
+    the trailing ProcessError carries no information beyond "exit code 1".
+    Replace it with the structured error text the CLI already reported so
+    the exception is actionable. Mirrors the TypeScript SDK (Query.ts)."""
 
     def _make_transport_then_raise(self, messages, exc):
         mock_transport = AsyncMock()
@@ -975,20 +975,31 @@ async def mock_receive():
         mock_transport.is_ready = Mock(return_value=True)
         return mock_transport
 
-    def test_process_error_after_error_result_is_suppressed(self):
+    def _error_result(self, subtype="error_max_turns", errors=None, **overrides):
+        msg = {
+            "type": "result",
+            "subtype": subtype,
+            "is_error": True,
+            "num_turns": 1,
+            "session_id": "s",
+            "duration_ms": 1,
+            "duration_api_ms": 1,
+            "total_cost_usd": 0.0,
+        }
+        if errors is not None:
+            msg["errors"] = errors
+        msg.update(overrides)
+        return msg
+
+    def test_process_error_after_error_result_uses_result_error_text(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,
-                    }
+                    self._error_result(
+                        subtype="error_max_turns",
+                        errors=["Reached maximum number of turns (60)"],
+                        num_turns=60,
+                    )
                 ],
                 exc=ProcessError(
                     "Command failed with exit code 1", exit_code=1, stderr=""
@@ -998,16 +1009,71 @@ async def _test():
             await q.start()
 
             received = []
-            async for msg in q.receive_messages():
-                received.append(msg)
+            with pytest.raises(
+                Exception,
+                match=r"Claude Code returned an error result: "
+                r"Reached maximum number of turns \(60\)",
+            ):
+                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):
+    def test_process_error_after_error_result_falls_back_to_subtype(self):
+        """When the result has no errors[] (older CLI / minimal payload), the
+        improved message falls back to the subtype so it's still actionable."""
+
+        async def _test():
+            transport = self._make_transport_then_raise(
+                messages=[self._error_result(subtype="error_during_execution")],
+                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=r"Claude Code returned an error result: error_during_execution",
+            ):
+                async for _ in q.receive_messages():
+                    pass
+            await q.close()
+
+        anyio.run(_test)
+
+    def test_process_error_after_error_result_joins_multiple_errors(self):
+        async def _test():
+            transport = self._make_transport_then_raise(
+                messages=[
+                    self._error_result(
+                        subtype="error_during_execution",
+                        errors=["tool timed out", "ENOENT: missing file"],
+                    )
+                ],
+                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=r"tool timed out; ENOENT: missing file",
+            ):
+                async for _ in q.receive_messages():
+                    pass
+            await q.close()
+
+        anyio.run(_test)
+
+    def test_process_error_without_result_keeps_original_message(self):
         async def _test():
             transport = self._make_transport_then_raise(
                 messages=[],
@@ -1025,7 +1091,7 @@ async def _test():
 
         anyio.run(_test)
 
-    def test_process_error_after_success_result_still_raises(self):
+    def test_process_error_after_success_result_keeps_original_message(self):
         async def _test():
             transport = self._make_transport_then_raise(
                 messages=[
@@ -1058,22 +1124,13 @@ async def _test():
 
         anyio.run(_test)
 
-    def test_process_error_after_error_then_success_result_still_raises(self):
-        """The flag tracks the *most recent* result, not a sticky latch."""
+    def test_process_error_after_error_then_success_result_keeps_original(self):
+        """Tracks the *most recent* result, not a sticky latch."""
 
         async def _test():
             transport = self._make_transport_then_raise(
                 messages=[
-                    {
-                        "type": "result",
-                        "subtype": "error_during_execution",
-                        "is_error": True,
-                        "num_turns": 1,
-                        "session_id": "s",
-                        "duration_ms": 1,
-                        "duration_api_ms": 1,
-                        "total_cost_usd": 0.0,
-                    },
+                    self._error_result(subtype="error_during_execution"),
                     {
                         "type": "result",
                         "subtype": "success",
@@ -1102,23 +1159,49 @@ async def _test():
 
         anyio.run(_test)
 
-    def test_process_error_after_error_result_then_new_turn_still_raises(self):
-        """A new user turn invalidates the 'expecting imminent exit' state from
-        a prior turn's error result; a crash mid-new-turn must propagate."""
+    def test_session_state_changed_after_error_result_preserves_replacement(self):
+        """The CLI emits a post-turn `system: session_state_changed(idle)`
+        marker after the result and before exit. It must not reset the
+        tracking flag — the conversation hasn't moved on."""
 
         async def _test():
             transport = self._make_transport_then_raise(
                 messages=[
+                    self._error_result(
+                        subtype="error_max_turns",
+                        errors=["Reached maximum number of turns (10)"],
+                    ),
                     {
-                        "type": "result",
-                        "subtype": "error_during_execution",
-                        "is_error": True,
-                        "num_turns": 1,
+                        "type": "system",
+                        "subtype": "session_state_changed",
+                        "state": "idle",
                         "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()
+
+            with pytest.raises(
+                Exception, match=r"Claude Code returned an error result"
+            ):
+                async for _ in q.receive_messages():
+                    pass
+            await q.close()
+
+        anyio.run(_test)
+
+    def test_new_turn_after_error_result_keeps_original_message(self):
+        """A new user turn invalidates the 'expecting imminent exit' state from
+        a prior turn's error result; a crash mid-new-turn must surface as-is."""
+
+        async def _test():
+            transport = self._make_transport_then_raise(
+                messages=[
+                    self._error_result(subtype="error_during_execution"),
                     {
                         "type": "user",
                         "message": {"role": "user", "content": "next turn"},
@@ -1142,25 +1225,13 @@ async def _test():
 
         anyio.run(_test)
 
-    def test_pending_control_requests_fail_fast_on_suppressed_exit(self):
-        """Even when the ProcessError is suppressed for the message stream,
-        in-flight control requests must still fail fast (process is dead;
-        no control_response will ever arrive)."""
+    def test_pending_control_requests_fail_fast_on_replaced_error(self):
+        """In-flight control requests must still fail fast (process is dead;
+        no control_response will ever arrive) regardless of message replacement."""
 
         async def _test():
             transport = self._make_transport_then_raise(
-                messages=[
-                    {
-                        "type": "result",
-                        "subtype": "error_max_turns",
-                        "is_error": True,
-                        "num_turns": 1,
-                        "session_id": "s",
-                        "duration_ms": 1,
-                        "duration_api_ms": 1,
-                        "total_cost_usd": 0.0,
-                    }
-                ],
+                messages=[self._error_result(subtype="error_max_turns")],
                 exc=ProcessError(
                     "Command failed with exit code 1", exit_code=1, stderr=""
                 ),
@@ -1172,8 +1243,11 @@ async def _test():
             q.pending_control_responses["req_1"] = event
 
             await q.start()
-            async for _ in q.receive_messages():
-                pass
+            with pytest.raises(
+                Exception, match=r"Claude Code returned an error result"
+            ):
+                async for _ in q.receive_messages():
+                    pass
             await q.close()
 
             assert event.is_set()