fix(mcp): honor subprocess PATHEXT for stdio resolution

tests/tools/test_mcp_stdio_resolution.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+import os
+from unittest.mock import patch
+
+from tools import mcp_tool
+
+
+def test_which_with_subprocess_env_uses_subprocess_pathext_on_windows(monkeypatch):
+    seen = {}
+    monkeypatch.setattr(mcp_tool.os, "name", "nt")
+    monkeypatch.setenv("PATHEXT", ".EXE")
+
+    def fake_which(command, path=None):
+        seen["command"] = command
+        seen["path"] = path
+        seen["pathext"] = os.environ.get("PATHEXT")
+        return "C:/tools/server.CMD"
+
+    with patch("tools.mcp_tool.shutil.which", side_effect=fake_which):
+        resolved = mcp_tool._which_with_subprocess_env(
+            "server",
+            path="C:/tools",
+            env={"PATH": "C:/tools", "PATHEXT": ".CMD"},
+        )
+
+    assert resolved == "C:/tools/server.CMD"
+    assert seen == {"command": "server", "path": "C:/tools", "pathext": ".CMD"}
+    assert os.environ["PATHEXT"] == ".EXE"
+
+
+def test_resolve_stdio_command_passes_subprocess_path_and_pathext(monkeypatch):
+    monkeypatch.setattr(mcp_tool.os, "name", "nt")
+
+    def fake_which(command, path=None):
+        assert command == "server"
+        assert path == "C:/mcp/bin"
+        assert os.environ.get("PATHEXT") == ".BAT"
+        return "C:/mcp/bin/server.BAT"
+
+    with patch("tools.mcp_tool.shutil.which", side_effect=fake_which):
+        command, env = mcp_tool._resolve_stdio_command(
+            "server",
+            {"PATH": "C:/mcp/bin", "PATHEXT": ".BAT"},
+        )
+
+    assert command == "C:/mcp/bin/server.BAT"
+    assert env["PATH"].startswith("C:/mcp/bin")
+    assert env["PATHEXT"] == ".BAT"

tools/mcp_tool.py

@@ -515,6 +515,22 @@ def _prepend_path(env: dict, directory: str) -> dict:
     return updated
 
 
+def _which_with_subprocess_env(command: str, *, path: str | None, env: dict) -> str | None:
+    """Resolve *command* using PATH/PATHEXT from the intended subprocess env."""
+    if os.name != "nt" or "PATHEXT" not in env:
+        return shutil.which(command, path=path)
+
+    previous = os.environ.get("PATHEXT")
+    try:
+        os.environ["PATHEXT"] = str(env.get("PATHEXT") or "")
+        return shutil.which(command, path=path)
+    finally:
+        if previous is None:
+            os.environ.pop("PATHEXT", None)
+        else:
+            os.environ["PATHEXT"] = previous
+
+
 def _resolve_stdio_command(command: str, env: dict) -> tuple[str, dict]:
     """Resolve a stdio MCP command against the exact subprocess environment.
 
@@ -526,7 +542,11 @@ def _resolve_stdio_command(command: str, env: dict) -> tuple[str, dict]:
 
     if os.sep not in resolved_command:
         path_arg = resolved_env["PATH"] if "PATH" in resolved_env else None
-        which_hit = shutil.which(resolved_command, path=path_arg)
+        which_hit = _which_with_subprocess_env(
+            resolved_command,
+            path=path_arg,
+            env=resolved_env,
+        )
         if which_hit:
             resolved_command = which_hit
         elif resolved_command in {"npx", "npm", "node"}:
@@ -1867,7 +1887,15 @@ async def _run_stdio(self, config: dict):
                 write_stream,
             ):
                 # Capture the newly spawned subprocess PID for force-kill cleanup.
-                new_pids = _snapshot_child_pids() - pids_before
+                # Filter out non-MCP children that race into the snapshot window:
+                # slash_worker and LSP servers (jdtls/pyright/yaml-ls) are spawned
+                # directly by the gateway without start_new_session, so their pgid
+                # equals the TUI parent PID. If they leak into _stdio_pgids, the
+                # shutdown sweep's killpg() kills the TUI parent itself.
+                # See agent/lsp/client.py for the complementary start_new_session fix.
+                new_pids = _filter_mcp_children(
+                    _snapshot_child_pids() - pids_before
+                )
                 if new_pids:
                     # Capture pgid while the child is alive — once it exits we
                     # can no longer call ``os.getpgid`` on it, and the cleanup
@@ -3005,6 +3033,56 @@ def _snapshot_child_pids() -> set:
     return set()
 
 
+# Non-MCP gateway children that can race into the _snapshot_child_pids() delta
+# during stdio MCP server spawn. LSP servers and slash_worker now use
+# start_new_session=True too; this remains defense-in-depth for any future
+# non-MCP child spawn that briefly appears in the MCP snapshot delta. Match
+# argv markers instead of argv[0] because Python/Java children begin with the
+# interpreter or binary path.
+_NON_MCP_CHILD_CMDLINE_MARKERS: tuple[str, ...] = (
+    "tui_gateway.slash_worker",
+    "tui_gateway.entry",
+    "-dorg.eclipse.equinox.launcher",  # jdtls (legacy arg style)
+    "eclipse.jdt.ls",
+    "org.eclipse.equinox.launcher_",
+)
+
+
+def _filter_mcp_children(pids: set) -> set:
+    """Remove non-MCP children from a PID snapshot delta.
+
+    _snapshot_child_pids() returns *all* direct children of the gateway. When
+    a stdio MCP server spawns concurrently with a slash_worker or LSP server
+    spawn, the delta ``_snapshot_child_pids() - pids_before`` can include
+    PIDs that are NOT the MCP server. Tracking those PIDs in _stdio_pgids is
+    catastrophic if a future child lacks start_new_session: its pgid can be the
+    TUI parent's PID, so the shutdown sweep's killpg() kills the TUI itself.
+    """
+    if not pids:
+        return pids
+    try:
+        import psutil
+    except ImportError:
+        # psutil unavailable — keep all PIDs (preserves prior behavior).
+        return pids
+    filtered: set = set()
+    for pid in pids:
+        try:
+            argv = psutil.Process(pid).cmdline()
+        except (psutil.NoSuchProcess, psutil.AccessDenied, OSError):
+            # Process raced away or is a zombie — skip it; it cannot be the
+            # MCP server we just spawned and is not safe to track.
+            continue
+        if any(
+            marker in arg
+            for arg in argv[1:]
+            for marker in _NON_MCP_CHILD_CMDLINE_MARKERS
+        ):
+            continue
+        filtered.add(pid)
+    return filtered
+
+
 def _mcp_loop_exception_handler(loop, context):
     """Suppress benign 'Event loop is closed' noise during shutdown.
 
@@ -3731,11 +3809,43 @@ def _normalize_mcp_input_schema(schema: dict | None) -> dict:
         return {"type": "object", "properties": {}}
 
     def _rewrite_local_refs(node):
+        """Walk the schema, promoting legacy ``definitions`` to ``$defs``.
+
+        The promotion is contextual: ``definitions`` is renamed only when it
+        appears as a JSON Schema *meta-keyword* (sibling of ``properties`` /
+        ``$ref`` at a schema node), never when it appears as the *name of a
+        property* (i.e., as a key inside a ``properties`` dict).
+
+        Without this gate, MCP servers that legitimately expose a tool
+        parameter named ``definitions`` (e.g. a CI/pipelines tool that uses
+        ``definitions`` for an array of pipeline-definition IDs) would have
+        that user-facing property name silently rewritten to ``$defs``.
+        Anthropic and OpenAI both reject ``$`` in property names
+        (``^[a-zA-Z0-9_.-]{1,64}$``), so the whole tool array gets a 400 and
+        every conversation breaks.
+
+        The gate works by treating ``properties`` and ``patternProperties``
+        specially during descent: we iterate the property-name -> schema map
+        directly, leaving the property names verbatim, then recurse into each
+        property's schema where ordinary JSON Schema semantics resume (so any
+        legitimately-nested ``definitions`` meta-keyword inside a property's
+        schema is still promoted).
+        """
         if isinstance(node, dict):
             normalized = {}
             for key, value in node.items():
-                out_key = "$defs" if key == "definitions" else key
-                normalized[out_key] = _rewrite_local_refs(value)
+                if key in ("properties", "patternProperties") and isinstance(value, dict):
+                    # Keys of this dict are user-facing property names, not
+                    # meta-keywords. Preserve them verbatim; recurse only into
+                    # each property's schema, where ``definitions`` again has
+                    # its JSON Schema meaning.
+                    normalized[key] = {
+                        prop_name: _rewrite_local_refs(prop_schema)
+                        for prop_name, prop_schema in value.items()
+                    }
+                else:
+                    out_key = "$defs" if key == "definitions" else key
+                    normalized[out_key] = _rewrite_local_refs(value)
             ref = normalized.get("$ref")
             if isinstance(ref, str) and ref.startswith("#/definitions/"):
                 normalized["$ref"] = "#/$defs/" + ref[len("#/definitions/"):]