Address review findings: comment accuracy, test hardening, trio reap pin

Source and docs (comment/docstring-level; one dead-code removal):
- Correct the shutdown shield comment: a native task.cancel() delivered
  while cleanup is in progress can abort it regardless of count; the
  consumed-at-the-yield arithmetic only covered cancellation that
  initiated teardown.
- Scope the pipe-gated process.wait() claims to asyncio on Python
  3.11+ (3.10 and trio resolve on process exit alone), in the
  _wait_for_process_exit docstring and the migration guide.
- Add sunset notes to the cpython#106749 tracer workarounds (3.11-only;
  revert when 3.11 support is dropped).
- Document stdio_client's raised exceptions (Raises: section).
- Document the Job Object pre-assignment window in create_windows_process
  and at the assignment call site; document the _process_jobs
  WeakKeyDictionary's two load-bearing invariants (PyHANDLE values,
  identity-hashed weakref keys); fix a wrong comment about the
  CloseHandle failure path in _maybe_assign_process_to_job.
- Drop FallbackProcess.stdin_raw/stdout_raw (write-only attributes).

Tests:
- Fix the inert leak canary in the OSError spawn-failure test: the
  pytest.raises ExceptionInfo pinned the leaked streams across
  gc.collect(); drop it before collecting (mutation-verified).
- Correct the EPERM escalation test's pre-fix description (the old code
  fell back to a leader-only kill, leaking other group members - it did
  not hang forever) and trim an unasserted claim from the
  write-after-death docstring.
- Add a CRLF-framed message to the chunk-reframing test, pinning the
  trailing-CR tolerance Windows servers rely on, on every platform.
- Pin the grace wait's returncode read (what reaps the leader's zombie
  on trio) with a read-counting stub, parametrized over asyncio and
  trio; deleting the read fails the test on both backends.
- Add the missing width-justification comment to one fail_after(10.0).

Author: maxisbey

docs/migration.md

@@ -120,7 +120,10 @@ child processes the server leaves behind are no longer killed on POSIX — their
 lifetime is the server's business. The old behavior was a side effect of a shutdown
 wait gated on the stdio pipes closing rather than on process exit: a child holding
 an inherited pipe made a well-behaved server look hung, so its whole process tree
-was killed. A server that does not exit within the grace period is still terminated
+was killed. (That gating is an asyncio behavior specific to Python 3.11+ — on
+Python 3.10 and the trio backend the old wait already resolved on process exit, so
+the spurious kill never fired there.) A server that does not exit within the grace
+period is still terminated
 along with its entire process group. On Windows, children stay in the server's Job
 Object and are still killed at shutdown — now deterministically when the job handle
 is closed, rather than whenever the handle happened to be garbage-collected.

src/mcp/client/stdio.py

@@ -126,6 +126,12 @@ class StdioServerParameters(BaseModel):
 async def stdio_client(server: StdioServerParameters, errlog: TextIO = sys.stderr):
     """Client transport for stdio: this will connect to a server by spawning a
     process and communicating with it over stdin/stdout.
+
+    Raises:
+        OSError: If the server process cannot be spawned (for example, the command
+            does not exist or is not executable).
+        ValueError: If the spawn parameters are invalid (for example, an embedded
+            NUL byte in the command or an argument).
     """
     read_stream: MemoryObjectReceiveStream[SessionMessage | Exception]
     read_stream_writer: MemoryObjectSendStream[SessionMessage | Exception]
@@ -168,7 +174,8 @@ async def stdio_client(server: StdioServerParameters, errlog: TextIO = sys.stder
     # Shutdown's final cancellation targets these instead of the task group's own
     # scope: cancelling the host scope would deliver the cancellation by throwing
     # through the caller's suspended frames, and Python 3.11's tracer loses
-    # coverage events after such a throw() traversal (python/cpython#106749).
+    # coverage events after such a throw() traversal (python/cpython#106749;
+    # 3.11-only — see the sunset note in `_wait_for_process_exit`).
     # These scope exactly the work the pipe tasks own.
     reader_scope = anyio.CancelScope()
     writer_scope = anyio.CancelScope()
@@ -261,9 +268,9 @@ async def stdin_writer() -> None:
             # 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.
+            # anyio-level cancellation; a native task.cancel() delivered while the
+            # cleanup is in progress can still abort it — there is no backend-neutral
+            # way to refuse native cancellation.
             with anyio.CancelScope(shield=True):
                 # Let the writer hand any message the transport already accepted to
                 # the server's stdin before that stdin closes; a zero-buffer send
@@ -377,19 +384,22 @@ def _close_subprocess_transport(process: Process | FallbackProcess) -> None:
 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.
+    Deliberately does not use `process.wait()`: on asyncio under Python 3.11+ it
+    resolves only once the process has exited *and* every one of its pipes has
+    closed (3.10 and trio resolve on exit alone) — 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 on every backend.
     """
     # Implemented as a plain deadline loop rather than `anyio.move_on_after()`: a
     # cancel scope's deadline fires by throwing a cancellation through every frame
     # suspended in the await chain, including the caller's, and Python 3.11's tracer
     # loses coverage events in a frame resumed after such a throw() traversal
     # (python/cpython#106749). With no cancel scope, the timeout path completes a
-    # normal `sleep()` and returns, so no frame is ever thrown through.
+    # normal `sleep()` and returns, so no frame is ever thrown through. The tracer
+    # bug is 3.11-only (fixed in 3.12, wontfix on 3.11): revert this and the other
+    # workaround sites that cite it to their natural cancel-scope forms when
+    # Python 3.11 support is dropped.
     deadline = anyio.current_time() + timeout
     while process.returncode is None:
         if anyio.current_time() >= deadline:

src/mcp/os/win32/utilities.py

@@ -33,8 +33,20 @@
 _EXIT_POLL_INTERVAL = 0.01
 
 # The Job Object each spawned process was assigned to, so the process tree can be
-# terminated through it later. Keyed weakly: if a process object goes away without
-# explicit termination, the entry goes with it.
+# terminated through it later.
+#
+# Values must stay the pywin32 `PyHANDLE` returned by `CreateJobObject`, never a
+# detached int: on abandoned-shutdown paths where neither pop site runs, the dying
+# weak entry drops the last reference to the `PyHANDLE`, whose destructor closes
+# the OS handle — and `KILL_ON_JOB_CLOSE` then reaps the orphaned tree. That
+# destructor-close is the only backstop on those paths; storing anything but the
+# `PyHANDLE` would turn it into a permanent handle leak.
+#
+# Keys rely on anyio's `Process` being weakref-able and identity-hashed (it is a
+# `dataclass(eq=False)` without `__slots__`); if that ever changes, registration
+# fails loudly with a `TypeError` rather than silently. Entries are written once,
+# after assignment can no longer fail, and consumed via `pop()` on the event
+# loop — no locking needed.
 _process_jobs: "weakref.WeakKeyDictionary[Process | FallbackProcess, object]" = weakref.WeakKeyDictionary()
 
 
@@ -80,11 +92,11 @@ class FallbackProcess:
 
     def __init__(self, popen_obj: subprocess.Popen[bytes]):
         self.popen: subprocess.Popen[bytes] = popen_obj
-        self.stdin_raw = popen_obj.stdin
-        self.stdout_raw = popen_obj.stdout
+        stdin = popen_obj.stdin
+        stdout = popen_obj.stdout
 
-        self.stdin = FileWriteStream(cast(BinaryIO, self.stdin_raw)) if self.stdin_raw else None
-        self.stdout = FileReadStream(cast(BinaryIO, self.stdout_raw)) if self.stdout_raw else None
+        self.stdin = FileWriteStream(cast(BinaryIO, stdin)) if stdin else None
+        self.stdout = FileReadStream(cast(BinaryIO, stdout)) if stdout else None
 
     async def wait(self) -> int:
         """Wait for process exit by polling.
@@ -133,8 +145,11 @@ async def create_windows_process(
     when using the SelectorEventLoop, which does not support async subprocesses.
     In that case, we fall back to using subprocess.Popen.
 
-    The process is automatically added to a Job Object to ensure all child
-    processes are terminated when the parent is terminated.
+    The process is added to a Job Object so that child processes are terminated
+    with it. Children the server spawns before the assignment completes — a
+    window of two API calls against the server's interpreter cold start — are
+    not captured: job membership is inherited at process creation, never
+    acquired retroactively.
 
     Args:
         command (str): The executable to run
@@ -160,7 +175,11 @@ async def create_windows_process(
         process = await _create_windows_fallback_process(command, args, env, errlog, cwd)
 
     # Created only after a successful spawn: a failed spawn raises before any job
-    # exists, so there is no handle to leak on that path.
+    # exists, so there is no handle to leak on that path. Children the server
+    # spawns before AssignProcessToJobObject completes land outside the job
+    # (membership is inherited at CreateProcess, never acquired retroactively);
+    # the window is two API calls racing the server's interpreter cold start. If
+    # it ever bites, the fix is a CREATE_SUSPENDED spawn -> assign -> resume.
     job = _create_job_object()
     _maybe_assign_process_to_job(process, job)
     return process
@@ -239,8 +258,11 @@ def _maybe_assign_process_to_job(process: Process | FallbackProcess, job: object
             win32job.AssignProcessToJobObject(job, process_handle)
         finally:
             win32api.CloseHandle(process_handle)
-        # Recorded only once nothing else can fail, so the failure branch below can
-        # assume the job is unregistered and just close it.
+        # Recorded only after the process-handle close above. If that close failed
+        # post-assignment, the except below would close the job handle and
+        # KILL_ON_JOB_CLOSE would take the just-assigned healthy server with it —
+        # accepted, because CloseHandle cannot realistically fail on a handle
+        # OpenProcess just returned.
         _process_jobs[process] = job
     except pywintypes.error:
         logger.warning("Failed to assign process %d to Job Object", process.pid, exc_info=True)

tests/client/test_stdio.py

@@ -210,29 +210,34 @@ async def _next_message(read_stream: MemoryObjectReceiveStream[SessionMessage |
 
 @pytest.mark.anyio
 async def test_messages_split_and_packed_across_chunks_are_reframed(monkeypatch: pytest.MonkeyPatch) -> None:
-    """Framing survives arbitrary chunk boundaries: a message split mid-byte and two
-    messages packed into one chunk are each delivered exactly once, and a trailing
-    line without a newline is not delivered.
+    """Framing survives arbitrary chunk boundaries: a message split mid-byte, two
+    messages packed into one chunk, and a CRLF-terminated message are each delivered
+    exactly once, and a trailing line without a newline is not delivered.
 
     A real pipe delivers short writes as whole lines, so only an in-process stdout
     can pin the reassembly buffer deterministically.
     """
     ping = JSONRPCRequest(jsonrpc="2.0", id=1, method="ping")
     pong = JSONRPCResponse(jsonrpc="2.0", id=1, result={})
+    ping2 = JSONRPCRequest(jsonrpc="2.0", id=2, method="ping")
     process = FakeProcess(on_stdin_close=lambda: process.exit(0))
 
     install_fake_process(monkeypatch, process)
 
     with anyio.fail_after(5):
         async with stdio_client(FAKE_PARAMS) as (read_stream, _):
             # Split the first message mid-bytes, pack the rest of it together with
-            # the whole second message and a partial third into one chunk.
+            # the whole second message, a CRLF-framed third (the SDK's own server
+            # emits \r\n on Windows; jiter treats the \r as JSON whitespace), and a
+            # partial fourth into one chunk.
             wire = _line(ping)
+            crlf_wire = ping2.model_dump_json(by_alias=True, exclude_unset=True).encode() + b"\r\n"
             await process.feed(wire[:7])
-            await process.feed(wire[7:] + _line(pong) + b'{"jsonrpc": "2.0", "id": 99')
+            await process.feed(wire[7:] + _line(pong) + crlf_wire + b'{"jsonrpc": "2.0", "id": 99')
 
             assert await _next_message(read_stream) == ping
             assert await _next_message(read_stream) == pong
+            assert await _next_message(read_stream) == ping2
 
             # The partial trailing message is dropped at EOF rather than delivered
             # broken: EOF ends the read stream with nothing further on it.
@@ -411,8 +416,8 @@ async def run_client_until_cancelled() -> None:
 @pytest.mark.anyio
 async def test_writing_after_the_server_dies_reports_clean_closure(monkeypatch: pytest.MonkeyPatch) -> None:
     """A send racing the server's death must not surface a raw backend exception
-    (ConnectionResetError in an exception group); the death is reported through the
-    read stream's closure and the transport still shuts down cleanly."""
+    (ConnectionResetError in an exception group) out of the context manager; the
+    transport still shuts down cleanly."""
     ping = JSONRPCRequest(jsonrpc="2.0", id=1, method="ping")
     process = FakeProcess(on_stdin_close=lambda: process.exit(0))
 
@@ -471,6 +476,9 @@ async def failing_spawn(
             pass  # pragma: no cover
 
     assert exc_info.value.errno == errno.EACCES
+    # Drop the ExceptionInfo before collecting: its traceback references the suspended
+    # stdio_client frame, which would keep leaked streams alive across the collect.
+    del exc_info
     gc.collect()
 
 
@@ -822,8 +830,8 @@ async def test_an_eperm_group_that_outlives_the_grace_period_is_still_sigkilled(
 ) -> None:
     """Even when every probe reports EPERM, the SIGKILL escalation still fires after
     the grace period (and its own EPERM is tolerated) — pre-fix, EPERM at SIGTERM
-    returned early and a SIGTERM-ignoring server survived shutdown forever. The tiny
-    timeout is the time-based grace period under test."""
+    abandoned the group escalation for a leader-only kill, leaking every other group
+    member. The tiny timeout is the time-based grace period under test."""
     calls: list[tuple[int, int]] = []
 
     def fake_killpg(pgid: int, sig: int) -> None:
@@ -843,6 +851,53 @@ def fake_killpg(pgid: int, sig: int) -> None:
     assert set(calls[1:-1]) == {(stub.pid, 0)}
 
 
+@pytest.mark.anyio
+@pytest.mark.parametrize("anyio_backend", ["asyncio", "trio"])
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX killpg semantics")
+# Excluded from coverage for the same Windows-runner reason as above.
+async def test_the_grace_wait_reads_returncode_so_trio_can_reap_the_leaders_zombie(  # pragma: lax no cover
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The wait between SIGTERM and SIGKILL reads `process.returncode` while it polls:
+    on trio that property calls `Popen.poll()`, and that reap is what stops the
+    leader's zombie from keeping the process group alive for the full timeout (see the
+    comment in `terminate_posix_process_tree`). Regression pin for the read itself, on
+    both backends — the reaping side effect is trio's documented `returncode`
+    behaviour, deliberately not re-tested here."""
+
+    calls: list[tuple[int, int]] = []
+
+    def fake_killpg(pgid: int, sig: int) -> None:
+        # SIGTERM is accepted and every liveness probe reports survivors, so the
+        # grace wait runs to its (tiny) timeout and the SIGKILL escalation fires.
+        calls.append((pgid, sig))
+
+    class _ReadCountingProcess:
+        """A live-forever leader whose `returncode` property counts its reads."""
+
+        pid = 54321
+
+        def __init__(self) -> None:
+            self.returncode_reads = 0
+
+        @property
+        def returncode(self) -> int | None:
+            self.returncode_reads += 1
+            return None
+
+    monkeypatch.setattr(posix_utilities.os, "killpg", fake_killpg)
+    stub = _ReadCountingProcess()
+
+    with anyio.fail_after(5):
+        await terminate_posix_process_tree(cast(anyio.abc.Process, stub), timeout_seconds=0.05)
+
+    # The wait ran to its deadline (the escalation fired)...
+    assert calls[0] == (stub.pid, signal.SIGTERM)
+    assert calls[-1] == (stub.pid, signal.SIGKILL)
+    # ...and `returncode` was read while it polled — the read that reaps on trio.
+    assert stub.returncode_reads >= 1
+
+
 # ---------------------------------------------------------------------------
 # Real-process tests: the OS facts no fake can certify
 # ---------------------------------------------------------------------------
@@ -1056,6 +1111,8 @@ async def test_terminating_an_already_exited_process_is_a_no_op() -> None:  # pr
     proc = await _create_platform_compatible_process(sys.executable, ["-c", "pass"])
     assert isinstance(proc, anyio.abc.Process)
 
+    # The bound covers one interpreter cold start on a loaded runner; a healthy run
+    # takes well under a second.
     with anyio.fail_after(10.0):
         await _wait_until_exited(proc)
         await _terminate_process_tree(proc)