fix(_task_compat,query): contextvar parity + close message streams

- spawn_detached(trio): pass context=contextvars.copy_context() to
  trio.lowlevel.spawn_system_task so spawned tasks inherit the caller's
  contextvars, matching asyncio's loop.create_task() semantics. Adds
  TestContextVarPropagation parity tests.
- _read_messages finally: close _message_send after the send_nowait so
  receivers exit on EndOfStream even if the sentinel was dropped due to
  WouldBlock (buffer-full natural-EOF path).
- Query.close(): close _message_send/_message_receive (sync, idempotent)
  to drop the 'ResourceWarning: Unclosed <MemoryObject*Stream>' warnings
  surfaced by the new trio-backend tests.

Author: qing-ant

src/claude_agent_sdk/_internal/_task_compat.py

@@ -17,6 +17,7 @@
 
 from __future__ import annotations
 
+import contextvars
 from collections.abc import Callable, Coroutine
 from contextlib import suppress
 from typing import Any
@@ -151,7 +152,10 @@ async def _runner() -> None:
             finally:
                 handle._mark_done(exc)
 
-        trio.lowlevel.spawn_system_task(_runner)
+        # Pass context= so trio system tasks inherit the caller's
+        # contextvars (asyncio's loop.create_task() does this implicitly;
+        # spawn_system_task does not).
+        trio.lowlevel.spawn_system_task(_runner, context=contextvars.copy_context())
         return handle
     # Unsupported backend: close the coroutine so we don't leak a "coroutine
     # was never awaited" RuntimeWarning on top of the RuntimeError.

src/claude_agent_sdk/_internal/query.py

@@ -325,8 +325,12 @@ async def _read_messages(self) -> None:
             # Always signal end of stream. send_nowait: trio's level-triggered
             # cancellation would re-raise Cancelled at an await checkpoint
             # here, dropping the sentinel and leaving receive_messages() hung.
+            # close() is the fallback for the buffer-full case where
+            # send_nowait raises WouldBlock — receivers then exit on
+            # EndOfStream after draining.
             with suppress(anyio.WouldBlock):
                 self._message_send.send_nowait({"type": "end"})
+            self._message_send.close()
 
     async def _handle_control_request(self, request: SDKControlRequest) -> None:
         """Handle incoming control request from CLI."""
@@ -813,6 +817,14 @@ async def close(self) -> None:
             self._read_task.cancel()
             await self._read_task.wait()
         self._read_task = None
+        # The read task's finally closed the send side; close the receive
+        # side here so callers get EndOfStream and anyio doesn't emit
+        # ResourceWarning: Unclosed <MemoryObject*Stream> at GC time.
+        # _message_send.close() is repeated for the case where start() was
+        # never called (so _read_messages' finally never ran). Both are
+        # sync, idempotent, and checkpoint-free.
+        self._message_send.close()
+        self._message_receive.close()
         await self.transport.close()
 
     # Make Query an async iterator

tests/test_task_compat.py

@@ -119,6 +119,41 @@ async def coro():
         anyio.run(_test, backend="trio")
 
 
+class TestContextVarPropagation:
+    """Spawned tasks must see the caller's contextvars on both backends.
+
+    asyncio's ``loop.create_task()`` copies the current context implicitly;
+    trio's ``spawn_system_task`` does not unless ``context=`` is passed.
+    """
+
+    @staticmethod
+    def _run(backend: str) -> str:
+        import contextvars
+
+        cv: contextvars.ContextVar[str] = contextvars.ContextVar(
+            "cv", default="DEFAULT"
+        )
+        seen: list[str] = []
+
+        async def _test():
+            cv.set("PARENT")
+
+            async def coro():
+                seen.append(cv.get())
+
+            handle = spawn_detached(coro())
+            await handle.wait()
+
+        anyio.run(_test, backend=backend)
+        return seen[0]
+
+    def test_contextvar_propagates_asyncio(self):
+        assert self._run("asyncio") == "PARENT"
+
+    def test_contextvar_propagates_trio(self):
+        assert self._run("trio") == "PARENT"
+
+
 class TestCrossTaskCancel:
     def test_cancel_from_different_task_trio(self):
         """Cancelling from a different task than the spawner must not raise.