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."""