From e6cab50fc337daf81d58629c7bc425d74b51340f Mon Sep 17 00:00:00 2001
From: Echolonius <sonicrocks150@gmail.com>
Date: Fri, 10 Jul 2026 22:37:24 -0500
Subject: [PATCH] fix: surface prompt stream errors instead of hanging query()
 forever

stream_input() caught every exception from the caller's prompt
AsyncIterable (or a failed stdin write) and logged it at debug level.
Stdin stayed open, the CLI kept waiting for input that would never
come, no result ever arrived, and query() blocked forever with no
surfaced error.

On failure (outside teardown): log at error level, close stdin so the
CLI can wind down, and push an error message into the message stream so
receive_messages() raises instead of blocking indefinitely.
---
 src/claude_agent_sdk/_internal/query.py | 17 +++++++-
 tests/test_query.py                     | 54 +++++++++++++++++++++++++
 2 files changed, 70 insertions(+), 1 deletion(-)

diff --git a/src/claude_agent_sdk/_internal/query.py b/src/claude_agent_sdk/_internal/query.py
index 7aded9384..a255692c6 100644
--- a/src/claude_agent_sdk/_internal/query.py
+++ b/src/claude_agent_sdk/_internal/query.py
@@ -840,7 +840,22 @@ async def stream_input(self, stream: AsyncIterable[dict[str, Any]]) -> None:
 
             await self.wait_for_result_and_end_input()
         except Exception as e:
-            logger.debug(f"Error streaming input: {e}")
+            if self._closed:
+                # Teardown noise: close() interrupted an in-flight write.
+                logger.debug(f"Error streaming input: {e}")
+                return
+            logger.error(f"Error streaming input: {e}")
+            # Close stdin so the CLI can wind down instead of waiting for
+            # input that will never come.
+            with suppress(Exception):
+                await self.transport.end_input()
+            # Surface the failure to consumers; otherwise receive_messages()
+            # blocks forever on turns the CLI will never run.
+            # If the read task already tore the stream down, its error wins.
+            with suppress(anyio.ClosedResourceError):
+                await self._message_send.send(
+                    {"type": "error", "error": f"Error streaming input: {e}"}
+                )
 
     async def receive_messages(self) -> AsyncIterator[dict[str, Any]]:
         """Receive SDK messages (not control messages)."""
diff --git a/tests/test_query.py b/tests/test_query.py
index 3ae65093d..cc0384083 100644
--- a/tests/test_query.py
+++ b/tests/test_query.py
@@ -309,6 +309,60 @@ async def mock_receive():
 
         anyio.run(_test)
 
+    def test_streaming_prompt_input_stream_error_surfaces(self):
+        """An exception raised by the caller's prompt AsyncIterable must
+        surface to the consumer. Regression test: stream_input() swallowed
+        the exception at debug level and returned without closing stdin, so
+        the CLI kept waiting for input, no result ever arrived, and query()
+        hung forever with no error."""
+
+        async def _test():
+            mock_transport = AsyncMock()
+            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)
+
+            async def mock_receive():
+                # Like the real CLI, stdout stays open until the SDK sends
+                # input or closes stdin; nothing arrives on its own. Without
+                # the fix the consumer blocks here forever.
+                await anyio.sleep(30)
+                yield {}  # pragma: no cover - never reached
+
+            mock_transport.read_messages = mock_receive
+
+            async def raising_stream():
+                yield {
+                    "type": "user",
+                    "session_id": "",
+                    "message": {"role": "user", "content": "hi"},
+                    "parent_tool_use_id": None,
+                }
+                raise ValueError("prompt generator exploded")
+
+            with (
+                patch(
+                    "claude_agent_sdk._internal.client.SubprocessCLITransport"
+                ) as mock_cls,
+                patch(
+                    "claude_agent_sdk._internal.query.Query.initialize",
+                    new_callable=AsyncMock,
+                ),
+            ):
+                mock_cls.return_value = mock_transport
+
+                with anyio.fail_after(5):
+                    with pytest.raises(Exception, match="prompt generator exploded"):
+                        async for _ in query(prompt=raising_stream()):
+                            pass
+
+            # The fix also closes stdin so the CLI can wind down cleanly.
+            mock_transport.end_input.assert_called()
+
+        anyio.run(_test)
+
     def test_string_prompt_with_hooks_waits_for_result(self):
         """end_input() should wait for first result when hooks are configured,
         even without SDK MCP servers."""