Fix query() stream hang when CLI subprocess dies mid-run

[anishesg]

Motivation

When the subprocess dies mid-run (e.g., when a Pro/Max subscription hits its rate-limit window and the CLI exits with code 1), the SDK logs "Fatal error in message reader" but the query() async stream never terminates or raises, causing consumers awaiting the next message to hang indefinitely.

The problem

In _read_messages() (line 353 of query.py), when an exception occurs, the error handler attempts to send an error message to the stream using await self._message_send.send(). However, this await can block if the consumer has stopped iterating or the buffer is full. The finally block already handles the end sentinel correctly using send_nowait() with error suppression (line 372), but the error message path did not follow this pattern.

What this does

Changes the error message send in _read_messages() from await self._message_send.send() to self._message_send.send_nowait() wrapped in suppress(anyio.WouldBlock). This ensures subprocess fatal errors reach consumers without blocking, matching the pattern used for stream termination. If the buffer is full, the stream closes in the finally block and consumers see EndOfStream after draining buffered messages.

Adds regression tests that verify subprocess death during message iteration properly propagates exceptions to consumers without hanging, including tests for parked consumers blocked on receive.

Fixes #1110

[Echolonius]

Thanks for jumping on this — I was investigating #1110 in parallel and ran your branch through some experiments; sharing them because two results seem important for review:

1. Both new tests pass on unfixed main (applied only the tests/ hunks: 2 passed in 1.00s). In both tests the consumer is parked in receive() when the error arrives, and the channel has max_buffer_size=100, so the awaited send() they guard against never actually blocks — meaning they don't reproduce the #1110 hang, and the source change isn't what makes them green. I also couldn't reproduce the issue's plain repro (real subprocess dying mid-run, exit 1 or SIGTERM, asyncio/trio/uvloop): the error reaches the consumer in milliseconds. What does deterministically reproduce the eternal hang is the dying CLI leaving a child process holding its inherited stdout pipe — then the reader never wakes at all (no EOF, and on asyncio even process.wait() is gated on pipe disconnection), so the send() in the error handler is never reached in the first place. Details in #1110 (comment), fix attempt in #1112.

2. The send_nowait swap silently loses the error under real backpressure. With a CLI that bursts 120 messages then dies (exit 1) and a consumer that falls behind ~5s at message 20 (buffer genuinely full when the reader errors):

• main: consumer receives all 120 messages, then raises Command failed with exit code 1 — the awaited send blocks until the consumer drains, which is exactly the backpressure behavior you want here.
• this branch: consumer receives all 120 messages, then the stream ends cleanly — send_nowait hit WouldBlock, the error was dropped, and the crash became indistinguishable from success. For the artifact-loss workflows described in #32411-style pipelines that's arguably worse than a hang.

The blocking send() is only reachable with an abandoned consumer (nobody iterating → nobody observes either way, and close() cancels the read task regardless), so I don't think it needs the nowait treatment. If a maintainer does want belt-and-braces delivery, an ordering-independent option would be stashing the fatal error on the Query and having receive_messages() raise it when the stream ends — that survives a full buffer without racing it.

[anishesg]

yeah, good points. you're right that the tests don't actually hit the blocking case — they're testing the happy path where the consumer is actively reading. and the send_nowait swap is definitely a problem if it silently drops errors under backpressure, that's worse than hanging.

i think the stash-on-Query approach makes sense — lets us deliver the error reliably without racing the buffer state. lemme revert the send_nowait change and try that instead.

[anishesg]

yeah, good catch on both fronts. the tests aren't actually hitting the blocking case, and silently dropping errors under backpressure is definitely worse than hanging. gonna revert the send_nowait and go with the stash-on-Query approach instead — that should deliver the error reliably without the race condition.