fix(sdk): raise asyncio StreamReader buffer in Python AsyncHostTransport (#2760)

* fix(sdk): raise asyncio StreamReader buffer in Python AsyncHostTransport

The Python async transport spawned the host CLI without passing a `limit=`
to `asyncio.create_subprocess_exec`, so its stdout `StreamReader` inherited
asyncio's default 64 KiB buffer. Every host response is written as a single
newline-delimited JSON line, so any `cli.invoke` whose serialized result
exceeds 64 KiB (e.g. `superdoc_get_content` on larger documents) caused
`readline()` to raise `ValueError: Separator is not found, and chunk
exceed the limit` inside `_reader_loop`. The exception was caught by the
generic reader-loop handler and pending requests were rejected with the
misleading `HOST_DISCONNECTED` error — even though the host process was
still alive and healthy.

Pass `limit=` to `create_subprocess_exec` and expose it as a new
`stdout_buffer_limit_bytes` constructor option on `AsyncHostTransport`,
threaded through `SuperDocAsyncRuntime` and `AsyncSuperDocClient`. The
default of 64 MiB safely covers the host's own 32 MiB
`DEFAULT_MAX_STDIN_BYTES` input cap with room for ~2x JSON expansion.

`SyncHostTransport` is unaffected — it uses raw blocking `subprocess.Popen`
which has no asyncio buffer limit.

Adds a `TestAsyncLargeResponse` regression suite that:
  1. Round-trips a 200 KB response through the default-configured transport.
  2. Pins that an explicitly tightened `stdout_buffer_limit_bytes` still
     reproduces the original failure mode, guaranteeing the option is
     wired through to `create_subprocess_exec`.

* fix(sdk): tear down host process on async reader-loop failure

AsyncHostTransport._reader_loop caught reader exceptions by rejecting
pending futures and flipping state to DISCONNECTED, but never killed
self._process. Because dispose() early-returns on DISCONNECTED, any
reader-loop failure left an orphaned host subprocess running with no public API to reap it. This is a pre-existing bug, but the previous
commit made it easier to trip by exposing stdout_buffer_limit_bytes:
any caller who sets it below their real response size hits the orphan path.

Route both the buffer-overflow and generic-error branches through a
new _schedule_cleanup helper that fires _cleanup() as a separate task
(it can't be awaited inline — _cleanup cancels and awaits the reader
task itself). _cleanup kills the process, waits on it, rejects pending,
and only then transitions to DISCONNECTED, so a subsequent dispose()
is a safe no-op instead of leaking the host.

Also catch asyncio.LimitOverrunError / ValueError separately and
surface HOST_PROTOCOL_ERROR with a "raise stdout_buffer_limit_bytes"
hint plus the current limit in details. The previous HOST_DISCONNECTED
code pointed users at the wrong problem since the host was still alive.

Extends TestAsyncLargeResponse to assert HOST_PROTOCOL_ERROR, verify
the hint is in the message, confirm the subprocess is actually reaped
(returncode set, _process cleared), and that dispose() after an overflow is a safe no-op.

* refactor(sdk): dedupe stdout_buffer_limit default and add wiring test

Address review follow-ups on the async transport buffer-limit option.

- Hoist DEFAULT_STDOUT_BUFFER_LIMIT_BYTES (64 MiB) to module scope in
  transport.py and reference it from AsyncHostTransport, the async
  runtime, and AsyncSuperDocClient so the default lives in one place
  instead of three copies of 64 * 1024 * 1024.

- Add a short "raise if a single host response can exceed this size"
  comment on the client.py parameter so callers see the guidance at
  the public API boundary, not buried in transport.py.

- Rename test_response_above_default_64kb_buffer to
  test_response_above_asyncio_default_streamreader_limit. 64 KiB is
  asyncio's default, not the SDK's (which is now 64 MiB), so the old
  name read backwards after this PR.

- Add test_client_threads_stdout_buffer_limit_to_transport: builds
  AsyncSuperDocClient with a custom limit and asserts the value
  reaches AsyncHostTransport. Without this, a silent drop of the arg
  in client.py or runtime.py would leave the existing overflow test
  passing while the public API reverts to the asyncio 64 KiB default.

* fix(sdk): mark transport DISPOSING synchronously on reader teardown

Round-2 review follow-ups:

- _schedule_cleanup now flips state to DISPOSING before scheduling the
  cleanup task. Previously, between the reader returning and the async
  _cleanup running, _ensure_connected's CONNECTED fast path would still
  accept invoke() calls; they then blocked on a future the dead reader
  could never resolve until watchdog_timeout_ms (default 30s).

- Narrow the buffer-overflow catch to readline() only and drop
  asyncio.LimitOverrunError from the tuple. readline() re-raises
  LimitOverrunError as ValueError (it is not a ValueError subclass on
  any supported CPython), so the previous broad except could
  reclassify unrelated ValueErrors from dispatch as a buffer-limit
  error with a misleading remediation hint. Comment corrected to match.

- Re-export DEFAULT_STDOUT_BUFFER_LIMIT_BYTES from superdoc/__init__.py
  so consumers tuning the option don't import from the implementation
  module.

- Tighten test_host_crash to assert HOST_DISCONNECTED specifically and
  verify process teardown via the new _schedule_cleanup path.

- Strengthen the dispose-after-overflow assertion to actually verify
  the no-op claim (state stays DISCONNECTED, _process stays None, a
  second dispose is also safe). Replace the timing-sensitive
  process.returncode read with await process.wait().

* fix(sdk): serialize teardown across reader, _kill_and_reset, and dispose

Round-2 follow-up — addresses the residual race that the synchronous
DISPOSING flip didn't cover.

Before: `_kill_and_reset()` (called from `_send_request` on stdin write
failure or watchdog timeout) `await`ed `_cleanup` directly. If a
reader-triggered `_schedule_cleanup` was in flight, both ran
concurrently and raced on `_reject_all_pending`'s read-then-clear of
`self._pending` (futures added between snapshot and clear were leaked)
and on `process.kill()`/`reader_task.cancel()`. `dispose()` similarly
short-circuited on DISPOSING without waiting for the in-flight cleanup
to finish — the caller saw "disposed" before the host was fully torn
down.

Now:

- `_kill_and_reset` and `dispose` both check the cleanup-task slot and
  `await` an in-flight cleanup rather than starting a parallel one.
  Single-flight teardown across all three entry points.
- `_cleanup` clears `self._cleanup_task` in `finally` when it owns the
  slot, so introspection doesn't surface a stale done handle and the
  next teardown gets a fresh slot.
- `dispose()` after a reader-triggered cleanup now blocks until that
  cleanup finishes, restoring the "host fully torn down on return"
  contract.

Tests:

- `test_schedule_cleanup_dedupe_guard_drops_reentrant_call` — second
  `_schedule_cleanup` does not replace the in-flight task slot.
- `test_overflow_during_dispose_does_not_schedule_cleanup` — `_stopping`
  suppression is honored.
- `test_kill_and_reset_awaits_in_flight_cleanup` — `_kill_and_reset`
  observes the existing task instead of running a parallel `_cleanup`.
- `test_dispose_waits_for_in_flight_cleanup` — `dispose()` blocks until
  reader-triggered cleanup completes before returning.

95 transport tests pass; 5 consecutive runs with PYTHONASYNCIODEBUG=1
show no flakes.

* fix(sdk): close residual races in async transport teardown

Two correctness regressions and three test gaps surfaced in the
final-pass review of the cleanup-task lifecycle.

**1. _ensure_connected race (HIGH).** The synchronous DISPOSING flip
in _schedule_cleanup did not gate _ensure_connected, so a concurrent
connect()/invoke() reaching _start_host during the DISPOSING window
would reassign self._process and self._reader_task. The pending
cleanup task then read those slots after its first await and killed
the freshly-spawned process. Fix: drain self._cleanup_task at the top
of _ensure_connected via asyncio.shield (so a cancelled caller doesn't
abort the in-flight cleanup).

**2. Cancellation propagation race (HIGH).** _kill_and_reset and
dispose() awaited the cleanup task without asyncio.shield. When the
caller (e.g. an invoke task at the watchdog branch) was cancelled,
asyncio cancelled the awaited cleanup task too — _cleanup did not
catch CancelledError around process.wait(), so teardown stopped before
clearing _process / setting state. dispose() then saw DISPOSING with
_cleanup_task=None and returned without finishing teardown, leaking
the host process. Fix: wrap the awaited cleanup in asyncio.shield in
both call sites; restructure _cleanup so it captures handles and sets
state synchronously up-front, before any awaits, so observable state
is always consistent.

**3. Move _stopping guard into _schedule_cleanup.** The previous
test_overflow_during_dispose_does_not_schedule_cleanup was tautological
— it set _stopping=True and then re-checked the same condition in the
test body before calling _schedule_cleanup, so the call never ran and
the assertion passed trivially. Move the guard into _schedule_cleanup
itself (it's the correct authoritative location anyway), remove the
now-redundant call-site checks in _reader_loop, and rewrite the test
to call _schedule_cleanup unconditionally with _stopping=True. The
test now actually exercises the production guard.

**4. Multi-pending-invoke overflow test.** Codex round-2 gap that
remained open. Locks down that _reject_all_pending fails ALL pending
futures with HOST_PROTOCOL_ERROR plus the actionable hint, not just
the one whose response overflowed.

**5. Async reconnect-after-buffer-overflow test.** Sync transport
already had test_reconnect_after_failure; async only covered reconnect
after explicit dispose. Validates that reader-triggered cleanup leaves
the transport reusable for a fresh invoke without wedging
_cleanup_task / _connecting / _process.

Plus: replaced asyncio.sleep(0) with asyncio.Event-based
synchronization in lifecycle tests (Codex/Opus medium — sleep(0) is
implementation-defined under uvloop / Python scheduling changes); two
new tests directly cover the round-3 races
(test_ensure_connected_drains_in_flight_cleanup_before_spawn,
test_kill_and_reset_caller_cancellation_does_not_cancel_cleanup).

99 transport tests pass; 5 consecutive runs with PYTHONASYNCIODEBUG=1
show no flakes; new tests pass under -W error::ResourceWarning.

---------

Co-authored-by: Caio Pizzol <97641911+caio-pizzol@users.noreply.github.com>
Co-authored-by: Caio Pizzol <caio@harbourshare.com>

Author: michaelreavant

packages/sdk/langs/python/superdoc/__init__.py

@@ -10,13 +10,15 @@
     get_tool_catalog,
     list_tools,
 )
+from .transport import DEFAULT_STDOUT_BUFFER_LIMIT_BYTES
 
 __all__ = [
     "SuperDocClient",
     "AsyncSuperDocClient",
     "SuperDocDocument",
     "AsyncSuperDocDocument",
     "SuperDocError",
+    "DEFAULT_STDOUT_BUFFER_LIMIT_BYTES",
     "get_skill",
     "install_skill",
     "list_skills",

packages/sdk/langs/python/superdoc/client.py

@@ -22,6 +22,7 @@
     DocOpenResult as GeneratedDocOpenResult,
 )
 from .runtime import SuperDocAsyncRuntime, SuperDocSyncRuntime
+from .transport import DEFAULT_STDOUT_BUFFER_LIMIT_BYTES
 
 UserIdentity = Dict[str, str]
 
@@ -340,6 +341,9 @@ def __init__(
         request_timeout_ms: int | None = None,
         watchdog_timeout_ms: int = 30_000,
         max_queue_depth: int = 100,
+        # Raise if a single host response can exceed this size (e.g. reading
+        # very large documents); otherwise the default is safe.
+        stdout_buffer_limit_bytes: int = DEFAULT_STDOUT_BUFFER_LIMIT_BYTES,
         default_change_mode: Literal['direct', 'tracked'] | None = None,
         user: UserIdentity | None = None,
     ) -> None:
@@ -350,6 +354,7 @@ def __init__(
             request_timeout_ms=request_timeout_ms,
             watchdog_timeout_ms=watchdog_timeout_ms,
             max_queue_depth=max_queue_depth,
+            stdout_buffer_limit_bytes=stdout_buffer_limit_bytes,
             default_change_mode=default_change_mode,
             user=user,
         )

packages/sdk/langs/python/superdoc/runtime.py

@@ -14,7 +14,11 @@
 from .embedded_cli import resolve_embedded_cli_path
 from .generated.contract import OPERATION_INDEX
 from .protocol import normalize_default_change_mode
-from .transport import AsyncHostTransport, SyncHostTransport
+from .transport import (
+    DEFAULT_STDOUT_BUFFER_LIMIT_BYTES,
+    AsyncHostTransport,
+    SyncHostTransport,
+)
 
 
 class SuperDocSyncRuntime:
@@ -79,6 +83,7 @@ def __init__(
         request_timeout_ms: Optional[int] = None,
         watchdog_timeout_ms: int = 30_000,
         max_queue_depth: int = 100,
+        stdout_buffer_limit_bytes: int = DEFAULT_STDOUT_BUFFER_LIMIT_BYTES,
         default_change_mode: Optional[str] = None,
         user: Optional[Dict[str, str]] = None,
     ) -> None:
@@ -93,6 +98,7 @@ def __init__(
             request_timeout_ms=request_timeout_ms,
             watchdog_timeout_ms=watchdog_timeout_ms,
             max_queue_depth=max_queue_depth,
+            stdout_buffer_limit_bytes=stdout_buffer_limit_bytes,
             default_change_mode=self._default_change_mode,
             user=user,
         )

packages/sdk/langs/python/superdoc/transport.py

@@ -43,6 +43,12 @@
 
 logger = logging.getLogger('superdoc.transport')
 
+# Default stdout StreamReader buffer for the async transport. Host responses
+# are single newline-delimited JSON lines, so this caps the largest individual
+# response a caller can receive. Raise it if your workload routinely produces
+# responses above this size (e.g. whole-document reads on very large docs).
+DEFAULT_STDOUT_BUFFER_LIMIT_BYTES = 64 * 1024 * 1024
+
 # Opt-in debug logging via SUPERDOC_DEBUG=1 or SUPERDOC_LOG_LEVEL=debug.
 # Only configures the named logger — never mutates root logging config.
 _log_level = os.environ.get('SUPERDOC_LOG_LEVEL', '').lower()
@@ -399,6 +405,7 @@ def __init__(
         request_timeout_ms: Optional[int] = None,
         watchdog_timeout_ms: int = 30_000,
         max_queue_depth: int = 100,
+        stdout_buffer_limit_bytes: int = DEFAULT_STDOUT_BUFFER_LIMIT_BYTES,
         default_change_mode: Optional[ChangeMode] = None,
         user: Optional[Dict[str, str]] = None,
     ) -> None:
@@ -409,11 +416,13 @@ def __init__(
         self._request_timeout_ms = request_timeout_ms
         self._watchdog_timeout_ms = watchdog_timeout_ms
         self._max_queue_depth = max_queue_depth
+        self._stdout_buffer_limit_bytes = stdout_buffer_limit_bytes
         self._default_change_mode = default_change_mode
         self._user = user
 
         self._process: Optional[asyncio.subprocess.Process] = None
         self._reader_task: Optional[asyncio.Task] = None
+        self._cleanup_task: Optional[asyncio.Task] = None
         self._pending: Dict[int, asyncio.Future] = {}
         self._state = _State.DISCONNECTED
         self._next_request_id = 1
@@ -428,7 +437,22 @@ async def connect(self) -> None:
 
     async def dispose(self) -> None:
         """Gracefully shut down the host process."""
-        if self._state == _State.DISCONNECTED or self._state == _State.DISPOSING:
+        if self._state == _State.DISCONNECTED:
+            return
+        if self._state == _State.DISPOSING:
+            # A reader-triggered cleanup is in flight (or an earlier teardown
+            # left state in DISPOSING briefly). Wait for it so the caller
+            # observes "host fully torn down" by the time dispose() returns.
+            # shield() so a cancelled dispose() doesn't interrupt _cleanup
+            # mid-flight and leak the host process.
+            existing = self._cleanup_task
+            if existing and not existing.done():
+                try:
+                    await asyncio.shield(existing)
+                except asyncio.CancelledError:
+                    raise
+                except Exception:
+                    pass
             return
 
         self._stopping = True
@@ -507,6 +531,20 @@ async def invoke(
 
     async def _ensure_connected(self) -> None:
         """Lazy connect: spawn and handshake if not already connected."""
+        # Drain any in-flight teardown before spawning a new host. Without
+        # this, a concurrent reader-triggered cleanup would still be running
+        # when _start_host reassigns self._process / self._reader_task; the
+        # cleanup task would then cancel the fresh reader and kill the fresh
+        # process. shield() so we don't cancel the cleanup if our caller is.
+        cleanup = self._cleanup_task
+        if cleanup and not cleanup.done():
+            try:
+                await asyncio.shield(cleanup)
+            except asyncio.CancelledError:
+                raise
+            except Exception:
+                pass
+
         if self._state == _State.CONNECTED and self._process and self._process.returncode is None:
             return
 
@@ -531,12 +569,15 @@ async def _start_host(self) -> None:
         args = [*prefix_args, 'host', '--stdio']
 
         try:
+            # ``limit`` raises asyncio's StreamReader buffer above its 64 KiB
+            # default; host responses are single JSON lines and can exceed it.
             self._process = await asyncio.create_subprocess_exec(
                 command, *args,
                 stdin=asyncio.subprocess.PIPE,
                 stdout=asyncio.subprocess.PIPE,
                 stderr=asyncio.subprocess.DEVNULL,
                 env={**os.environ, **self._env},
+                limit=self._stdout_buffer_limit_bytes,
             )
             logger.debug('Host spawned (pid=%s, bin=%s).', self._process.pid, self._cli_bin)
         except Exception as exc:
@@ -582,7 +623,29 @@ async def _reader_loop(self) -> None:
 
         try:
             while True:
-                raw = await process.stdout.readline()
+                try:
+                    raw = await process.stdout.readline()
+                except ValueError as exc:
+                    # asyncio.StreamReader.readline() re-raises LimitOverrunError
+                    # from readuntil() as ValueError when a single line exceeds
+                    # `limit` (see CPython asyncio/streams.py). The host is still
+                    # alive — schedule cleanup so a later dispose() doesn't
+                    # short-circuit on DISCONNECTED state. Scoped to readline()
+                    # only so unrelated ValueErrors from dispatch aren't
+                    # reclassified as a buffer-limit error. _schedule_cleanup
+                    # is a no-op when _stopping is set (graceful dispose path).
+                    logger.debug('Reader loop buffer overflow: %s', exc)
+                    self._schedule_cleanup(SuperDocError(
+                        'Host response exceeded stdout buffer limit. '
+                        'Raise stdout_buffer_limit_bytes to accommodate larger responses.',
+                        code=HOST_PROTOCOL_ERROR,
+                        details={
+                            'message': str(exc),
+                            'stdout_buffer_limit_bytes': self._stdout_buffer_limit_bytes,
+                        },
+                    ))
+                    return
+
                 if not raw:
                     # EOF — process died.
                     break
@@ -614,16 +677,16 @@ async def _reader_loop(self) -> None:
         except Exception as exc:
             logger.debug('Reader loop error: %s', exc)
 
-        # Reader exited (EOF or error) — reject all pending futures.
-        if not self._stopping:
-            exit_code = process.returncode
-            error = SuperDocError(
-                'Host process disconnected.',
-                code=HOST_DISCONNECTED,
-                details={'exit_code': exit_code, 'signal': None},
-            )
-            self._reject_all_pending(error)
-            self._state = _State.DISCONNECTED
+        # Reader exited (EOF or unexpected error) — tear down the process so
+        # no orphaned host is left running, then reject pending futures.
+        # _schedule_cleanup is a no-op when _stopping is set (graceful
+        # dispose path) so we don't race the dispose teardown.
+        exit_code = process.returncode
+        self._schedule_cleanup(SuperDocError(
+            'Host process disconnected.',
+            code=HOST_DISCONNECTED,
+            details={'exit_code': exit_code, 'signal': None},
+        ))
 
     async def _send_request(self, method: str, params: Any, watchdog_ms: int) -> Any:
         """Send a JSON-RPC request and await the matching response future."""
@@ -687,39 +750,120 @@ def _reject_all_pending(self, error: SuperDocError) -> None:
                 future.set_exception(error)
 
     async def _kill_and_reset(self) -> None:
-        """Kill the host process and reset to DISCONNECTED."""
-        await self._cleanup(
-            SuperDocError('Host process disconnected.', code=HOST_DISCONNECTED),
-        )
-
-    async def _cleanup(self, error: Optional[SuperDocError]) -> None:
-        """Cancel reader, kill process, reject pending, reset state."""
-        if self._reader_task and not self._reader_task.done():
-            self._reader_task.cancel()
+        """Kill the host process and reset to DISCONNECTED.
+
+        Coordinates with `_schedule_cleanup` so callers (e.g. `_send_request`
+        on watchdog timeout or stdin write failure) don't run a parallel
+        `_cleanup` that races a reader-triggered cleanup on
+        `_reject_all_pending` and `process.kill`. If a cleanup is already in
+        flight, await it; otherwise own a fresh task in the same slot so a
+        later concurrent caller sees us instead of starting its own.
+
+        shield() the await so caller cancellation (e.g. an `invoke()` task
+        that times out and is then cancelled by the user) does NOT propagate
+        into `_cleanup` — interrupting cleanup mid-flight would leak the
+        subprocess and wedge state in DISPOSING.
+        """
+        existing = self._cleanup_task
+        if existing and not existing.done():
             try:
-                await self._reader_task
-            except (asyncio.CancelledError, Exception):
+                await asyncio.shield(existing)
+            except asyncio.CancelledError:
+                raise
+            except Exception:
                 pass
-        self._reader_task = None
+            return
+        self._state = _State.DISPOSING
+        task = asyncio.create_task(self._cleanup(
+            SuperDocError('Host process disconnected.', code=HOST_DISCONNECTED),
+        ))
+        self._cleanup_task = task
+        try:
+            await asyncio.shield(task)
+        except asyncio.CancelledError:
+            raise
+        except Exception:
+            pass
 
+    def _schedule_cleanup(self, error: SuperDocError) -> None:
+        """Fire-and-forget teardown from inside the reader task.
+
+        Why a separate task: `_cleanup` cancels and awaits `self._reader_task`.
+        Awaiting it from inside the reader itself would deadlock — so we punt
+        to a fresh task, and by the time it runs the reader has already
+        returned (so cancel+await is a no-op).
+
+        Synchronously flips state to DISPOSING so concurrent `invoke()` callers
+        observe the failed transport immediately rather than passing the
+        CONNECTED fast path and blocking on a future the dead reader can never
+        resolve until `watchdog_timeout_ms`.
+
+        Skips when `_stopping` is set: a graceful `dispose()` is already
+        tearing down, and a parallel cleanup task would race on
+        `_reject_all_pending` and `process.kill`.
+
+        Idempotent: if a cleanup is already in flight, subsequent errors are
+        dropped — the first one wins. Callers may observe completion via
+        `self._cleanup_task`.
+        """
+        if self._stopping:
+            return
+        if self._cleanup_task and not self._cleanup_task.done():
+            return
+        self._state = _State.DISPOSING
+        self._cleanup_task = asyncio.create_task(self._cleanup(error))
+
+    async def _cleanup(self, error: Optional[SuperDocError]) -> None:
+        """Cancel reader, kill process, reject pending, reset state.
+
+        Capture handles and flip user-visible state SYNCHRONOUSLY at the top
+        before any awaits. That way, even if cancellation arrives during
+        `process.wait()`, observers see a consistent "torn down" transport
+        (state DISCONNECTED, _process None, pending futures rejected) rather
+        than a half-disposed one. The async work below is best-effort
+        process reaping.
+        """
+        # Snapshot and clear before any await so concurrent callers see a
+        # fully torn-down transport from this point on.
+        reader_task = self._reader_task
         process = self._process
-        if process:
-            try:
-                process.kill()
-            except Exception:
-                pass
-            try:
-                await asyncio.wait_for(process.wait(), timeout=2)
-            except (asyncio.TimeoutError, Exception):
-                pass
+        self._reader_task = None
         self._process = None
+        self._state = _State.DISCONNECTED
 
-        if error:
+        if error is not None:
             self._reject_all_pending(error)
         else:
             # Dispose path — reject remaining with generic disconnect.
             self._reject_all_pending(
                 SuperDocError('Host process was disposed.', code=HOST_DISCONNECTED),
             )
 
-        self._state = _State.DISCONNECTED
+        try:
+            if reader_task and not reader_task.done():
+                reader_task.cancel()
+                try:
+                    await reader_task
+                except (asyncio.CancelledError, Exception):
+                    pass
+
+            if process:
+                try:
+                    process.kill()
+                except Exception:
+                    pass
+                try:
+                    await asyncio.wait_for(process.wait(), timeout=2)
+                except (asyncio.TimeoutError, asyncio.CancelledError, Exception):
+                    pass
+        finally:
+            # Release the task handle if we are the in-flight cleanup task,
+            # so introspection doesn't surface a stale done handle and the
+            # next teardown gets a fresh slot. Skip when called inline (e.g.
+            # from dispose) — that current task is not our cleanup task.
+            try:
+                current = asyncio.current_task()
+            except RuntimeError:
+                current = None
+            if current is not None and self._cleanup_task is current:
+                self._cleanup_task = None