fix(mcp): use direct parent identity in stdio watchdog

Use the direct POSIX parent relationship instead of process creation time and pid_exists checks. Remove the dead create-time argument chain while preserving process-group cleanup and signal forwarding.\n\nRefs #62505

tests/tools/test_mcp_stdio_watchdog.py

@@ -0,0 +1,47 @@
+"""Contract tests for the direct POSIX stdio MCP child watchdog."""
+
+import os
+import sys
+
+import pytest
+
+from tools import mcp_stdio_watchdog, mcp_tool
+
+
+def test_is_orphaned_is_false_while_direct_parent_is_unchanged():
+    original_ppid = 1234
+
+    assert mcp_stdio_watchdog._is_orphaned(
+        original_ppid,
+        getppid=lambda: original_ppid,
+    ) is False
+
+
+def test_is_orphaned_is_true_after_direct_parent_changes():
+    assert mcp_stdio_watchdog._is_orphaned(
+        1234,
+        getppid=lambda: 5678,
+    ) is True
+
+
+@pytest.mark.skipif(os.name != "posix", reason="watchdog wrapping is POSIX-only")
+def test_wrap_command_uses_stable_parent_pid_and_preserves_command_tail():
+    parent_pid = os.getpid()
+    command = "/opt/hermes/bin/mcp-server"
+    command_args = ["--label", "value with spaces", "--", "literal-tail"]
+
+    wrapped_command, wrapped_args = mcp_tool._wrap_command_with_watchdog(
+        command,
+        command_args,
+    )
+
+    assert wrapped_command == sys.executable
+    assert wrapped_args == [
+        os.path.join(os.path.dirname(mcp_tool.__file__), "mcp_stdio_watchdog.py"),
+        "--ppid",
+        str(parent_pid),
+        "--",
+        command,
+        *command_args,
+    ]
+    assert "--create-time" not in wrapped_args

tools/mcp_stdio_watchdog.py

@@ -25,23 +25,19 @@ instead, which:
   2. transparently passes stdin/stdout/stderr through — the MCP stdio
      protocol talks directly over those pipes, so the supervisor must be a
      no-op relay, not a bytes-in-the-middle proxy;
-  3. runs a background thread that polls the ORIGINAL parent PID using the
-     exact same orphan-detection algorithm already proven in
-     ``tui_gateway/slash_worker.py`` (``_is_orphaned``): compare current
-     ``getppid()`` against the recorded original, and guard PID reuse via
-     ``psutil`` process creation time;
+  3. runs a background thread that polls the direct POSIX parent identity:
+     compare current ``getppid()`` against the parent PID recorded when the
+     wrapper was created;
   4. the instant the original parent is gone, terminates the real child's
      process group (SIGTERM, grace period, then SIGKILL) and exits.
 
-This is intentionally a thin, dependency-light script (``psutil`` only,
-already a hard dependency via ``tui_gateway/slash_worker.py``) so it starts
-fast and can't itself become a resource leak.
+This is intentionally a thin, standard-library-only script so it starts fast
+and can't itself become a resource leak.
 
 Usage (see ``tools/mcp_tool.py::_run_stdio``)::
 
     python3 -m tools.mcp_stdio_watchdog \\
-        --ppid <original_parent_pid> --create-time <original_parent_create_time> \\
-        -- <real_command> <arg1> <arg2> ...
+        --ppid <original_parent_pid> -- <real_command> <arg1> <arg2> ...
 """
 
 from __future__ import annotations
@@ -54,35 +50,13 @@ import sys
 import threading
 import time
 
-try:
-    import psutil
-except ImportError:  # pragma: no cover - psutil is a hard dependency elsewhere
-    psutil = None
-
 _POLL_INTERVAL_S = 2.0
 _TERM_GRACE_S = 3.0
 
 
-def _is_orphaned(original_ppid: int, parent_create_time: float, getppid=os.getppid) -> bool:
-    """Mirrors ``tui_gateway.slash_worker._is_orphaned`` exactly.
-
-    True once the process that spawned us is gone. Never trusts a bare
-    ``getppid() == 1`` check (Linux reparents orphans to a subreaper, not
-    always PID 1), and guards against PID reuse via the recorded creation
-    time of the original parent.
-    """
-    if getppid() != original_ppid:
-        return True
-    if psutil is None:
-        # No reliable staleness check available; fall back to the ppid
-        # comparison alone (still catches the common case).
-        return False
-    try:
-        if not psutil.pid_exists(original_ppid):
-            return True
-        return psutil.Process(original_ppid).create_time() != parent_create_time
-    except psutil.Error:
-        return True
+def _is_orphaned(original_ppid: int, getppid=os.getppid) -> bool:
+    """Return whether this process no longer has its original POSIX parent."""
+    return getppid() != original_ppid
 
 
 def _terminate_process_group(proc: subprocess.Popen) -> None:
@@ -118,9 +92,9 @@ def _terminate_process_group(proc: subprocess.Popen) -> None:
             continue
 
 
-def _watchdog_loop(proc: subprocess.Popen, original_ppid: int, parent_create_time: float) -> None:
+def _watchdog_loop(proc: subprocess.Popen, original_ppid: int) -> None:
     while proc.poll() is None:
-        if _is_orphaned(original_ppid, parent_create_time):
+        if _is_orphaned(original_ppid):
             _terminate_process_group(proc)
             return
         time.sleep(_POLL_INTERVAL_S)
@@ -131,7 +105,6 @@ def main(argv: list[str] | None = None) -> int:
         description="Parent-death watchdog for a stdio MCP subprocess.",
     )
     parser.add_argument("--ppid", type=int, required=True)
-    parser.add_argument("--create-time", type=float, required=True)
     parser.add_argument("command", nargs=argparse.REMAINDER)
     args = parser.parse_args(argv)
 
@@ -168,7 +141,7 @@ def main(argv: list[str] | None = None) -> int:
 
     watchdog = threading.Thread(
         target=_watchdog_loop,
-        args=(proc, args.ppid, args.create_time),
+        args=(proc, args.ppid),
         daemon=True,
     )
     watchdog.start()

tools/mcp_tool.py

@@ -672,10 +672,9 @@ def _resolve_stdio_command(command: str, env: dict) -> tuple[str, dict]:
 def _wrap_command_with_watchdog(command: str, args: list) -> tuple[str, list]:
     """Wrap a stdio MCP server command in the parent-death watchdog supervisor.
 
-    See ``tools/mcp_stdio_watchdog.py`` module docstring for the full
-    rationale. Returns the (command, args) unchanged on any platform/failure
-    where the wrap can't safely apply, so this can never be the reason a
-    previously-working MCP server stops starting.
+    On POSIX, the watchdog records this process's PID and later detects parent
+    death directly through ``getppid()``. Returns the (command, args) unchanged
+    on non-POSIX platforms or if the PID cannot be read.
     """
     if os.name != "posix":
         # Relies on process groups (os.getpgid/os.killpg); no POSIX
@@ -685,18 +684,12 @@ def _wrap_command_with_watchdog(command: str, args: list) -> tuple[str, list]:
         return command, args
     try:
         my_pid = os.getpid()
-        try:
-            import psutil
-            create_time = psutil.Process(my_pid).create_time()
-        except ImportError:
-            create_time = time.time()
     except Exception:
         # Never let watchdog bookkeeping failure block a real MCP connection.
         return command, args
     watchdog_args = [
         os.path.join(os.path.dirname(os.path.abspath(__file__)), "mcp_stdio_watchdog.py"),
         "--ppid", str(my_pid),
-        "--create-time", repr(create_time),
         "--",
         command,
         *args,