fix: spawn wait_for_result_and_end_input as background task for string prompts (#780)

## Problem

`query()` with a string prompt and hooks/MCP servers deadlocks once the
internal 100-slot anyio message buffer fills up (~50 tool calls). Each
tool call produces ~2 messages, so the buffer fills after about 50 tool
calls.

### Root cause

For string prompts, `client.py:141` awaited
`wait_for_result_and_end_input()` **before** `receive_messages()`
started draining the buffer:

```python
if isinstance(prompt, str):
    await chosen_transport.write(json.dumps(user_message) + "\n")
    await query.wait_for_result_and_end_input()   # blocks until "result" arrives

async for data in query.receive_messages():        # buffer drain starts here
```

Meanwhile `_read_messages()` keeps reading CLI stdout and pushing into
the 100-slot channel. After ~50 tool calls the channel is full and
`_message_send.send()` blocks. Now `_read_messages` can't read anything
else from stdout, including the "result" message that
`wait_for_result_and_end_input` needs — deadlock.

## Fix

Spawn `wait_for_result_and_end_input()` as a background task instead of
awaiting it inline. This matches the existing `AsyncIterable` path which
already uses `spawn_task(stream_input())`, and allows
`receive_messages()` to start draining the buffer immediately.

```python
# Before (deadlocks)
await query.wait_for_result_and_end_input()

# After (concurrent)
query.spawn_task(query.wait_for_result_and_end_input())
```

## Testing

- Added regression test verifying `spawn_task` is called instead of
direct `await`
- Fixed existing test warnings from unawaited mock coroutines
- All 425 tests pass
- Lint, format, and mypy all clean

Fixes #779

Author: qing-ant

src/claude_agent_sdk/_internal/client.py

@@ -138,7 +138,7 @@ async def process_query(
                     "parent_tool_use_id": None,
                 }
                 await chosen_transport.write(json.dumps(user_message) + "\n")
-                await query.wait_for_result_and_end_input()
+                query.spawn_task(query.wait_for_result_and_end_input())
             elif isinstance(prompt, AsyncIterable):
                 # Stream input in background for async iterables
                 query.spawn_task(query.stream_input(prompt))

tests/test_client.py

@@ -155,6 +155,12 @@ async def _test():
                 mock_query.close = AsyncMock()
                 mock_query._tg = None
 
+                def _consume_coro(coro):
+                    coro.close()
+                    return Mock()
+
+                mock_query.spawn_task = Mock(side_effect=_consume_coro)
+
                 async def mock_receive():
                     yield {
                         "type": "result",
@@ -192,3 +198,60 @@ def test_query_uses_default_initialize_timeout(self):
                 env_patch={},
                 expected_timeout=60.0,
             )
+
+    def test_string_prompt_spawns_wait_for_result_as_task(self):
+        """Test that string prompts spawn wait_for_result_and_end_input as a background
+        task instead of awaiting it inline, preventing deadlock when the message
+        buffer fills up (e.g. >50 tool calls with hooks)."""
+
+        async def _test():
+            with (
+                patch(
+                    "claude_agent_sdk._internal.client.SubprocessCLITransport"
+                ) as mock_transport_class,
+                patch("claude_agent_sdk._internal.client.Query") as mock_query_class,
+            ):
+                mock_transport = AsyncMock()
+                mock_transport_class.return_value = mock_transport
+                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)
+
+                mock_query = AsyncMock()
+                mock_query_class.return_value = mock_query
+                mock_query.start = AsyncMock()
+                mock_query.initialize = AsyncMock()
+                mock_query.close = AsyncMock()
+                mock_query._tg = None
+
+                def _consume_coro(coro):
+                    coro.close()
+                    return Mock()
+
+                mock_query.spawn_task = Mock(side_effect=_consume_coro)
+
+                async def mock_receive():
+                    yield {
+                        "type": "result",
+                        "subtype": "success",
+                        "duration_ms": 100,
+                        "duration_api_ms": 80,
+                        "is_error": False,
+                        "num_turns": 1,
+                        "session_id": "test",
+                    }
+
+                mock_query.receive_messages = mock_receive
+
+                async for _ in query(prompt="test", options=ClaudeAgentOptions()):
+                    pass
+
+                mock_query.spawn_task.assert_called_once()
+                assert not mock_query.wait_for_result_and_end_input.await_args_list, (
+                    "wait_for_result_and_end_input should be spawned as a task, "
+                    "not awaited directly"
+                )
+
+        anyio.run(_test)