Consolidate stdio tests: in-process logic tests, minimal real-process set

The stdio client's logic — framing, parse errors, the shutdown
escalation decisions — is now tested in process against a fake process
injected through the spawn seam, and only the properties that are OS
behaviour keep real subprocesses. The in-process tests include
regressions for the shutdown bugs fixed in the previous commit:
cancellation must still run the full shutdown, tree kill must reach
children after the leader has exited, writes racing server death must
surface clean closure, exiting with a server message still undelivered
must be a clean exit, and a server exiting on stdin close must never
be terminated. Each was verified to fail against the previous
implementation.

The real-process set shrinks to: one consolidated tree-kill test driven
through the public stdio_client (a parent that exits instantly on
SIGTERM, a child, and a grandchild — pinning group inheritance, atomic
group kill, and dead-leader robustness in one spawn), the
SIGTERM-ignoring SIGKILL escalation test, a dead-group no-op test, and
exec-failure ENOENT. Grace and kill timeouts are shortened via
monkeypatch in the real tests: the escalation decision is pinned in
process against a shortened grace, the SIGTERM-then-SIGKILL escalation
itself is exercised only by the real escalation test, and the
production constants' values are deliberately unpinned. The stdio
surface drops from ~11s to ~3.5s.

Deleted as subsumed or broken:
- The two tee-based tests: the interaction suite's subprocess e2e pins
  the real-pipe round trip with a real server, and tee could never
  deterministically exercise the chunk-reassembly buffer that the new
  in-process framing test pins (real pipes deliver short messages as
  whole lines).
- test_stdio_client_bad_path: its script only "failed" because
  non-existent-file.py happens to be a Python expression that raises
  NameError; the property (stdout EOF fails initialize with
  CONNECTION_CLOSED) is now pinned in process.
- The basic/early-parent cleanup tests and universal-cleanup /
  graceful-stdin-exit tests, folded into the consolidated tree test and
  the in-process graceful/escalation tests.
- tests/issues/test_1027_win_unreachable_cleanup.py: its lifespan
  cleanup chain is covered by the interaction e2e's clean-exit marker
  plus the lifespan tests, it polled marker files with fixed sleeps,
  and it leaked its child process when an assertion failed mid-test.
  Its only helper module (tests/shared/test_win32_utils.py) goes with
  it. The MCPServer.run("stdio") coverage its subprocess scripts
  provided is replaced by an in-process test that drives run() over
  injected stdio and asserts a real response.

tests/issues/test_552_windows_hang.py now asserts initialize succeeded
instead of swallowing every exception (its handcrafted response also
sent the invalid protocolVersion "1.0", so it errored on every run and
passed anyway), the stdio server tests are bounded by fail_after, and
the elicitation tests that claimed "stdio" in their names run, and now
say they run, over the in-memory transport.

Author: maxisbey

tests/client/test_stdio.py

@@ -69,7 +69,12 @@ async def collect(params: LoggingMessageNotificationParams) -> None:
                 # stdio_client deliberately filters the inherited environment to a safe minimum,
                 # which drops the variables coverage.py's subprocess support uses; pass them through
                 # so the server module is measured. Empty when not running under coverage.
-                env={key: value for key, value in os.environ.items() if key.startswith("COVERAGE_")},
+                # SyntaxWarning is suppressed because the child compiles dependencies from source
+                # (pytest's pyc tag doesn't match a plain python child's): at the anyio>=4.9 floor,
+                # Python 3.14 emits a compile-time warning for anyio's return-in-finally, which
+                # would land on the snapshot-asserted stderr below.
+                env={key: value for key, value in os.environ.items() if key.startswith("COVERAGE_")}
+                | {"PYTHONWARNINGS": "ignore::SyntaxWarning"},
             ),
             errlog=errlog,
         )

tests/interaction/transports/test_stdio.py

@@ -1,5 +1,6 @@
 """Test for issue #552: stdio_client hangs on Windows."""
 
+import json
 import sys
 from textwrap import dedent
 
@@ -8,56 +9,48 @@
 
 from mcp import ClientSession, StdioServerParameters
 from mcp.client.stdio import stdio_client
+from mcp.types import LATEST_PROTOCOL_VERSION, InitializeResult
 
 
 @pytest.mark.skipif(sys.platform != "win32", reason="Windows-specific test")  # pragma: no cover
 @pytest.mark.anyio
-async def test_windows_stdio_client_with_session():
-    """Test the exact scenario from issue #552: Using ClientSession with stdio_client.
-
-    This reproduces the original bug report where stdio_client hangs on Windows 11
-    when used with ClientSession.
+async def test_initialize_succeeds_and_shutdown_returns_after_the_server_exits_mid_session():
+    """Initialize completes (and shutdown returns) against a server that responds and
+    then exits mid-session — the proactor pipe scenario that hung on Windows 11 in
+    issue #552. The positive assertion matters: a session that *errors* quickly would
+    also "not hang", so finishing fast alone proves nothing.
     """
-    # Create a minimal MCP server that responds to initialization
-    server_script = dedent("""
+    # A minimal server: answer initialize correctly, then exit.
+    server_script = dedent(f"""
         import json
         import sys
 
-        # Read initialization request
         line = sys.stdin.readline()
+        request = json.loads(line)
 
-        # Send initialization response
-        response = {
+        response = {{
             "jsonrpc": "2.0",
-            "id": 1,
-            "result": {
-                "protocolVersion": "1.0",
-                "capabilities": {},
-                "serverInfo": {"name": "test-server", "version": "1.0"}
-            }
-        }
+            "id": request["id"],
+            "result": {{
+                "protocolVersion": {json.dumps(LATEST_PROTOCOL_VERSION)},
+                "capabilities": {{}},
+                "serverInfo": {{"name": "test-server", "version": "1.0"}}
+            }}
+        }}
         print(json.dumps(response))
         sys.stdout.flush()
-
-        # Exit after a short delay
-        import time
-        time.sleep(0.1)
-        sys.exit(0)
     """).strip()
 
     params = StdioServerParameters(
         command=sys.executable,
         args=["-c", server_script],
     )
 
-    # This is the exact pattern from the bug report
     with anyio.fail_after(10):
-        try:
-            async with stdio_client(params) as (read, write):
-                async with ClientSession(read, write) as session:
-                    await session.initialize()
-                # Should exit ClientSession without hanging
-            # Should exit stdio_client without hanging
-        except Exception:
-            # Connection errors are expected when process exits
-            pass
+        async with stdio_client(params) as (read, write):
+            async with ClientSession(read, write) as session:
+                result = await session.initialize()
+                assert isinstance(result, InitializeResult)
+                assert result.server_info.name == "test-server"
+            # Exiting ClientSession and stdio_client must not hang even though the
+            # server process is already gone.

tests/issues/test_1027_win_unreachable_cleanup.py

@@ -1,4 +1,4 @@
-"""Test the elicitation feature using stdio transport."""
+"""Test the elicitation feature over the in-memory client transport."""
 
 from typing import Any
 
@@ -58,9 +58,9 @@ async def call_tool_and_assert(
 
 
 @pytest.mark.anyio
-async def test_stdio_elicitation():
-    """Test the elicitation feature using stdio transport."""
-    mcp = MCPServer(name="StdioElicitationServer")
+async def test_elicitation_accept_returns_the_users_answer_to_the_tool():
+    """An accepted elicitation delivers the user's content back to the requesting tool."""
+    mcp = MCPServer(name="ElicitationServer")
     create_ask_user_tool(mcp)
 
     # Create a custom handler for elicitation requests
@@ -76,9 +76,9 @@ async def elicitation_callback(context: RequestContext[ClientSession], params: E
 
 
 @pytest.mark.anyio
-async def test_stdio_elicitation_decline():
-    """Test elicitation with user declining."""
-    mcp = MCPServer(name="StdioElicitationDeclineServer")
+async def test_elicitation_decline_reaches_the_tool_without_content():
+    """A declined elicitation reports the decline to the tool, with no content attached."""
+    mcp = MCPServer(name="ElicitationDeclineServer")
     create_ask_user_tool(mcp)
 
     async def elicitation_callback(context: RequestContext[ClientSession], params: ElicitRequestParams):