fix(mcp): isolate stdio parsing from process pools

CHANGELOG.md

@@ -20,6 +20,10 @@
 - Incremental Git change discovery now reads NUL-delimited byte output, so
   Unicode, whitespace, newline, and literal arrow paths are preserved while
   rename/copy records keep destination-only semantics (PR #618).
+- MCP stdio servers now use thread-based parallel parsing on every platform,
+  preventing inherited transport descriptors from keeping Unix servers and
+  workers alive after host disconnects, while normal non-interactive CLI/CI
+  builds retain the faster process-pool default (PR #615).
 - Corrected TESTED_BY edge direction across graph, refactor, and transitive-test
   consumers, with a parser-to-store-to-query regression (#527/#559/#598 class).
 - C# receiver calls now capture bare, chained, member, and null-conditional

code_review_graph/incremental.py

@@ -24,15 +24,22 @@
 
 _MAX_PARSE_WORKERS = int(os.environ.get("CRG_PARSE_WORKERS", str(min(os.cpu_count() or 4, 8))))
 
+# Set only while the in-process FastMCP server is using stdio transport.
+# This is deliberately separate from ``sys.stdin.isatty()``: CI, cron, and
+# redirected CLI builds also have non-TTY stdin, but do not share the MCP
+# transport's file-descriptor lifetime problem.
+_MCP_STDIO_ACTIVE = False
+
 
 def _select_executor_kind() -> str:
     """Return 'process' or 'thread' for parallel parsing.
 
     Defaults to ``process`` (the original behavior, fastest on Linux/macOS).
-    Auto-switches to ``thread`` when running on Windows with stdin not
-    attached to a TTY — that combination indicates an MCP/stdio host, where
-    ``ProcessPoolExecutor`` workers inherit the parent's pipe handles and
-    leak as zombies after the pool closes (issues #46, #136).
+    Auto-switches to ``thread`` for an active MCP stdio server on every
+    platform, where ``ProcessPoolExecutor`` workers can inherit the transport
+    pipe/socket and prevent EOF shutdown. The older Windows non-TTY fallback
+    remains for direct integrations that predate the explicit transport flag
+    (issues #46, #136, PR #615).
 
     Override explicitly with ``CRG_PARSE_EXECUTOR={process,thread}``.
 
@@ -44,6 +51,8 @@ def _select_executor_kind() -> str:
     explicit = os.environ.get("CRG_PARSE_EXECUTOR", "").strip().lower()
     if explicit in ("process", "thread"):
         return explicit
+    if _MCP_STDIO_ACTIVE:
+        return "thread"
     if sys.platform == "win32" and not sys.stdin.isatty():
         return "thread"
     return "process"
@@ -915,9 +924,10 @@ def full_build(
                 logger.info("Progress: %d/%d files parsed", i, file_count)
     else:
         # Parallel parsing — store calls remain serial (SQLite single-writer).
-        # Executor kind auto-selected: process on Linux/macOS/Windows-TTY,
-        # thread on Windows-MCP-stdio to avoid pipe-handle inheritance
-        # deadlock (issues #46, #136). Override via CRG_PARSE_EXECUTOR env.
+        # Executor kind auto-selected: process for normal CLI/automation;
+        # thread for MCP stdio to avoid pipe-handle inheritance deadlocks and
+        # orphan workers (issues #46, #136, PR #615). Override via
+        # CRG_PARSE_EXECUTOR env.
         args_list = [(rel_path, str(repo_root)) for rel_path in files]
         with _make_executor(_MAX_PARSE_WORKERS) as executor:
             for i, (rel_path, nodes, edges, error, fhash) in enumerate(

code_review_graph/main.py

@@ -17,6 +17,7 @@
 
 from fastmcp import FastMCP
 
+from . import incremental as _incremental
 from .graph import GraphStore
 from .incremental import find_project_root, get_db_path, start_watch_thread
 from .prompts import (
@@ -1077,26 +1078,29 @@ def main(
     _default_repo_root = str(root)
     _apply_tool_filter(tools)
 
+    previous_stdio_state = _incremental._MCP_STDIO_ACTIVE
+    _incremental._MCP_STDIO_ACTIVE = transport == "stdio"
     watch_store: GraphStore | None = None
-    if auto_watch:
-        watch_store = GraphStore(get_db_path(root))
-        thread = start_watch_thread(root, watch_store, daemon=True)
-        if thread is None:
-            logger.warning("Auto-watch was requested but could not be started")
-
-    if sys.platform == "win32":
-        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
-        # Pre-warm sentence-transformers on the main thread before fastmcp's
-        # event loop starts. Lazy-loading ``torch`` + tokenizers inside an
-        # executor worker thread deadlocks ``semantic_search_nodes_tool`` on
-        # Windows stdio MCP (DLL init / OpenMP thread-pool registration grabs
-        # locks the loop needs). #385 added ``asyncio.to_thread`` to peer
-        # tools but cannot fix this case — the dangerous initialization has
-        # to happen on the main thread before any worker thread is spawned.
-        from .embeddings import prewarm_local_embeddings
-        prewarm_local_embeddings()
-
     try:
+        if auto_watch:
+            watch_store = GraphStore(get_db_path(root))
+            thread = start_watch_thread(root, watch_store, daemon=True)
+            if thread is None:
+                logger.warning("Auto-watch was requested but could not be started")
+
+        if sys.platform == "win32":
+            asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
+            # Pre-warm sentence-transformers on the main thread before fastmcp's
+            # event loop starts. Lazy-loading ``torch`` + tokenizers inside an
+            # executor worker thread deadlocks ``semantic_search_nodes_tool`` on
+            # Windows stdio MCP (DLL init / OpenMP thread-pool registration grabs
+            # locks the loop needs). #385 added ``asyncio.to_thread`` to peer
+            # tools but cannot fix this case — the dangerous initialization has
+            # to happen on the main thread before any worker thread is spawned.
+            from .embeddings import prewarm_local_embeddings
+
+            prewarm_local_embeddings()
+
         if transport == "stdio":
             # Stdio MCP must keep stdout strictly JSON-RPC. FastMCP's banner/update
             # notices corrupt the handshake stream on clients like Codex CLI.
@@ -1110,6 +1114,7 @@ def main(
     finally:
         if watch_store is not None:
             watch_store.close()
+        _incremental._MCP_STDIO_ACTIVE = previous_stdio_state
 
 
 if __name__ == "__main__":

tests/test_incremental.py

@@ -3,6 +3,7 @@
 import subprocess
 from unittest.mock import MagicMock, patch  # noqa: F401 – patch used in tests
 
+import code_review_graph.incremental as incremental_module
 from code_review_graph.graph import GraphStore
 from code_review_graph.incremental import (
     _is_binary,
@@ -24,6 +25,36 @@
 )
 
 
+class TestParseExecutorSelection:
+    def test_stdio_mcp_uses_threads_on_unix(self, monkeypatch):
+        monkeypatch.delenv("CRG_PARSE_EXECUTOR", raising=False)
+        monkeypatch.setattr(
+            incremental_module, "_MCP_STDIO_ACTIVE", True, raising=False,
+        )
+        monkeypatch.setattr(incremental_module.sys, "platform", "linux")
+        monkeypatch.setattr(incremental_module.sys.stdin, "isatty", lambda: False)
+
+        assert incremental_module._select_executor_kind() == "thread"
+
+    def test_non_mcp_unix_automation_keeps_process_default(self, monkeypatch):
+        monkeypatch.delenv("CRG_PARSE_EXECUTOR", raising=False)
+        monkeypatch.setattr(
+            incremental_module, "_MCP_STDIO_ACTIVE", False, raising=False,
+        )
+        monkeypatch.setattr(incremental_module.sys, "platform", "linux")
+        monkeypatch.setattr(incremental_module.sys.stdin, "isatty", lambda: False)
+
+        assert incremental_module._select_executor_kind() == "process"
+
+    def test_explicit_process_override_wins_in_stdio_mcp(self, monkeypatch):
+        monkeypatch.setenv("CRG_PARSE_EXECUTOR", "process")
+        monkeypatch.setattr(
+            incremental_module, "_MCP_STDIO_ACTIVE", True, raising=False,
+        )
+
+        assert incremental_module._select_executor_kind() == "process"
+
+
 class TestFindRepoRoot:
     def test_finds_git_dir(self, tmp_path):
         (tmp_path / ".git").mkdir()

tests/test_main.py

@@ -14,6 +14,7 @@
 
 import pytest
 
+import code_review_graph.incremental as incremental_module
 import code_review_graph.tools.docs as docs_module
 from code_review_graph import main as crg_main
 
@@ -95,18 +96,29 @@ def test_stdio_calls_mcp_run_stdio(self, monkeypatch):
         calls: list[dict] = []
 
         def fake_run(**kwargs):
+            assert incremental_module._MCP_STDIO_ACTIVE is True
+            assert incremental_module._select_executor_kind() == "thread"
             calls.append(kwargs)
 
+        monkeypatch.delenv("CRG_PARSE_EXECUTOR", raising=False)
+        monkeypatch.setattr(
+            incremental_module, "_MCP_STDIO_ACTIVE", False, raising=False,
+        )
         monkeypatch.setattr(crg_main.mcp, "run", fake_run)
         crg_main.main(repo_root=None)
         assert calls == [{"transport": "stdio", "show_banner": False}]
+        assert incremental_module._MCP_STDIO_ACTIVE is False
 
     def test_http_calls_mcp_run_with_host_port(self, monkeypatch):
         calls: list[dict] = []
 
         def fake_run(**kwargs):
+            assert incremental_module._MCP_STDIO_ACTIVE is False
             calls.append(kwargs)
 
+        monkeypatch.setattr(
+            incremental_module, "_MCP_STDIO_ACTIVE", False, raising=False,
+        )
         monkeypatch.setattr(crg_main.mcp, "run", fake_run)
         crg_main.main(
             repo_root="/tmp/r",

tests/test_mcp_stdio_shutdown.py

@@ -0,0 +1,140 @@
+"""End-to-end regression for MCP stdio executor shutdown (PR #615)."""
+
+from __future__ import annotations
+
+import json
+import os
+import select
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import pytest
+
+
+def _send(proc: subprocess.Popen[str], message: dict) -> None:
+    assert proc.stdin is not None
+    proc.stdin.write(json.dumps(message) + "\n")
+    proc.stdin.flush()
+
+
+def _read_response(
+    proc: subprocess.Popen[str],
+    request_id: int,
+    timeout: float = 20,
+) -> dict:
+    assert proc.stdout is not None
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        ready, _, _ = select.select(
+            [proc.stdout],
+            [],
+            [],
+            max(0, deadline - time.monotonic()),
+        )
+        if not ready:
+            break
+        line = proc.stdout.readline()
+        if not line:
+            break
+        response = json.loads(line)
+        if response.get("id") == request_id:
+            return response
+    raise AssertionError(f"MCP response {request_id} did not arrive within {timeout}s")
+
+
+@pytest.mark.skipif(os.name == "nt", reason="select() cannot poll Windows pipes")
+def test_stdio_server_parallel_build_then_eof_exits_cleanly(tmp_path):
+    """The real stdio server must build in parallel and exit cleanly on EOF."""
+    (tmp_path / ".git").mkdir()
+    for index in range(10):
+        (tmp_path / f"module_{index}.py").write_text(
+            f"def function_{index}():\n    return {index}\n",
+            encoding="utf-8",
+        )
+
+    env = os.environ.copy()
+    env.pop("CRG_PARSE_EXECUTOR", None)
+    env.pop("CRG_SERIAL_PARSE", None)
+    env.pop("CRG_TOOLS", None)
+    env.pop("CRG_DATA_DIR", None)
+    env.pop("CRG_REPO_ROOT", None)
+    env["CRG_PARSE_WORKERS"] = "2"
+    repo_root = str(Path(__file__).resolve().parents[1])
+    env["PYTHONPATH"] = os.pathsep.join(
+        value for value in (repo_root, env.get("PYTHONPATH")) if value
+    )
+
+    proc = subprocess.Popen(
+        [
+            sys.executable,
+            "-m",
+            "code_review_graph",
+            "serve",
+            "--repo",
+            str(tmp_path),
+        ],
+        cwd=tmp_path,
+        env=env,
+        stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+    )
+    try:
+        _send(
+            proc,
+            {
+                "jsonrpc": "2.0",
+                "id": 1,
+                "method": "initialize",
+                "params": {
+                    "capabilities": {},
+                    "clientInfo": {"name": "shutdown-test", "version": "1"},
+                    "protocolVersion": "2024-11-05",
+                },
+            },
+        )
+        assert "result" in _read_response(proc, 1)
+        _send(
+            proc,
+            {
+                "jsonrpc": "2.0",
+                "method": "notifications/initialized",
+                "params": {},
+            },
+        )
+        _send(
+            proc,
+            {
+                "jsonrpc": "2.0",
+                "id": 2,
+                "method": "tools/call",
+                "params": {
+                    "name": "build_or_update_graph_tool",
+                    "arguments": {
+                        "repo_root": str(tmp_path),
+                        "full_rebuild": True,
+                        "postprocess": "none",
+                    },
+                },
+            },
+        )
+        build_response = _read_response(proc, 2)
+        assert "error" not in build_response
+        build_payload = json.loads(build_response["result"]["content"][0]["text"])
+        assert build_payload["status"] == "ok"
+        assert build_payload["build_type"] == "full"
+        assert build_payload["files_parsed"] == 10
+        assert (tmp_path / ".code-review-graph" / "graph.db").is_file()
+
+        assert proc.stdin is not None
+        proc.stdin.close()
+        proc.wait(timeout=10)
+        stderr = proc.stderr.read() if proc.stderr is not None else ""
+        assert proc.returncode == 0, stderr
+    finally:
+        if proc.poll() is None:
+            proc.kill()
+            proc.wait(timeout=3)