From a40ecfcbd2bdcdc104a11e604500be7c7d75e0fc Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Thu, 3 Sep 2026 03:03:44 -0400 Subject: [PATCH 1/4] Fix raw asyncio cancellation during transport close Signed-off-by: 1fanwang <1fannnw@gmail.com> --- .../_internal/transport/subprocess_cli.py | 31 +++++++++---------- tests/test_close_cancellation.py | 29 +++++++++++++++-- 2 files changed, 42 insertions(+), 18 deletions(-) diff --git a/src/claude_agent_sdk/_internal/transport/subprocess_cli.py b/src/claude_agent_sdk/_internal/transport/subprocess_cli.py index 58abc438d..49101c250 100644 --- a/src/claude_agent_sdk/_internal/transport/subprocess_cli.py +++ b/src/claude_agent_sdk/_internal/transport/subprocess_cli.py @@ -942,12 +942,9 @@ def emit(line: str) -> None: async def close(self) -> None: """Close the transport and clean up resources. - The whole body runs inside a shielded cancel scope. Cleanup is - routinely reached while the caller's task is being cancelled (e.g. - `async with ClaudeSDKClient()` unwinding on cancel), and an - unshielded close() would abort at the first await — before the - terminate/kill escalation ran — orphaning the CLI child, which then - surfaces as `` once nothing is left to wait() on it. + Cleanup is retried after raw asyncio cancellation and runs inside a + shielded cancel scope for anyio cancellation. Both paths finish the + terminate/kill escalation before propagating cancellation. Every await in *this* scope is bounded (~20s worst case), so an anyio cancellation is delayed but never blocked: the stream `aclose()`s are a @@ -956,17 +953,19 @@ async def close(self) -> None: stderr task is cancelled before it is awaited, and the lock acquire and every process `wait()` carry an explicit deadline. - Caveat: an anyio shield only defers cancellation that *originates from - an anyio cancel scope*. A raw asyncio cancellation (`asyncio.wait_for` / - `asyncio.timeout` firing, a bare `task.cancel()`, loop shutdown) is - still delivered at the next await in here, and the escalation below only - catches `TimeoutError` — so it would be skipped. That is a pre-existing - limitation of the shield on the asyncio backend rather than something - this scope introduces, and it is still strictly better than before: the - `finally` keeps a still-running child in `_ACTIVE_CHILDREN` for the - atexit reaper instead of dropping it. Making the escalation robust to a - foreign `CancelledError` is a follow-up. + A raw asyncio cancellation is re-raised after cleanup completes. """ + cancellation: BaseException | None = None + while True: + try: + await self._close_impl() + break + except anyio.get_cancelled_exc_class() as exc: + cancellation = exc + if cancellation is not None: + raise cancellation + + async def _close_impl(self) -> None: if not self._process: self._ready = False return diff --git a/tests/test_close_cancellation.py b/tests/test_close_cancellation.py index d3355de9f..a18faef98 100644 --- a/tests/test_close_cancellation.py +++ b/tests/test_close_cancellation.py @@ -6,12 +6,14 @@ child is left running -- an orphan that surfaces as `[claude] ` once nothing is left to wait() on it. -Every test here runs under both asyncio and trio (``anyio_backend`` in -conftest.py): the leak reproduces on both. +Cancellation-scope tests run under both asyncio and trio. The raw +``asyncio.wait_for`` regression runs only under asyncio. """ +import asyncio import json import os +import signal import subprocess import sys import textwrap @@ -119,6 +121,29 @@ async def test_close_under_cancellation_still_reaps_child(tmp_path: Path) -> Non assert process not in _ACTIVE_CHILDREN +@posix_only +@pytest.mark.parametrize("anyio_backend", ["asyncio"]) +async def test_asyncio_timeout_still_reaps_child(tmp_path: Path) -> None: + script = _write_fake_cli(tmp_path) + script.write_text(FAKE_CLI + "\nimport time; time.sleep(60)\n") + transport = SubprocessCLITransport( + prompt="hi", + options=ClaudeAgentOptions(cli_path=str(script)), + ) + await transport.connect() + process = transport._process + assert process is not None + + try: + with pytest.raises(TimeoutError): + await asyncio.wait_for(transport.close(), timeout=0.05) + assert process.returncode == -signal.SIGTERM + assert _process_state(process.pid) == "gone" + assert process not in _ACTIVE_CHILDREN + finally: + await transport.close() + + @posix_only async def test_cancelled_client_context_leaves_no_child(tmp_path: Path) -> None: """A cancelled `async with ClaudeSDKClient()` must not leak the CLI child.""" From 552fc878babbd499c1bf738762a73023e90dc45d Mon Sep 17 00:00:00 2001 From: Stefan Wang <1fannnw@gmail.com> Date: Thu, 3 Sep 2026 20:58:29 -0400 Subject: [PATCH 2/4] Bound repeated asyncio cancellation during close Signed-off-by: Stefan Wang <1fannnw@gmail.com> --- .../_internal/transport/subprocess_cli.py | 24 +++++--- tests/test_close_cancellation.py | 61 +++++++++++++++++-- 2 files changed, 73 insertions(+), 12 deletions(-) diff --git a/src/claude_agent_sdk/_internal/transport/subprocess_cli.py b/src/claude_agent_sdk/_internal/transport/subprocess_cli.py index 49101c250..6c999866f 100644 --- a/src/claude_agent_sdk/_internal/transport/subprocess_cli.py +++ b/src/claude_agent_sdk/_internal/transport/subprocess_cli.py @@ -1,5 +1,6 @@ """Subprocess transport implementation using Claude Code CLI.""" +import asyncio import atexit import json import logging @@ -15,6 +16,7 @@ from typing import Any, cast import anyio +import sniffio from anyio.abc import Process from anyio.streams.text import TextReceiveStream, TextSendStream @@ -942,9 +944,10 @@ def emit(line: str) -> None: async def close(self) -> None: """Close the transport and clean up resources. - Cleanup is retried after raw asyncio cancellation and runs inside a - shielded cancel scope for anyio cancellation. Both paths finish the - terminate/kill escalation before propagating cancellation. + On asyncio, cleanup runs in a separate task shielded from raw task + cancellation. On Trio, the anyio shield in `_close_impl()` handles + cancellation. Both paths finish the terminate/kill escalation before + propagating cancellation. Every await in *this* scope is bounded (~20s worst case), so an anyio cancellation is delayed but never blocked: the stream `aclose()`s are a @@ -955,13 +958,18 @@ async def close(self) -> None: A raw asyncio cancellation is re-raised after cleanup completes. """ - cancellation: BaseException | None = None - while True: + if sniffio.current_async_library() != "asyncio": + await self._close_impl() + return + + cancellation: asyncio.CancelledError | None = None + cleanup_task: asyncio.Task[None] = asyncio.create_task(self._close_impl()) + while not cleanup_task.done(): try: - await self._close_impl() - break - except anyio.get_cancelled_exc_class() as exc: + await asyncio.shield(cleanup_task) + except asyncio.CancelledError as exc: cancellation = exc + await cleanup_task if cancellation is not None: raise cancellation diff --git a/tests/test_close_cancellation.py b/tests/test_close_cancellation.py index a18faef98..0c3f68740 100644 --- a/tests/test_close_cancellation.py +++ b/tests/test_close_cancellation.py @@ -18,6 +18,7 @@ import sys import textwrap from collections.abc import Iterator +from contextlib import suppress from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch @@ -112,10 +113,14 @@ async def test_close_under_cancellation_still_reaps_child(tmp_path: Path) -> Non assert process is not None pid = process.pid - with anyio.CancelScope() as scope: - scope.cancel() # every await inside close() would raise - await transport.close() + with patch.object( + transport, "_close_impl", wraps=transport._close_impl + ) as close_impl: + with anyio.CancelScope() as scope: + scope.cancel() # every await inside close() would raise + await transport.close() + close_impl.assert_awaited_once() assert process.returncode is not None assert _process_state(pid) == "gone" assert process not in _ACTIVE_CHILDREN @@ -135,8 +140,14 @@ async def test_asyncio_timeout_still_reaps_child(tmp_path: Path) -> None: assert process is not None try: - with pytest.raises(TimeoutError): + with ( + patch.object( + transport, "_close_impl", wraps=transport._close_impl + ) as close_impl, + pytest.raises(TimeoutError), + ): await asyncio.wait_for(transport.close(), timeout=0.05) + close_impl.assert_awaited_once() assert process.returncode == -signal.SIGTERM assert _process_state(process.pid) == "gone" assert process not in _ACTIVE_CHILDREN @@ -144,6 +155,48 @@ async def test_asyncio_timeout_still_reaps_child(tmp_path: Path) -> None: await transport.close() +@posix_only +@pytest.mark.parametrize("anyio_backend", ["asyncio"]) +async def test_repeated_asyncio_cancellation_still_reaps_child( + tmp_path: Path, +) -> None: + script = _write_fake_cli(tmp_path) + script.write_text(FAKE_CLI + "\nimport time; time.sleep(60)\n") + transport = SubprocessCLITransport( + prompt="hi", + options=ClaudeAgentOptions(cli_path=str(script)), + ) + await transport.connect() + process = transport._process + assert process is not None + + with patch.object( + transport, "_close_impl", wraps=transport._close_impl + ) as close_impl: + close_task: asyncio.Task[None] = asyncio.create_task(transport.close()) + while close_impl.await_count == 0: + await anyio.sleep(0) + deadline = anyio.current_time() + 7 + try: + while not close_task.done() and anyio.current_time() < deadline: + close_task.cancel() + await anyio.sleep(0.02) + assert close_task.done(), ( + f"_close_impl restarted {close_impl.await_count} times" + ) + with pytest.raises(asyncio.CancelledError): + await close_task + finally: + if not close_task.done(): + with suppress(asyncio.CancelledError): + await close_task + + close_impl.assert_awaited_once() + assert process.returncode == -signal.SIGTERM + assert _process_state(process.pid) == "gone" + assert process not in _ACTIVE_CHILDREN + + @posix_only async def test_cancelled_client_context_leaves_no_child(tmp_path: Path) -> None: """A cancelled `async with ClaudeSDKClient()` must not leak the CLI child.""" From c7ea485eaabf412f009e3e62c8efec76e5a1e2a7 Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Thu, 3 Sep 2026 21:39:33 -0400 Subject: [PATCH 3/4] Fix repeated cancellation during transport cleanup Signed-off-by: 1fanwang <1fannnw@gmail.com> --- .../_internal/transport/subprocess_cli.py | 32 +++++++------------ 1 file changed, 11 insertions(+), 21 deletions(-) diff --git a/src/claude_agent_sdk/_internal/transport/subprocess_cli.py b/src/claude_agent_sdk/_internal/transport/subprocess_cli.py index 6c999866f..5e6a508c6 100644 --- a/src/claude_agent_sdk/_internal/transport/subprocess_cli.py +++ b/src/claude_agent_sdk/_internal/transport/subprocess_cli.py @@ -942,34 +942,24 @@ def emit(line: str) -> None: emit(framer.flush()) async def close(self) -> None: - """Close the transport and clean up resources. + """Finish subprocess cleanup before cancellation reaches the caller. - On asyncio, cleanup runs in a separate task shielded from raw task - cancellation. On Trio, the anyio shield in `_close_impl()` handles - cancellation. Both paths finish the terminate/kill escalation before - propagating cancellation. - - Every await in *this* scope is bounded (~20s worst case), so an anyio - cancellation is delayed but never blocked: the stream `aclose()`s are a - non-blocking `close()` plus a checkpoint on both anyio backends (they - never await `wait_closed()`, so undrained stdin cannot wedge them), the - stderr task is cancelled before it is awaited, and the lock acquire and - every process `wait()` carry an explicit deadline. - - A raw asyncio cancellation is re-raised after cleanup completes. + Raw asyncio cancellation bypasses AnyIO shields, so cleanup runs in a + separate task and the cancellation is re-raised after that task ends. """ if sniffio.current_async_library() != "asyncio": await self._close_impl() return - cancellation: asyncio.CancelledError | None = None cleanup_task: asyncio.Task[None] = asyncio.create_task(self._close_impl()) - while not cleanup_task.done(): - try: - await asyncio.shield(cleanup_task) - except asyncio.CancelledError as exc: - cancellation = exc - await cleanup_task + cancellation: asyncio.CancelledError | None = None + with anyio.CancelScope(shield=True): + while not cleanup_task.done(): + try: + await asyncio.shield(cleanup_task) + except asyncio.CancelledError as exc: + cancellation = exc + cleanup_task.result() if cancellation is not None: raise cancellation From 645e095c2e68a0d87b64822c92281f3bbc5a099c Mon Sep 17 00:00:00 2001 From: Stefan Wang <1fannnw@gmail.com> Date: Sat, 5 Sep 2026 01:19:26 -0400 Subject: [PATCH 4/4] Fix cleanup error precedence under cancellation Signed-off-by: Stefan Wang <1fannnw@gmail.com> --- .../_internal/transport/subprocess_cli.py | 7 +- tests/test_close_cancellation.py | 78 +++++++++++++++++++ 2 files changed, 84 insertions(+), 1 deletion(-) diff --git a/src/claude_agent_sdk/_internal/transport/subprocess_cli.py b/src/claude_agent_sdk/_internal/transport/subprocess_cli.py index 5e6a508c6..6197def77 100644 --- a/src/claude_agent_sdk/_internal/transport/subprocess_cli.py +++ b/src/claude_agent_sdk/_internal/transport/subprocess_cli.py @@ -959,9 +959,14 @@ async def close(self) -> None: await asyncio.shield(cleanup_task) except asyncio.CancelledError as exc: cancellation = exc - cleanup_task.result() + except Exception: + break if cancellation is not None: + cleanup_error = cleanup_task.exception() + if cleanup_error is not None: + raise cancellation from cleanup_error raise cancellation + cleanup_task.result() async def _close_impl(self) -> None: if not self._process: diff --git a/tests/test_close_cancellation.py b/tests/test_close_cancellation.py index 0c3f68740..6627537fd 100644 --- a/tests/test_close_cancellation.py +++ b/tests/test_close_cancellation.py @@ -197,6 +197,84 @@ async def test_repeated_asyncio_cancellation_still_reaps_child( assert process not in _ACTIVE_CHILDREN +@pytest.mark.parametrize("anyio_backend", ["asyncio"]) +async def test_cleanup_failure_does_not_swallow_asyncio_cancellation() -> None: + transport = SubprocessCLITransport( + prompt="hi", + options=ClaudeAgentOptions(cli_path="/usr/bin/claude"), + ) + cleanup_started = asyncio.Event() + fail_cleanup = asyncio.Event() + + async def failing_close() -> None: + cleanup_started.set() + await fail_cleanup.wait() + raise RuntimeError("cleanup failed") + + with patch.object( + transport, "_close_impl", side_effect=failing_close + ) as close_impl: + close_task: asyncio.Task[None] = asyncio.create_task(transport.close()) + await cleanup_started.wait() + close_task.cancel() + await anyio.sleep(0) + assert not close_task.done() + fail_cleanup.set() + with pytest.raises(asyncio.CancelledError) as raised: + await close_task + + close_impl.assert_awaited_once() + assert isinstance(raised.value.__cause__, RuntimeError) + + +@pytest.mark.parametrize("anyio_backend", ["asyncio"]) +async def test_cleanup_failure_does_not_break_asyncio_timeout() -> None: + transport = SubprocessCLITransport( + prompt="hi", + options=ClaudeAgentOptions(cli_path="/usr/bin/claude"), + ) + cleanup_started = asyncio.Event() + fail_cleanup = asyncio.Event() + + async def failing_close() -> None: + cleanup_started.set() + await fail_cleanup.wait() + raise RuntimeError("cleanup failed") + + async def release_cleanup() -> None: + await cleanup_started.wait() + await anyio.sleep(0.05) + fail_cleanup.set() + + release_task = asyncio.create_task(release_cleanup()) + try: + with ( + patch.object(transport, "_close_impl", side_effect=failing_close), + pytest.raises(TimeoutError), + ): + await asyncio.wait_for(transport.close(), timeout=0.01) + finally: + await release_task + + +@pytest.mark.parametrize("anyio_backend", ["asyncio"]) +async def test_cleanup_failure_surfaces_without_cancellation() -> None: + transport = SubprocessCLITransport( + prompt="hi", + options=ClaudeAgentOptions(cli_path="/usr/bin/claude"), + ) + + with ( + patch.object( + transport, + "_close_impl", + side_effect=RuntimeError("cleanup failed"), + ), + pytest.raises(RuntimeError, match="cleanup failed"), + ): + await transport.close() + + @posix_only async def test_cancelled_client_context_leaves_no_child(tmp_path: Path) -> None: """A cancelled `async with ClaudeSDKClient()` must not leak the CLI child."""