review: track most-recent result; fail-fast control requests on suppressed exit

- _got_error_result now tracks the most recent result (bool(is_error))
  rather than latching True forever, so [error result -> success result
  -> ProcessError] still propagates in long-lived ClaudeSDKClient sessions.
- Hoist the pending_control_responses fail-fast loop above the if/else so
  in-flight control requests (interrupt, set_model, ...) fail immediately
  even when the ProcessError is suppressed for the message stream.
- Add tests for both.

Author: qing-ant

src/claude_agent_sdk/_internal/query.py

@@ -296,8 +296,7 @@ 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
+                    self._got_error_result = bool(message.get("is_error"))
 
                 # Regular SDK messages go to the stream
                 await self._message_send.send(message)
@@ -307,6 +306,11 @@ async def _read_messages(self) -> None:
             logger.debug("Read task cancelled")
             raise  # Re-raise to properly handle cancellation
         except Exception as 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()
             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
@@ -319,11 +323,6 @@ async def _read_messages(self) -> None:
                 )
             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:

tests/test_query.py

@@ -1057,3 +1057,86 @@ async def _test():
             assert received[0]["subtype"] == "success"
 
         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."""
+
+        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,
+                    },
+                    {
+                        "type": "result",
+                        "subtype": "success",
+                        "is_error": False,
+                        "num_turns": 2,
+                        "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) == 2
+
+        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)."""
+
+        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,
+                    }
+                ],
+                exc=ProcessError(
+                    "Command failed with exit code 1", exit_code=1, stderr=""
+                ),
+            )
+            q = Query(transport=transport, is_streaming_mode=True)
+
+            # Register a pending control request before the read loop runs.
+            event = anyio.Event()
+            q.pending_control_responses["req_1"] = event
+
+            await q.start()
+            async for _ in q.receive_messages():
+                pass
+            await q.close()
+
+            assert event.is_set()
+            assert isinstance(q.pending_control_results["req_1"], ProcessError)
+
+        anyio.run(_test)