Fix native Windows managed-process capture and cleanup

.github/workflows/ci.yml

@@ -9,6 +9,24 @@ permissions:
   contents: read
 
 jobs:
+  windows-process-contract:
+    runs-on: windows-latest
+    timeout-minutes: 15
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.11", "3.12"]
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          persist-credentials: false
+      - uses: astral-sh/setup-uv@v6
+        with:
+          version: "0.11.2"
+      - run: uv sync --locked --python '${{ matrix.python-version }}'
+      - name: Verify native Windows pipe capture and process cleanup
+        run: uv run --locked pytest -q tests/tools/test_windows_process_runtime.py
+
   process-subagent-contract:
     runs-on: ${{ matrix.os }}
     timeout-minutes: 15

docs/docs/tools/processes.md

@@ -43,7 +43,18 @@ atomic, including concurrent reads from the same consumer.
 
 `process_stop` interrupts, terminates, or kills
 the owned process group, escalating when necessary. Processes use pipes, not a
-PTY. The local implementation requires POSIX shell and pipe support.
+PTY. POSIX hosts use Bash; Windows hosts use native `cmd.exe /d /s /c`.
+Use cmd syntax on Windows or invoke PowerShell explicitly. Windows capture
+checks pipe readiness with `PeekNamedPipe` and waits between scans (20 ms),
+including when both streams close before the process exits. It supports Python
+3.11 without `os.set_blocking` or socket-only selectors on subprocess pipes.
+
+Windows stop/timeout uses the system `taskkill /T /F` to target the local tree
+while its launching process remains alive. Interrupt first attempts CTRL_BREAK;
+headless hosts without a console fall back to tree termination. This is not a
+Windows Job Object: detached descendants or descendants whose launching process
+has already exited are not guaranteed to be recoverable. Stopping a local SSH
+client does not guarantee termination of remote work.
 
 Memory is bounded to 256,000 output characters and 10,000 chunks per process.
 Large reads report `truncated` and `history_lost`. The `log_path` points to captured

penguin/tools/process_runtime.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import errno
 import hashlib
 import json
 import logging
@@ -17,6 +18,7 @@
 from codecs import getincrementaldecoder
 from collections import OrderedDict, deque
 from dataclasses import dataclass, field
+from functools import partial
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
@@ -195,7 +197,11 @@ def start(
                 )
                 os.close(fd)
                 process = subprocess.Popen(
-                    ["bash", "-c", command],
+                    (
+                        f'"{os.environ.get("COMSPEC", "cmd.exe")}" /d /s /c "{command}"'
+                        if os.name == "nt"
+                        else ["bash", "-c", command]
+                    ),
                     cwd=resolved_cwd,
                     env=effective_env,
                     stdin=subprocess.PIPE,
@@ -204,6 +210,9 @@ def start(
                     text=True,
                     bufsize=1,
                     start_new_session=os.name == "posix",
+                    creationflags=(
+                        subprocess.CREATE_NEW_PROCESS_GROUP if os.name == "nt" else 0
+                    ),
                 )
             except OSError as exc:
                 if log_path:
@@ -462,15 +471,27 @@ def write_stdin(
                 if written == len(payload):
                     break
                 fd = record.process.stdin.fileno()
-                os.set_blocking(fd, False)
+                if os.name == "nt":
+                    from penguin.tools.windows_pipes import set_nonblocking
+
+                    set_nonblocking(fd)
+                else:
+                    os.set_blocking(fd, False)
                 try:
                     # Small chunks bound work between cancellation checks. All
                     # stdin writes use the raw descriptor; no text buffer is used.
                     written += os.write(fd, payload[written : written + 4096])
-                except BlockingIOError:
-                    with selectors.DefaultSelector() as selector:
-                        selector.register(fd, selectors.EVENT_WRITE)
-                        selector.select(timeout=min(remaining, 0.05))
+                except OSError as exc:
+                    if not isinstance(exc, BlockingIOError) and not (
+                        os.name == "nt" and exc.errno == errno.ENOSPC
+                    ):
+                        raise
+                    if os.name == "nt":
+                        time.sleep(min(remaining, 0.02))
+                    else:
+                        with selectors.DefaultSelector() as selector:
+                            selector.register(fd, selectors.EVENT_WRITE)
+                            selector.select(timeout=min(remaining, 0.05))
         except (OSError, ValueError) as exc:
             error = f"stdin_write_failed: {exc}"
         finally:
@@ -494,8 +515,40 @@ def write_stdin(
         return result
 
     def _signal(self, record: ManagedProcess, mode: str) -> None:
+        from penguin.system.tool_environment import build_tool_environment
+
         try:
-            if record.owns_group:
+            if os.name == "nt" and record.process.poll() is None:
+                if mode == "interrupt":
+                    try:
+                        record.process.send_signal(signal.CTRL_BREAK_EVENT)
+                        return
+                    except OSError:
+                        # Services/headless hosts may have no console. Use the
+                        # same bounded tree termination as other stop modes.
+                        if record.process.poll() is not None:
+                            return
+                # Killing only cmd.exe leaves its Python/build children alive.
+                result = subprocess.run(
+                    [
+                        str(
+                            Path(os.environ.get("SystemRoot", "C:\\Windows"))
+                            / "System32"
+                            / "taskkill.exe"
+                        ),
+                        "/PID",
+                        str(record.process.pid),
+                        "/T",
+                        "/F",
+                    ],
+                    capture_output=True,
+                    timeout=5,
+                    check=False,
+                    env=build_tool_environment(),
+                )
+                if result.returncode and record.process.poll() is None:
+                    raise OSError(f"taskkill failed: {result.stderr!r}")
+            elif record.owns_group:
                 os.killpg(
                     record.process.pid,
                     {"interrupt": signal.SIGINT, "kill": signal.SIGKILL}.get(
@@ -585,13 +638,20 @@ def _capture(self, record: ManagedProcess) -> None:
         }
         exit_deadline = None
         try:
+            if os.name == "nt":
+                from penguin.tools.windows_pipes import PipeSelector
+
+                selector_factory = partial(PipeSelector, record.stop_reader)
+            else:
+                selector_factory = selectors.DefaultSelector
             with (
-                selectors.DefaultSelector() as selector,
+                selector_factory() as selector,
                 open(record.log_path, "a", encoding="utf-8") as log,
             ):
                 for stream in decoders:
                     pipe = getattr(record.process, stream)
-                    os.set_blocking(pipe.fileno(), False)
+                    if os.name != "nt":
+                        os.set_blocking(pipe.fileno(), False)
                     selector.register(pipe, selectors.EVENT_READ, stream)
                 while not record.stop_reader.is_set():
                     if (
@@ -636,11 +696,14 @@ def _capture(self, record: ManagedProcess) -> None:
                                 record.last_output_event = time.monotonic()
                                 self._emit(record, "output")
                 record.streams_closed = not selector.get_map()
-        except (OSError, ValueError) as exc:
+        except (OSError, ValueError, subprocess.TimeoutExpired) as exc:
             logger.exception("Process capture failed: %s", record.process_id)
             record.reader_error = str(exc)
-            self._signal(record, "kill")
-            record.process.wait()
+            try:
+                self._signal(record, "kill")
+                record.process.wait(timeout=5)
+            except (OSError, subprocess.TimeoutExpired):
+                logger.exception("Unable to stop process after capture failure")
         finally:
             for stream in ("stdout", "stderr"):
                 pipe = getattr(record.process, stream)
@@ -661,7 +724,12 @@ def enable_notifications(self, process_id: str, *, enabled: bool = True) -> None
             return
         with record.changed:
             record.notify_on_complete = enabled
-            if not enabled or not record.reader_done or record.notified:
+            if (
+                not enabled
+                or not record.reader_done
+                or record.notified
+                or record.process.poll() is None
+            ):
                 return
             record.notified = True
         self._emit(record, "completed")

penguin/tools/windows_pipes.py

@@ -0,0 +1,103 @@
+"""Bounded synchronous pipe polling for Windows subprocesses.
+
+Windows selectors accept sockets, not subprocess pipes. PeekNamedPipe lets the
+single capture worker drain ready pipes without blocking on a quiet stream.
+"""
+
+from __future__ import annotations
+
+import ctypes
+import msvcrt
+import selectors
+import threading
+import time
+from ctypes import wintypes
+from typing import Any
+
+__all__ = ["PipeSelector", "set_nonblocking"]
+
+_kernel = ctypes.WinDLL("kernel32", use_last_error=True)
+_kernel.PeekNamedPipe.argtypes = [
+    wintypes.HANDLE,
+    ctypes.c_void_p,
+    wintypes.DWORD,
+    ctypes.c_void_p,
+    ctypes.POINTER(wintypes.DWORD),
+    ctypes.c_void_p,
+]
+_kernel.PeekNamedPipe.restype = wintypes.BOOL
+_kernel.SetNamedPipeHandleState.argtypes = [
+    wintypes.HANDLE,
+    ctypes.POINTER(wintypes.DWORD),
+    ctypes.c_void_p,
+    ctypes.c_void_p,
+]
+_kernel.SetNamedPipeHandleState.restype = wintypes.BOOL
+
+
+def set_nonblocking(fd: int) -> None:
+    """Set PIPE_NOWAIT, including on Python versions before 3.12."""
+    mode = wintypes.DWORD(1)
+    if not _kernel.SetNamedPipeHandleState(
+        msvcrt.get_osfhandle(fd), ctypes.byref(mode), None, None
+    ):
+        raise ctypes.WinError(ctypes.get_last_error())
+
+
+class PipeSelector:
+    """Small read-only selector adapter for anonymous subprocess pipes."""
+
+    def __init__(self, stop_event: threading.Event | None = None) -> None:
+        self._keys: dict[int, selectors.SelectorKey] = {}
+        self._stop_event = stop_event or threading.Event()
+
+    def __enter__(self) -> PipeSelector:
+        return self
+
+    def __exit__(self, *_args: Any) -> None:
+        self._keys.clear()
+
+    def register(self, pipe: Any, events: int, data: str) -> None:
+        self._keys[pipe.fileno()] = selectors.SelectorKey(
+            pipe, pipe.fileno(), events, data
+        )
+
+    def unregister(self, pipe: Any) -> None:
+        del self._keys[pipe.fileno()]
+
+    def get_map(self) -> dict[int, selectors.SelectorKey]:
+        return self._keys
+
+    def select(self, timeout: float) -> list[tuple[selectors.SelectorKey, int]]:
+        """Return readable or closed pipes, waiting at most the given duration."""
+        # Windows Python 3.11/3.12 may use 15.625ms GetTickCount64 ticks for
+        # monotonic(). A completed 20ms wait can then look unfinished, causing
+        # extra scans and short tail waits. perf_counter uses high-resolution QPC.
+        deadline = time.perf_counter() + timeout
+        while True:
+            ready = []
+            for fd, key in self._keys.items():
+                available = wintypes.DWORD()
+                if not _kernel.PeekNamedPipe(
+                    msvcrt.get_osfhandle(fd),
+                    None,
+                    0,
+                    None,
+                    ctypes.byref(available),
+                    None,
+                ):
+                    error = ctypes.get_last_error()
+                    if error not in (109, 232):  # broken pipe / no data (closed)
+                        raise ctypes.WinError(error)
+                    ready.append((key, key.events))
+                elif available.value:
+                    ready.append((key, key.events))
+            remaining = deadline - time.perf_counter()
+            if ready or remaining <= 0 or self._stop_event.is_set():
+                return ready
+            # Anonymous Windows pipes cannot be selected. Bound readiness scans
+            # to the capture cadence and block between scans, including when
+            # both streams have closed but the child is still running.
+            self._stop_event.wait(min(remaining, 0.02))
+            if time.perf_counter() >= deadline:
+                return []

tests/system/test_tool_environment.py

@@ -10,6 +10,7 @@
 import subprocess
 import sys
 from pathlib import Path
+from types import SimpleNamespace
 
 import pytest
 
@@ -78,6 +79,38 @@ def test_local_mode_preserves_inheritance(monkeypatch: pytest.MonkeyPatch) -> No
     assert build_tool_environment({"PATH": "custom"})["PATH"] == "custom"
 
 
+@pytest.mark.parametrize("mode", ["local", "hosted"])
+def test_windows_cleanup_helper_uses_environment_policy(
+    mode: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Exercise the Windows launch site without needing Windows or taskkill."""
+    from penguin.tools import process_runtime
+
+    monkeypatch.setenv("PENGUIN_TOOL_ENVIRONMENT", mode)
+    monkeypatch.setenv("SYNTHETIC_CLEANUP_SECRET", "synthetic-secret")
+    monkeypatch.setattr(
+        process_runtime, "os", SimpleNamespace(name="nt", environ=os.environ)
+    )
+    calls = []
+
+    def run_helper(command, **kwargs):
+        calls.append((command, kwargs))
+        return SimpleNamespace(returncode=0)
+
+    monkeypatch.setattr(process_runtime.subprocess, "run", run_helper)
+    record = SimpleNamespace(process=SimpleNamespace(pid=12345, poll=lambda: None))
+    with tool_environment_scope({"CLEANUP_SCOPE": "scoped-value"}):
+        ProcessRuntime(log_dir=tmp_path)._signal(record, "kill")
+        assert calls[0][1]["env"] == build_tool_environment()
+
+    environment = calls[0][1]["env"]
+    assert calls[0][0][-4:] == ["/PID", "12345", "/T", "/F"]
+    assert ("SYNTHETIC_CLEANUP_SECRET" in environment) is (mode == "local")
+    if mode == "hosted":
+        assert environment["CLEANUP_SCOPE"] == "scoped-value"
+        assert "synthetic-secret" not in environment.values()
+
+
 def test_unknown_mode_fails_closed(monkeypatch: pytest.MonkeyPatch) -> None:
     """A misspelled deployment policy must not select local inheritance."""
     monkeypatch.setenv("PENGUIN_TOOL_ENVIRONMENT", "hosetd")