Fix Windows graceful process shutdown with CTRL_C_EVENT

src/mcp/client/stdio/win32.py

@@ -2,7 +2,9 @@
 Windows-specific functionality for stdio client operations.
 """
 
+import os
 import shutil
+import signal
 import subprocess
 import sys
 from pathlib import Path
@@ -76,8 +78,29 @@ async def __aexit__(
         exc_tb: object | None,
     ) -> None:
         """Terminate and wait on process exit inside a thread."""
-        self.popen.terminate()
-        await to_thread.run_sync(self.popen.wait)
+        # Try graceful shutdown with CTRL_C_EVENT on Windows
+        if sys.platform == "win32" and hasattr(signal, "CTRL_C_EVENT"):
+            try:
+                # Send CTRL_C_EVENT for graceful shutdown
+                os.kill(self.popen.pid, getattr(signal, "CTRL_C_EVENT"))
+                # Wait for process to exit gracefully (2 second timeout)
+                await to_thread.run_sync(lambda: self.popen.wait(timeout=2))
+            except (subprocess.TimeoutExpired, ProcessLookupError, OSError):
+                # If graceful shutdown fails, fall back to terminate
+                try:
+                    self.popen.terminate()
+                    await to_thread.run_sync(self.popen.wait)
+                except ProcessLookupError:
+                    # Process already exited
+                    pass
+        else:
+            # Non-Windows or fallback behavior
+            try:
+                self.popen.terminate()
+                await to_thread.run_sync(self.popen.wait)
+            except ProcessLookupError:
+                # Process already exited
+                pass
 
         # Close the file handles to prevent ResourceWarning
         if self.stdin:
@@ -163,20 +186,36 @@ async def create_windows_process(
 
 async def terminate_windows_process(process: Process | FallbackProcess):
     """
-    Terminate a Windows process.
+    Terminate a Windows process gracefully.
 
-    Note: On Windows, terminating a process with process.terminate() doesn't
-    always guarantee immediate process termination.
-    So we give it 2s to exit, or we call process.kill()
-    which sends a SIGKILL equivalent signal.
+    First attempts graceful shutdown using CTRL_C_EVENT signal,
+    which allows the process to run cleanup code.
+    Falls back to terminate() and then kill() if graceful shutdown fails.
 
     Args:
         process: The process to terminate
     """
+    # Try graceful shutdown with CTRL_C_EVENT first
+    if isinstance(process, FallbackProcess) and hasattr(signal, "CTRL_C_EVENT"):
+        try:
+            # Send CTRL_C_EVENT for graceful shutdown
+            os.kill(process.popen.pid, getattr(signal, "CTRL_C_EVENT"))
+            with anyio.fail_after(2.0):
+                await process.wait()
+            return
+        except (TimeoutError, ProcessLookupError, OSError):
+            # If CTRL_C_EVENT failed or timed out, continue to forceful termination
+            pass
+
+    # Fall back to terminate
     try:
         process.terminate()
         with anyio.fail_after(2.0):
             await process.wait()
     except TimeoutError:
         # Force kill if it doesn't terminate
-        process.kill()
+        try:
+            process.kill()
+        except ProcessLookupError:
+            # Process already exited
+            pass

tests/client/test_windows_fallback.py

@@ -0,0 +1,200 @@
+"""Test Windows-specific FallbackProcess functionality.
+
+Why this test approach is necessary:
+------------------------------------
+Testing Windows process signal handling requires actual subprocess creation because:
+
+1. SIGNAL HANDLING: We need to verify that CTRL_C_EVENT signals are properly sent and
+   received. This cannot be mocked as it involves OS-level signal propagation between
+   parent and child processes.
+
+2. CLEANUP VERIFICATION: The core issue (#1027) is that cleanup code in lifespan context
+   managers wasn't executing on Windows. We must verify that signal handlers actually run
+   and that cleanup code executes before process termination.
+
+3. WINDOWS-SPECIFIC BEHAVIOR: The FallbackProcess class exists specifically to work around
+   Windows asyncio limitations. Testing it requires actual Windows subprocess creation to
+   ensure the workarounds function correctly.
+
+4. INTEGRATION TESTING: These tests verify the integration between:
+   - FallbackProcess wrapper
+   - Windows signal handling (CTRL_C_EVENT)
+   - Asyncio file streams
+   - Process cleanup behavior
+
+Test Implementation:
+-------------------
+The tests create temporary Python scripts that:
+1. Set up signal handlers for CTRL_C_EVENT
+2. Write marker files to indicate execution state
+3. Allow verification that cleanup ran before termination
+
+This metaprogramming approach is used because:
+- The codebase doesn't have a test fixtures directory pattern
+- Inline `python -c` would be even less readable for complex scripts
+- We need actual subprocess execution to test OS-level behavior
+"""
+
+import os
+import signal
+import sys
+import textwrap
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+import pytest
+
+if TYPE_CHECKING or sys.platform == "win32":
+    from mcp.client.stdio.win32 import create_windows_process
+
+
+@pytest.mark.skipif(os.name != "nt", reason="Windows-specific functionality")
+class TestFallbackProcess:
+    """Test suite for Windows FallbackProcess graceful shutdown."""
+
+    @pytest.mark.anyio
+    async def test_fallback_process_graceful_shutdown(self, tmp_path: Path):
+        """Test that FallbackProcess sends CTRL_C_EVENT for graceful shutdown."""
+        # Create a test script that writes a marker on cleanup
+        test_script = tmp_path / "test_cleanup.py"
+        marker_file = tmp_path / "cleanup_marker.txt"
+
+        # Create a test script that handles CTRL_C_EVENT and writes a marker on cleanup
+        test_script.write_text(
+            textwrap.dedent(f"""
+            import signal
+            import time
+            from pathlib import Path
+
+            marker = Path(r"{marker_file}")
+            marker.write_text("STARTED")
+
+            def cleanup_handler(signum, frame):
+                # This handler should be called when CTRL_C_EVENT is received
+                marker.write_text("CLEANED_UP")
+                exit(0)
+
+            # Register CTRL_C_EVENT handler (SIGINT on Windows)
+            signal.signal(signal.SIGINT, cleanup_handler)
+
+            # Keep process alive waiting for signal
+            while True:
+                time.sleep(0.1)
+        """).strip()
+        )
+
+        # Create process using FallbackProcess
+        process = await create_windows_process(sys.executable, [str(test_script)], cwd=tmp_path)
+
+        # Wait for process to start
+        import asyncio
+
+        await asyncio.sleep(0.5)
+
+        # Verify process started
+        assert marker_file.exists()
+        assert marker_file.read_text() == "STARTED"
+
+        # Exit context manager - should trigger CTRL_C_EVENT
+        await process.__aexit__(None, None, None)
+
+        # Check if cleanup ran
+        await asyncio.sleep(0.5)
+
+        # This is the critical test: cleanup should have executed
+        assert marker_file.read_text() == "CLEANED_UP", "CTRL_C_EVENT cleanup did not execute - issue #1027 not fixed"
+
+    @pytest.mark.anyio
+    async def test_fallback_process_timeout_fallback(self, tmp_path: Path):
+        """Test that FallbackProcess falls back to terminate() if CTRL_C_EVENT times out."""
+        # Create a test script that ignores CTRL_C_EVENT
+        test_script = tmp_path / "test_ignore_signal.py"
+        marker_file = tmp_path / "status_marker.txt"
+
+        # Create a test script that ignores CTRL_C_EVENT to test fallback behavior
+        test_script.write_text(
+            textwrap.dedent(f"""
+            import signal
+            import time
+            from pathlib import Path
+
+            marker = Path(r"{marker_file}")
+            marker.write_text("STARTED")
+
+            # Explicitly ignore CTRL_C_EVENT to test fallback to terminate()
+            signal.signal(signal.SIGINT, signal.SIG_IGN)
+
+            # Keep process alive - should be forcefully terminated
+            while True:
+                time.sleep(0.1)
+        """).strip()
+        )
+
+        # Create process
+        process = await create_windows_process(sys.executable, [str(test_script)], cwd=tmp_path)
+
+        # Wait for process to start
+        import asyncio
+
+        await asyncio.sleep(0.5)
+
+        assert marker_file.exists()
+        assert marker_file.read_text() == "STARTED"
+
+        # Exit context manager - should try CTRL_C_EVENT, timeout, then terminate
+        await process.__aexit__(None, None, None)
+
+        # Process should be terminated even though it ignored CTRL_C_EVENT
+        # Check that process is no longer running
+        try:
+            # This should raise because process is terminated
+            os.kill(process.popen.pid, 0)
+            pytest.fail("Process should have been terminated")
+        except (ProcessLookupError, OSError):
+            # Expected - process is terminated
+            pass
+
+    def test_ctrl_c_event_availability(self):
+        """Test that CTRL_C_EVENT is available on Windows."""
+        assert hasattr(signal, "CTRL_C_EVENT"), "CTRL_C_EVENT not available on this Windows system"
+
+        # Verify it's the expected value (should be 0)
+        assert signal.CTRL_C_EVENT == 0
+
+    @pytest.mark.anyio
+    async def test_fallback_process_with_stdio(self, tmp_path: Path):
+        """Test that FallbackProcess properly wraps stdin/stdout streams."""
+        # Create a simple echo script to test stdio stream wrapping
+        echo_script = tmp_path / "echo.py"
+        echo_script.write_text(
+            textwrap.dedent("""
+            import sys
+            while True:
+                line = sys.stdin.readline()
+                if not line:
+                    break
+                sys.stdout.write(f"ECHO: {line}")
+                sys.stdout.flush()
+        """).strip()
+        )
+
+        # Create process
+        process = await create_windows_process(sys.executable, [str(echo_script)], cwd=tmp_path)
+
+        # Test async I/O
+        assert process.stdin is not None
+        assert process.stdout is not None
+
+        # Write to stdin
+        test_message = b"Hello Windows\\n"
+        await process.stdin.send(test_message)
+
+        # Read from stdout
+        import asyncio
+
+        response = await asyncio.wait_for(process.stdout.receive(1024), timeout=2.0)
+
+        assert b"ECHO: Hello Windows" in response
+
+        # Cleanup
+        await process.__aexit__(None, None, None)