Fix stdio client shutdown: cancellation deadlock, orphaned children, spurious kills

Four user-facing bugs in the stdio client's shutdown path, all rooted in
the same design problem — the shutdown sequence depended on pipe state
and was not protected from cancellation:

- Cancelling stdio_client (a client timeout, app shutdown) skipped the
  entire shutdown sequence: the first await in the finally block
  re-raised the pending cancellation, so stdin was never closed
  gracefully and the process tree was never terminated. Control then
  fell through to anyio's Process.aclose, whose shielded wait only
  returns once every pipe has closed - and a pipe inherited by a
  grandchild (npx- and uv-style servers always have one) never closes,
  so the client deadlocked forever. The shutdown now runs inside a
  shielded cancel scope with every wait time-bounded, and never relies
  on a pipe-gated wait.

- The grace-period check used process.wait(), which on the asyncio
  backend resolves only when the process has exited AND its pipes have
  closed. A well-behaved server that exited instantly on stdin closure
  but left a background child holding stdout was misclassified as hung,
  burned the full grace period, and got its tree terminated with a
  spurious warning. The wait now polls returncode, which reflects
  process death alone.

- Process-tree termination derived the group ID with os.getpgid(pid),
  which fails once the leader has been reaped even while its
  descendants are alive — and the fallback then "terminated" the dead
  leader, leaking every descendant. Since the process is spawned with
  start_new_session=True, the pgid is by definition the leader's pid;
  use it directly, and treat ProcessLookupError from killpg as "group
  already gone" rather than a reason to fall back.

- Writing to a dead server's stdin surfaced a raw backend exception
  (ConnectionResetError inside an exception group) to the caller
  instead of a clean closed-stream signal. stdin_writer now treats
  BrokenResourceError and ConnectionError like ClosedResourceError.

Windows fixes, by analysis (CI must validate): FallbackProcess.wait()
ran Popen.wait() in a non-cancellable worker thread, so the timeout
around the grace period could never fire and shutdown could hang
indefinitely — it now polls cancellably, and returncode reflects death
without requiring a wait. terminate_windows_process_tree dropped its
timeout_seconds parameter, which was documented but never used (Job
Object termination is immediate; the docstring now says so).

Cleanup in the same files: the escalation now lives in one function
with two named timeouts instead of three hardcoded 2.0s; the Job Object
is tracked in a WeakKeyDictionary instead of a monkey-patched private
attribute on anyio's Process; the deprecated terminate_windows_process
is removed (migration.md updated); the always-true CREATE_NO_WINDOW
hasattr dance and a retry path that double-spawned on spawn failure are
gone; the client's JSON parse error handler catches ValueError instead
of Exception.

Author: maxisbey

docs/migration.md

@@ -105,6 +105,14 @@ The `headers`, `timeout`, `sse_read_timeout`, and `auth` parameters have been re
 
 Note: `sse_client` retains its `headers`, `timeout`, `sse_read_timeout`, and `auth` parameters — only the streamable HTTP transport changed.
 
+### `terminate_windows_process` removed
+
+The deprecated `mcp.os.win32.utilities.terminate_windows_process` function has been
+removed. Process termination is handled internally by the `stdio_client` context
+manager; there is no replacement API. The Windows tree-termination helper
+`terminate_windows_process_tree` no longer accepts a `timeout_seconds` argument —
+the value was never used (Job Object termination is immediate).
+
 ### Removed type aliases and classes
 
 The following deprecated type aliases and classes have been removed from `mcp.types`:

src/mcp/client/stdio.py

@@ -7,7 +7,7 @@
 
 import anyio
 import anyio.lowlevel
-from anyio.abc import Process
+from anyio.abc import AsyncResource, Process
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
 from anyio.streams.text import TextReceiveStream
 from pydantic import BaseModel, Field
@@ -16,6 +16,7 @@
 from mcp.os.posix.utilities import terminate_posix_process_tree
 from mcp.os.win32.utilities import (
     FallbackProcess,
+    close_process_job,
     create_windows_process,
     get_windows_executable_command,
     terminate_windows_process_tree,
@@ -44,9 +45,18 @@
     else ["HOME", "LOGNAME", "PATH", "SHELL", "TERM", "USER"]
 )
 
-# Timeout for process termination before falling back to force kill
+# How long the server gets to exit on its own after its stdin is closed, before its
+# process tree is terminated.
 PROCESS_TERMINATION_TIMEOUT = 2.0
 
+# How long the process tree gets to die after a graceful termination request
+# (SIGTERM on POSIX) before it is force-killed. Windows tree termination is an
+# immediate hard kill, so this only stretches the POSIX path.
+FORCE_KILL_TIMEOUT = 2.0
+
+# How often to poll for process death while waiting out the grace period.
+_EXIT_POLL_INTERVAL = 0.01
+
 
 def get_default_environment() -> dict[str, str]:
     """Returns a default environment object including only environment variables deemed
@@ -128,13 +138,10 @@ async def stdio_client(server: StdioServerParameters, errlog: TextIO = sys.stder
         )
     except OSError:
         # Clean up streams if process creation fails
-        await read_stream.aclose()
-        await write_stream.aclose()
-        await read_stream_writer.aclose()
-        await write_stream_reader.aclose()
+        await _aclose_all(read_stream, write_stream, read_stream_writer, write_stream_reader)
         raise
 
-    async def stdout_reader():
+    async def stdout_reader() -> None:
         assert process.stdout, "Opened process is missing stdout"
 
         try:
@@ -151,17 +158,20 @@ async def stdout_reader():
                     for line in lines:
                         try:
                             message = types.jsonrpc_message_adapter.validate_json(line, by_name=False)
-                        except Exception as exc:  # pragma: no cover
+                        except ValueError as exc:
                             logger.exception("Failed to parse JSONRPC message from server")
                             await read_stream_writer.send(exc)
                             continue
 
                         session_message = SessionMessage(message)
                         await read_stream_writer.send(session_message)
-        except anyio.ClosedResourceError:  # pragma: lax no cover
+        except (anyio.ClosedResourceError, anyio.BrokenResourceError):
+            # ClosedResourceError: our own shutdown closed the writer. BrokenResourceError:
+            # the caller exited with a message still undelivered, so closing the read
+            # stream broke this task's blocked send. Both just mean shutdown is underway.
             await anyio.lowlevel.checkpoint()
 
-    async def stdin_writer():
+    async def stdin_writer() -> None:
         assert process.stdin, "Opened process is missing stdin"
 
         try:
@@ -174,41 +184,96 @@ async def stdin_writer():
                             errors=server.encoding_error_handler,
                         )
                     )
-        except anyio.ClosedResourceError:  # pragma: no cover
+        except (anyio.ClosedResourceError, anyio.BrokenResourceError, ConnectionError):
+            # The server stopped reading — its process died or closed stdin. The exact
+            # exception depends on platform and backend; all of them just mean the pipe
+            # is gone. The read side reports the closure to the caller.
             await anyio.lowlevel.checkpoint()
 
-    async with anyio.create_task_group() as tg, process:
+    async with anyio.create_task_group() as tg:
         tg.start_soon(stdout_reader)
         tg.start_soon(stdin_writer)
         try:
             yield read_stream, write_stream
         finally:
-            # MCP spec: stdio shutdown sequence
-            # 1. Close input stream to server
-            # 2. Wait for server to exit, or send SIGTERM if it doesn't exit in time
-            # 3. Send SIGKILL if still not exited
-            if process.stdin:  # pragma: no branch
-                try:
-                    await process.stdin.aclose()
-                except Exception:  # pragma: no cover
-                    # stdin might already be closed, which is fine
-                    pass
-
-            try:
-                # Give the process time to exit gracefully after stdin closes
-                with anyio.fail_after(PROCESS_TERMINATION_TIMEOUT):
-                    await process.wait()
-            except TimeoutError:
-                # Process didn't exit from stdin closure, use platform-specific termination
-                # which handles SIGTERM -> SIGKILL escalation
-                await _terminate_process_tree(process)
-            except ProcessLookupError:  # pragma: no cover
-                # Process already exited, which is fine
-                pass
-            await read_stream.aclose()
-            await write_stream.aclose()
-            await read_stream_writer.aclose()
-            await write_stream_reader.aclose()
+            # The shutdown sequence must run to completion even when the caller is
+            # being cancelled — a cancellation that skipped it would leak the server
+            # process (and its children) and could block forever on the way out.
+            # Every wait inside the shield is time-bounded. The shield holds against
+            # anyio-level cancellation and one native task.cancel(); a second native
+            # cancel delivered mid-cleanup can still abort it — there is no
+            # backend-neutral way to refuse repeated native cancellation.
+            with anyio.CancelScope(shield=True):
+                await _shutdown_process_tree(process)
+                await _aclose_all(read_stream, write_stream, read_stream_writer, write_stream_reader)
+
+
+async def _aclose_all(*streams: AsyncResource) -> None:
+    """Close every given stream."""
+    for stream in streams:
+        await stream.aclose()
+
+
+async def _shutdown_process_tree(process: Process | FallbackProcess) -> None:
+    """Shut the server process down per the MCP spec stdio sequence.
+
+    1. Close the server's stdin so it can exit on its own.
+    2. Give it ``PROCESS_TERMINATION_TIMEOUT`` seconds to exit.
+    3. Otherwise terminate its whole process tree (SIGTERM then SIGKILL on POSIX,
+       Job Object hard kill on Windows).
+    """
+    if process.stdin:  # pragma: no branch
+        try:
+            await process.stdin.aclose()
+        except (OSError, anyio.BrokenResourceError, anyio.ClosedResourceError):
+            # stdin is already closed or the pipe is already gone, which is fine
+            await anyio.lowlevel.checkpoint()
+
+    # The await is hoisted out of the `if` so no cancel-unwind from the timeout's
+    # internal scope traverses the frame at the branching line: Python 3.11's tracer
+    # loses the branch arc at a throw()-traversed suspension point (python/cpython#106749).
+    exited = await _wait_for_process_exit(process, PROCESS_TERMINATION_TIMEOUT)
+    if not exited:
+        # Process didn't exit from stdin closure; use platform-specific termination,
+        # which kills the entire process tree, not just the spawned process.
+        await _terminate_process_tree(process)
+        # Wait (bounded) for the kill to be observed by the event loop: on asyncio,
+        # ``returncode`` flips only once the child watcher's callback has been
+        # delivered, and that delivery is also what lets the subprocess transport
+        # close instead of leaking a ResourceWarning into whatever runs next.
+        # Kill-death is prompt, so this wait normally takes one poll interval.
+        await _wait_for_process_exit(process, FORCE_KILL_TIMEOUT)
+
+    # On Windows, drop the process's Job Object handle now: the job is configured to
+    # kill its remaining members when the handle closes, so closing it here makes
+    # that reaping deterministic instead of GC-timed. (POSIX deliberately leaves a
+    # well-behaved server's surviving children alive; no-op there.)
+    close_process_job(process)
+
+    # The process is dead, but its stdout pipe can still be held open by something
+    # that inherited it (an orphaned grandchild, say), in which case the reader task
+    # would never see EOF. Closing our wrapper poisons the Python-level reader so the
+    # reader task finishes either way; the OS-level pipe end itself lives until the
+    # subprocess transport is closed.
+    if process.stdout:  # pragma: no branch
+        await process.stdout.aclose()
+
+
+async def _wait_for_process_exit(process: Process | FallbackProcess, timeout: float) -> bool:
+    """Wait for the process itself to die, returning whether it did within ``timeout``.
+
+    Deliberately does not use ``process.wait()``: on the asyncio backend that resolves
+    only once the process has exited *and* every one of its pipes has closed — and
+    pipes are inherited by the server's own children, so a well-behaved server that
+    exits instantly but leaves a background child alive would be misclassified as
+    hung and get its whole tree terminated. ``returncode`` reflects process death
+    alone.
+    """
+    with anyio.move_on_after(timeout):
+        while process.returncode is None:
+            await anyio.sleep(_EXIT_POLL_INTERVAL)
+        return True
+    return False
 
 
 def _get_executable_command(command: str) -> str:
@@ -232,39 +297,33 @@ async def _create_platform_compatible_process(
     env: dict[str, str] | None = None,
     errlog: TextIO = sys.stderr,
     cwd: Path | str | None = None,
-):
+) -> Process | FallbackProcess:
     """Creates a subprocess in a platform-compatible way.
 
     Unix: Creates process in a new session/process group for killpg support
     Windows: Creates process in a Job Object for reliable child termination
     """
     if sys.platform == "win32":  # pragma: no cover
-        process = await create_windows_process(command, args, env, errlog, cwd)
+        return await create_windows_process(command, args, env, errlog, cwd)
     else:  # pragma: lax no cover
-        process = await anyio.open_process(
+        return await anyio.open_process(
             [command, *args],
             env=env,
             stderr=errlog,
             cwd=cwd,
             start_new_session=True,
         )
 
-    return process
-
 
-async def _terminate_process_tree(process: Process | FallbackProcess, timeout_seconds: float = 2.0) -> None:
+async def _terminate_process_tree(process: Process | FallbackProcess) -> None:
     """Terminate a process and all its children using platform-specific methods.
 
-    Unix: Uses os.killpg() for atomic process group termination
-    Windows: Uses Job Objects via pywin32 for reliable child process cleanup
-
-    Args:
-        process: The process to terminate
-        timeout_seconds: Timeout in seconds before force killing (default: 2.0)
+    Unix: SIGTERM to the process group, escalating to SIGKILL after
+    ``FORCE_KILL_TIMEOUT``. Windows: immediate Job Object termination.
     """
     if sys.platform == "win32":  # pragma: no cover
-        await terminate_windows_process_tree(process, timeout_seconds)
+        await terminate_windows_process_tree(process)
     else:  # pragma: lax no cover
-        # FallbackProcess should only be used for Windows compatibility
+        # Windows-only FallbackProcess never reaches the POSIX path.
         assert isinstance(process, Process)
-        await terminate_posix_process_tree(process, timeout_seconds)
+        await terminate_posix_process_tree(process, FORCE_KILL_TIMEOUT)

src/mcp/os/posix/utilities.py

@@ -9,49 +9,72 @@
 
 logger = logging.getLogger(__name__)
 
+# How often to probe for surviving process-group members between SIGTERM and SIGKILL.
+_GROUP_POLL_INTERVAL = 0.01
+
 
 async def terminate_posix_process_tree(process: Process, timeout_seconds: float = 2.0) -> None:
-    """Terminate a process and all its children on POSIX systems.
+    """Terminate a process and all its descendants on POSIX systems.
+
+    The process was spawned with ``start_new_session=True``, so it leads its own
+    process group and its pgid equals its pid. ``os.killpg`` on that group reaches
+    every descendant in one atomic call — including descendants whose parent (even
+    the group leader itself) has already exited, which a walk of the process tree
+    would miss.
 
-    Uses os.killpg() for atomic process group termination.
+    Sends SIGTERM to the group, waits up to ``timeout_seconds`` for the group to
+    disappear, then SIGKILLs whatever remains.
 
-    Args:
-        process: The process to terminate
-        timeout_seconds: Timeout in seconds before force killing (default: 2.0)
+    Descendants that move themselves into a new session or process group
+    (daemonizers) escape a group kill by design.
     """
-    pid = getattr(process, "pid", None) or getattr(getattr(process, "popen", None), "pid", None)
-    if not pid:
-        # No PID means there's no process to terminate - it either never started,
-        # already exited, or we have an invalid process object
-        return
+    # start_new_session=True at spawn makes the leader's pid the pgid; do not ask the
+    # OS via getpgid(), which fails with ProcessLookupError once the leader has been
+    # reaped even while other group members are still alive.
+    pgid = process.pid
 
     try:
-        pgid = os.getpgid(pid)
         os.killpg(pgid, signal.SIGTERM)
-
-        with anyio.move_on_after(timeout_seconds):
-            while True:
-                try:
-                    # Check if process group still exists (signal 0 = check only)
-                    os.killpg(pgid, 0)
-                    await anyio.sleep(0.1)
-                except ProcessLookupError:
-                    return
-
+    except ProcessLookupError:
+        # The entire group is already gone; nothing to terminate.
+        return
+    except PermissionError:
+        # kill(2): EPERM means permission was denied for *every* remaining member —
+        # none of the group is ours to signal. Try the leader directly anyway, the one
+        # process we spawned, in case it is the unsignalable survivor.
+        logger.warning("No permission to signal process group %d; terminating the spawned process only", pgid)
         try:
-            os.killpg(pgid, signal.SIGKILL)
-        except ProcessLookupError:
+            process.terminate()
+        except OSError:
+            # The leader itself is unsignalable or already gone; nothing more to do.
             pass
+        return
 
-    except (ProcessLookupError, PermissionError, OSError) as e:
-        logger.warning(f"Process group termination failed for PID {pid}: {e}, falling back to simple terminate")
-        try:
-            process.terminate()
-            with anyio.fail_after(timeout_seconds):
-                await process.wait()
-        except Exception:
-            logger.warning(f"Process termination failed for PID {pid}, attempting force kill")
+    with anyio.move_on_after(timeout_seconds):
+        while True:
             try:
-                process.kill()
-            except Exception:
-                logger.exception(f"Failed to kill process {pid}")
+                # Probe for surviving group members (signal 0 checks without signalling).
+                # The probe keeps succeeding while unreaped zombies remain in the group,
+                # so this also waits out reaping rather than racing it.
+                os.killpg(pgid, 0)
+            except ProcessLookupError:
+                return
+            except PermissionError:
+                # Someone survives but is no longer ours to signal; keep waiting — it
+                # may still exit on its own within the timeout.
+                pass
+            # Touching returncode reaps the leader on trio (the property calls
+            # Popen.poll()); without it nothing reaps during this loop and the
+            # leader's zombie keeps the group alive for the full timeout. On
+            # asyncio it is a cheap attribute read.
+            _ = process.returncode
+            await anyio.sleep(_GROUP_POLL_INTERVAL)
+
+    try:
+        os.killpg(pgid, signal.SIGKILL)
+    except ProcessLookupError:
+        # The group died between the last probe and the kill.
+        pass
+    except PermissionError:
+        # Whatever survives is not ours to kill.
+        pass