Add real-subprocess stdio lifecycle tests for POSIX and Windows

New tests/transports/stdio/ suite pinning the client transport's
process-lifecycle contracts against real subprocesses, synchronized
through TCP liveness sockets (kernel FIN/RST and connect-back) with no
sleeps and no timing assertions:

- Both platforms: a well-behaved server exits on stdin close and is
  reaped without escalation; cancellation mid-session still terminates
  the whole server tree; a server that exits mid-session keeps its own
  exit code; server stderr reaches the errlog file; FallbackProcess
  reports death through returncode without wait() and its wait() is
  cancellable (the SelectorEventLoop fallback wraps a plain Popen, so
  these run on every platform with waitid where available).
- POSIX: a gracefully-exited server's background child survives client
  shutdown (the documented policy), proven by an echo round trip plus a
  never-escalated seam; the surviving child's next write to its
  inherited stdout fails with EPIPE once the client has released its
  pipe ends.
- Windows: the same scenario's documented opposite outcome - the child
  is reaped when the Job Object handle closes at shutdown; a
  SelectorEventLoop session engages the FallbackProcess wrapper and
  completes a clean lifecycle; a print()-based server emitting CRLF
  line endings round-trips messages.

Shared helpers live in _liveness.py (extracted from the real-process
section of tests/client/test_stdio.py; consolidating that file onto
this module is follow-up work), with conftest fixtures recording the
spawn and terminate seams and reaping spawn-time process groups on
teardown.

Author: maxisbey

tests/transports/__init__.py

@@ -0,0 +1,98 @@
+"""Kernel-synchronized liveness probes for the real-subprocess stdio lifecycle suite.
+
+A spawned (grand)child connects back to a test-owned TCP listener and sends
+`b'alive'`. From there the kernel provides every signal a test needs, with no
+sleeps or polling anywhere:
+
+1. `accept_alive` blocks until the subprocess connects, proving it is running (and
+   that the script lines before the connect have executed).
+2. `assert_stream_closed` proves the peer terminated: the kernel closes all of a
+   process's file descriptors on exit, surfacing EOF (clean close / FIN) or
+   `BrokenResourceError` (abrupt close / RST, typical of SIGKILL and Windows job
+   termination).
+3. `assert_peer_echoes` proves the peer is *alive*: only a running process can
+   answer an echo, so a positive reply cannot race a kill the way a "no FIN yet"
+   observation could.
+
+These helpers are extracted from the real-process section of
+tests/client/test_stdio.py; the two copies on this branch are deliberate —
+consolidating that file onto this module is follow-up work.
+"""
+
+import anyio
+import anyio.abc
+import pytest
+
+
+def connect_back_script(port: int, *, echo: bool = False) -> str:
+    """Return a `python -c` script body that connects to 127.0.0.1:`port` and
+    sends `b'alive'`.
+
+    By default the process then blocks forever, serving as a pure liveness beacon
+    for kill/termination tests. With `echo=True` it instead echoes every received
+    chunk back (the recv parks it just as indefinitely), so a survival test can
+    prove the process is still running after the client is gone — see
+    `assert_peer_echoes`.
+    """
+    # Excluded from coverage (lax: exempt from strict-no-cover): echo mode is
+    # used only by POSIX-gated tests, and Windows runners enforce 100% per job.
+    if echo:  # pragma: lax no cover
+        tail = "while True:\n    data = s.recv(65536)\n    if not data:\n        break\n    s.sendall(data)\n"
+    else:
+        tail = "time.sleep(3600)\n"
+    return f"import socket, time\ns = socket.create_connection(('127.0.0.1', {port}))\ns.sendall(b'alive')\n" + tail
+
+
+async def open_liveness_listener() -> tuple[anyio.abc.SocketListener, int]:
+    """Open a TCP listener on localhost and return it along with its port."""
+    multi = await anyio.create_tcp_listener(local_host="127.0.0.1")
+    sock = multi.listeners[0]
+    assert isinstance(sock, anyio.abc.SocketListener)
+    addr = sock.extra(anyio.abc.SocketAttribute.local_address)
+    # IPv4 local_address is (host: str, port: int)
+    assert isinstance(addr, tuple) and len(addr) >= 2 and isinstance(addr[1], int)
+    return sock, addr[1]
+
+
+async def accept_alive(sock: anyio.abc.SocketListener) -> anyio.abc.SocketStream:
+    """Accept one connection and assert the peer sent `b'alive'`.
+
+    Blocks deterministically until a subprocess connects (no polling), reading
+    until the full 5-byte banner has arrived — TCP may legally split even a tiny
+    send. The calling test bounds this with `anyio.fail_after` to catch the case
+    where the subprocess chain failed to start.
+    """
+    stream = await sock.accept()
+    msg = b""
+    while len(msg) < 5:
+        msg += await stream.receive(5 - len(msg))
+    assert msg == b"alive", f"expected b'alive', got {msg!r}"
+    return stream
+
+
+async def assert_stream_closed(stream: anyio.abc.SocketStream) -> None:
+    """Assert the peer holding the other end of `stream` has terminated."""
+    with anyio.fail_after(5.0), pytest.raises((anyio.EndOfStream, anyio.BrokenResourceError)):
+        await stream.receive(1)
+
+
+async def assert_peer_echoes(stream: anyio.abc.SocketStream) -> None:  # pragma: lax no cover
+    """Assert the peer holding the other end of `stream` is still running, by
+    round-tripping one echo through it (the peer must use `echo=True`).
+
+    A dead process can never answer, so under a regression that kills the peer this
+    raises (EOF/reset) or times out via the bound — it cannot pass spuriously; the
+    sub-millisecond window between a kill being issued and taking effect is dwarfed
+    by the socket round trip that must complete after it.
+
+    Excluded from coverage (lax: exempt from strict-no-cover) like `connect_back_script`'s
+    echo mode: only POSIX-gated survival tests call this, and Windows runners enforce
+    100% coverage per job.
+    """
+    with anyio.fail_after(5.0):
+        await stream.send(b"ping")
+        # Read until the full echo has arrived: TCP may legally split even a tiny send.
+        echoed = b""
+        while len(echoed) < 4:
+            echoed += await stream.receive(4 - len(echoed))
+    assert echoed == b"ping", f"expected b'ping', got {echoed!r}"

tests/transports/stdio/__init__.py

@@ -0,0 +1,82 @@
+"""Fixtures for the stdio lifecycle suite: recording seams around the spawn and
+tree-termination internals of `stdio_client` (the real implementations still run),
+plus the failure-path safety net that keeps a crashed test from orphaning its
+sleep-forever subprocesses.
+"""
+
+import os
+import signal
+import sys
+from collections.abc import Generator
+from contextlib import suppress
+from pathlib import Path
+from typing import TextIO
+
+import anyio.abc
+import pytest
+
+from mcp.client import stdio
+from mcp.client.stdio import _create_platform_compatible_process, _terminate_process_tree
+from mcp.os.win32.utilities import FallbackProcess
+
+
+@pytest.fixture
+def spawned_processes(
+    monkeypatch: pytest.MonkeyPatch,
+) -> Generator[list[anyio.abc.Process | FallbackProcess]]:
+    """Record every process `stdio_client` spawns; the real spawn still runs.
+
+    Tests inspect the recorded processes afterwards (exit codes, concrete type on
+    the Windows fallback path). Teardown SIGKILLs each spawn-time process group on
+    POSIX, in both of its roles: failure-path safety net (a test that dies mid-body
+    cannot orphan its sleep-forever descendants for an hour) and the reaper for
+    tests that deliberately leave a survivor running, like the POSIX survival
+    test's echo child. On Windows there is no process group to signal (the Job
+    Object covers strays).
+    """
+    spawned: list[anyio.abc.Process | FallbackProcess] = []
+
+    async def recording_spawn(
+        command: str,
+        args: list[str],
+        env: dict[str, str] | None = None,
+        errlog: TextIO = sys.stderr,
+        cwd: Path | str | None = None,
+    ) -> anyio.abc.Process | FallbackProcess:
+        process = await _create_platform_compatible_process(command, args, env, errlog, cwd)
+        spawned.append(process)
+        return process
+
+    monkeypatch.setattr(stdio, "_create_platform_compatible_process", recording_spawn)
+    yield spawned
+    _kill_spawn_groups(spawned)
+
+
+@pytest.fixture
+def terminate_calls(monkeypatch: pytest.MonkeyPatch) -> list[anyio.abc.Process | FallbackProcess]:
+    """Record every invocation of `stdio_client`'s tree-termination seam; the real
+    termination still runs.
+
+    An empty list after the context exits proves the graceful path: the server was
+    never escalated against, which a socket signal alone cannot establish (a FIN
+    looks the same whether the peer exited on stdin closure or was killed).
+    """
+    terminated: list[anyio.abc.Process | FallbackProcess] = []
+
+    async def recording_terminate(process: anyio.abc.Process | FallbackProcess) -> None:
+        terminated.append(process)
+        await _terminate_process_tree(process)
+
+    monkeypatch.setattr(stdio, "_terminate_process_tree", recording_terminate)
+    return terminated
+
+
+# Excluded from coverage (lax: exempt from strict-no-cover): registered on every
+# platform but a no-op on Windows, whose runners enforce 100% coverage per job.
+def _kill_spawn_groups(spawned: list[anyio.abc.Process | FallbackProcess]) -> None:  # pragma: lax no cover
+    """SIGKILL each spawn-time process group; see `spawned_processes`."""
+    if sys.platform == "win32":
+        return
+    for process in spawned:
+        with suppress(ProcessLookupError):
+            os.killpg(process.pid, signal.SIGKILL)