fix(acp): shield the JSON-RPC stdin from every child process

acp_adapter/entry.py

@@ -217,6 +217,88 @@ def _run_setup_browser(assume_yes: bool = False) -> int:
         return 1
 
 
+def _point_win32_stdin_at(fd: int) -> bool:
+    """Repoint the process-wide Win32 ``STD_INPUT_HANDLE`` at *fd*.
+
+    This is the handle a child launched with ``stdin=None`` actually inherits
+    on Windows — fd 0 alone does not shield anything there — so its success is
+    load-bearing rather than advisory.
+
+    Returns True when there is nothing to do (non-Windows) or the redirect
+    succeeded, False when the Win32 call reported failure.
+    """
+    if sys.platform != "win32":
+        return True
+
+    import ctypes
+    import msvcrt
+    from ctypes import wintypes
+
+    STD_INPUT_HANDLE = -10
+
+    kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
+    # Without explicit argtypes ctypes passes the HANDLE as a C int, which
+    # truncates on 64-bit Windows.
+    kernel32.SetStdHandle.argtypes = [wintypes.DWORD, wintypes.HANDLE]
+    kernel32.SetStdHandle.restype = wintypes.BOOL
+
+    if not kernel32.SetStdHandle(STD_INPUT_HANDLE, msvcrt.get_osfhandle(fd)):
+        logging.getLogger(__name__).warning(
+            "SetStdHandle(STD_INPUT_HANDLE) failed (WinError %d); "
+            "child processes would still inherit the ACP stdin pipe",
+            ctypes.get_last_error(),
+        )
+        return False
+    return True
+
+
+def _shield_stdin_from_children() -> None:
+    """Keep the ACP JSON-RPC stdin out of every child process.
+
+    The host IDE hands this process a pipe as stdin. Any subprocess spawned
+    with ``stdin=None`` inherits that handle — on Windows via the process
+    STD_INPUT_HANDLE, not fd 0 — and MSYS/mingw programs (git, bash) can
+    block on it at startup. When they do, ``subprocess.run(timeout=...)``
+    kills only the ``Git\\cmd``/``Git\\bin`` wrapper; the surviving mingw
+    grandchild keeps the pipe open and the post-kill ``communicate()`` drain
+    (which has no timeout) wedges the calling thread — and with it the whole
+    agent turn. Re-home the transport's stdin onto a private duplicate and
+    point fd 0 plus the Win32 std input handle at the null device so every
+    child inherits NUL instead.
+    """
+    import io
+
+    logger = logging.getLogger(__name__)
+    try:
+        original_fd = sys.stdin.fileno()
+    except (AttributeError, OSError, ValueError):
+        return  # no real stdin (embedded/test harness) — nothing to shield
+    try:
+        private_fd = os.dup(original_fd)
+        os.set_inheritable(private_fd, False)
+        devnull_fd = os.open(os.devnull, os.O_RDONLY)
+        os.dup2(devnull_fd, original_fd)
+        os.close(devnull_fd)
+        if not _point_win32_stdin_at(original_fd):
+            # Half-shielded is the worst outcome: fd 0 reads NUL, but children
+            # still inherit the real pipe through STD_INPUT_HANDLE, and the
+            # log would claim the stdin was shielded. Put fd 0 back so the
+            # process is left in the state it started in, unshielded and
+            # consistent, and let the caller proceed without the shield.
+            os.dup2(private_fd, original_fd)
+            os.close(private_fd)
+            return
+        # The ACP transport reads sys.stdin.buffer; hand it the private copy.
+        sys.stdin = io.TextIOWrapper(
+            io.BufferedReader(io.FileIO(private_fd, "rb")),
+            encoding="utf-8",
+            errors="replace",
+        )
+        logger.info("ACP stdin shielded from child processes (children inherit NUL)")
+    except Exception:
+        logger.warning("Could not shield ACP stdin from child processes", exc_info=True)
+
+
 def main(argv: list[str] | None = None) -> None:
     """Entry point: load env, configure logging, run the ACP agent."""
     args = _parse_args(argv)
@@ -237,6 +319,7 @@ def main(argv: list[str] | None = None) -> None:
 
     _setup_logging()
     _load_env()
+    _shield_stdin_from_children()
 
     logger = logging.getLogger(__name__)
     logger.info("Starting hermes-agent ACP adapter")

tests/acp/test_entry.py

@@ -1,12 +1,30 @@
 """Tests for acp_adapter.entry startup wiring."""
 
+import os
 import sys
+from pathlib import Path
 
 import acp
 import pytest
 
 from acp_adapter import entry
 
+_REPO_ROOT = Path(__file__).resolve().parents[2]
+
+
+def _env_with_repo_first():
+    """Environment for probe children with this tree first on the import path.
+
+    Without it a child resolves ``acp_adapter`` against whatever the venv has
+    installed — a worktree run against an install venv silently probes the
+    wrong code and dies with ``AttributeError`` on the new symbols.
+    """
+    env = dict(os.environ)
+    env["PYTHONPATH"] = os.pathsep.join(
+        p for p in (str(_REPO_ROOT), env.get("PYTHONPATH")) if p
+    )
+    return env
+
 
 def test_main_enables_unstable_protocol(monkeypatch):
     calls = {}
@@ -88,3 +106,131 @@ def fake_ensure(dep, interactive=True):
     with pytest.raises(SystemExit) as excinfo:
         entry.main(["--setup-browser"])
     assert excinfo.value.code == 1
+
+
+def test_shield_stdin_redirects_fd0_to_devnull(tmp_path):
+    """Children must inherit NUL, not the ACP JSON-RPC pipe (#73693).
+
+    Runs in a subprocess so the fd surgery cannot disturb the test session's
+    own stdin. The child reports what a grandchild would inherit on fd 0,
+    plus whether the transport can still read the original stream.
+    """
+    import subprocess
+    import sys as _sys
+
+    payload = tmp_path / "probe.py"
+    payload.write_text(
+        "import os, sys\n"
+        "from acp_adapter import entry\n"
+        "entry._shield_stdin_from_children()\n"
+        # What a child would inherit on fd 0:
+        "inherited = os.read(0, 16)\n"
+        # What the ACP transport still sees on the re-homed sys.stdin:
+        "transport = sys.stdin.buffer.readline()\n"
+        "print('INHERITED:' + repr(inherited))\n"
+        "print('TRANSPORT:' + repr(transport))\n",
+        encoding="utf-8",
+    )
+
+    result = subprocess.run(
+        [_sys.executable, str(payload)],
+        input=b"protocol-line\n",
+        capture_output=True,
+        timeout=60,
+        env=_env_with_repo_first(),
+    )
+
+    out = result.stdout.decode("utf-8", "replace")
+    assert result.returncode == 0, result.stderr.decode("utf-8", "replace")
+    # fd 0 now reads EOF (NUL), so nothing a child spawns can consume — or
+    # block on — the protocol stream.
+    assert "INHERITED:b''" in out
+    # The transport keeps the real stdin through the private duplicate.
+    assert r"TRANSPORT:b'protocol-line\n'" in out
+
+
+@pytest.mark.skipif(
+    sys.platform != "win32",
+    reason="STD_INPUT_HANDLE inheritance is Windows-specific",
+)
+def test_shield_stdin_denies_the_pipe_to_a_real_descendant(tmp_path):
+    """A spawned descendant must inherit NUL, not the JSON-RPC pipe.
+
+    The fd-0 probe above cannot prove this on Windows: a child launched with
+    ``stdin=None`` inherits the process-wide ``STD_INPUT_HANDLE``, not fd 0,
+    so only an actual descendant exercises the guarantee this shield makes.
+    """
+    import subprocess
+    import sys as _sys
+
+    payload = tmp_path / "descendant_probe.py"
+    payload.write_text(
+        "import subprocess, sys\n"
+        "from acp_adapter import entry\n"
+        "entry._shield_stdin_from_children()\n"
+        # Launch a real grandchild with stdin=None so it inherits whatever
+        # this process hands down — fd 0 on POSIX, STD_INPUT_HANDLE on Windows.
+        "child = subprocess.run(\n"
+        "    [sys.executable, '-c',\n"
+        "     'import os,sys; sys.stdout.write(repr(os.read(0, 16)))'],\n"
+        "    capture_output=True, timeout=30,\n"
+        ")\n"
+        "print('CHILD:' + child.stdout.decode('utf-8', 'replace'))\n"
+        # The transport must still own the real stream afterwards.
+        "print('TRANSPORT:' + repr(sys.stdin.buffer.readline()))\n",
+        encoding="utf-8",
+    )
+
+    result = subprocess.run(
+        [_sys.executable, str(payload)],
+        input=b"protocol-line\n",
+        capture_output=True,
+        timeout=60,
+        env=_env_with_repo_first(),
+    )
+
+    out = result.stdout.decode("utf-8", "replace")
+    assert result.returncode == 0, result.stderr.decode("utf-8", "replace")
+    # The descendant saw EOF: it cannot consume or block on the protocol pipe.
+    assert "CHILD:b''" in out
+    assert r"TRANSPORT:b'protocol-line\n'" in out
+
+
+def test_shield_stdin_rolls_back_when_the_win32_handle_redirect_fails(tmp_path):
+    """A failed ``SetStdHandle`` must not leave a half-shielded process.
+
+    fd 0 is repointed at NUL *before* the Win32 std-handle redirect. If that
+    redirect fails and we carry on, children still inherit the real pipe via
+    ``STD_INPUT_HANDLE`` while the log claims the stdin was shielded. Restore
+    the original fd 0 instead, so the process is left in a known state.
+    """
+    import subprocess
+    import sys as _sys
+
+    payload = tmp_path / "failing_redirect_probe.py"
+    payload.write_text(
+        "import os, sys\n"
+        "from acp_adapter import entry\n"
+        # Simulate SetStdHandle returning FALSE, on any platform.
+        "entry._point_win32_stdin_at = lambda fd: False\n"
+        "entry._shield_stdin_from_children()\n"
+        # Rolled back: fd 0 is the original stream again, not NUL.
+        "print('FD0:' + repr(os.read(0, 32)))\n",
+        encoding="utf-8",
+    )
+
+    result = subprocess.run(
+        [_sys.executable, str(payload)],
+        input=b"protocol-line\n",
+        capture_output=True,
+        timeout=60,
+        env=_env_with_repo_first(),
+    )
+
+    out = result.stdout.decode("utf-8", "replace")
+    err = result.stderr.decode("utf-8", "replace")
+    assert result.returncode == 0, err
+    # Rollback restored the original stdin on fd 0 rather than leaving NUL.
+    assert r"FD0:b'protocol-line\n'" in out, out
+    # And the failure was reported rather than logged as a success.
+    assert "shielded" not in err.lower(), err