diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e14acd63..dffa8623 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/docs/docs/tools/processes.md b/docs/docs/tools/processes.md index dda5dab5..25a9ab1c 100644 --- a/docs/docs/tools/processes.md +++ b/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 diff --git a/penguin/tools/process_runtime.py b/penguin/tools/process_runtime.py index 47c3f478..51086feb 100644 --- a/penguin/tools/process_runtime.py +++ b/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") diff --git a/penguin/tools/windows_pipes.py b/penguin/tools/windows_pipes.py new file mode 100644 index 00000000..77354c5a --- /dev/null +++ b/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 [] diff --git a/tests/system/test_tool_environment.py b/tests/system/test_tool_environment.py index 1da08e31..4dfeaaba 100644 --- a/tests/system/test_tool_environment.py +++ b/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") diff --git a/tests/tools/test_windows_process_runtime.py b/tests/tools/test_windows_process_runtime.py new file mode 100644 index 00000000..d6121971 --- /dev/null +++ b/tests/tools/test_windows_process_runtime.py @@ -0,0 +1,408 @@ +"""Real native Windows pipes and process-lifecycle regression contracts.""" + +from __future__ import annotations + +import asyncio +import math +import os +import subprocess +import sys +import threading +import time +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from penguin.tools.process_runtime import ProcessRuntime +from penguin.tools.process_tools import ProcessTools + +pytestmark = pytest.mark.skipif(os.name != "nt", reason="Native Windows pipe contract") + + +@pytest.fixture +def runtime(tmp_path): + instance = ProcessRuntime(log_dir=tmp_path / "logs") + yield instance + instance.cleanup() + assert not instance._processes + + +@pytest.fixture +def command(tmp_path): + sequence = 0 + + def build(source: str) -> str: + nonlocal sequence + sequence += 1 + script = tmp_path / f"script with spaces {sequence}.py" + script.write_text(source, encoding="utf-8") + return subprocess.list2cmdline( + [sys.executable, "-u", "-X", "utf8", str(script)] + ) + + return build + + +def finish(runtime, process_id): + result = runtime.poll( + process_id, wait_ms=10000, wait_for_exit=True, max_chars=500000 + ) + assert result["process_status"] == "exited", result + assert result["reader_error"] is None, result + assert result["returncode"] is not None + runtime._processes[process_id].reader.join(1) + assert not runtime._processes[process_id].reader.is_alive() + return result + + +def test_delayed_output_survives_return_and_is_not_consumed(runtime): + events = [] + complete = threading.Event() + + def receive(kind, payload): + events.append((kind, payload)) + if kind == "completed": + complete.set() + + pid = runtime.start( + subprocess.list2cmdline( + [ + sys.executable, + "-c", + "import time; print('first', flush=True); " + "time.sleep(.2); print('last', flush=True)", + ] + ), + notify_on_complete=True, + on_event=receive, + )["process_id"] + assert complete.wait(5), events + assert runtime._processes[pid].cursors == {} + result = finish(runtime, pid) + assert "first" in result["output"] and "last" in result["output"] + assert result["returncode"] == 0 + completed = [p for kind, p in events if kind == "completed"] + assert len(completed) == 1 + assert completed[0]["returncode"] == 0 + assert completed[0]["completion_reason"] == "exited" + assert "last" in completed[0]["output_tail"] + assert runtime.poll(pid)["output"] == result["output"] + + +def test_cmd_launch_and_both_streams(runtime): + pid = runtime.start("echo native-out & echo native-err 1>&2")["process_id"] + result = finish(runtime, pid) + assert result["returncode"] == 0 + assert "native-out" in result["output"] and "native-err" in result["output"] + + +def test_large_concurrent_unicode_streams_do_not_deadlock(runtime, command): + source = """ +import os, threading +def write(fd, value): + data = (value * 80000).encode('utf-8') + while data: + data = data[os.write(fd, data):] +threads = [threading.Thread(target=write, args=(1, '\\u96ea')), + threading.Thread(target=write, args=(2, '\\u00e9'))] +for thread in threads: thread.start() +for thread in threads: thread.join() +""" + pid = runtime.start(command(source))["process_id"] + result = finish(runtime, pid) + assert result["returncode"] == 0 + assert result["output"].count("\u96ea") == 80000 + assert result["output"].count("\u00e9") == 80000 + assert "\ufffd" not in result["output"] + log = Path(result["log_path"]).read_text(encoding="utf-8") + assert log.count("\u96ea") == 80000 + assert log.count("\u00e9") == 80000 + + +def test_nonzero_exit_retains_output_and_error(runtime, command): + pid = runtime.start( + command("import sys; print('failure detail', file=sys.stderr); sys.exit(7)") + )["process_id"] + result = finish(runtime, pid) + assert result["returncode"] == 7 + assert result["error"] == "command_failed" + assert "failure detail" in result["output"] + + +def test_utf8_python_text_output(runtime, command): + pid = runtime.start( + command("import sys; print('\u96ea'); print('\u00e9',file=sys.stderr)") + )["process_id"] + result = finish(runtime, pid) + assert result["returncode"] == 0 + assert "\u96ea" in result["output"] + assert "\u00e9" in result["output"] + + +def test_quiet_capture_blocks_between_bounded_readiness_scans( + runtime, command, monkeypatch +): + from penguin.tools import windows_pipes + + original = windows_pipes._kernel.PeekNamedPipe + scans = [] + + def count(*args): + scans.append(time.monotonic()) + return original(*args) + + monkeypatch.setattr(windows_pipes._kernel, "PeekNamedPipe", count) + started = time.monotonic() + pid = runtime.start(command("import time; time.sleep(.4)"))["process_id"] + result = finish(runtime, pid) + elapsed = time.monotonic() - started + assert result["returncode"] == 0 + # Two pipes, one readiness scan per 20ms plus startup/EOF boundary scans. + assert len(scans) <= 2 * math.ceil(elapsed / 0.02) + 12, (len(scans), elapsed) + + +@pytest.mark.parametrize("timeout", [0.02, 0.06]) +def test_quiet_selector_does_not_rescan_due_to_coarse_clock(monkeypatch, timeout): + from penguin.tools import windows_pipes + + elapsed = [0.0] + waits, scans = [], [] + + class ClockedStopEvent: + def is_set(self): + return False + + def wait(self, seconds): + waits.append(seconds) + elapsed[0] += seconds + return False + + monkeypatch.setattr( + windows_pipes, + "time", + SimpleNamespace( + monotonic=lambda: math.floor(elapsed[0] / 0.015625) * 0.015625, + perf_counter=lambda: elapsed[0], + ), + ) + real_peek = windows_pipes._kernel.PeekNamedPipe + + def peek(*args): + scans.append(elapsed[0]) + return real_peek(*args) + + monkeypatch.setattr(windows_pipes._kernel, "PeekNamedPipe", peek) + read_fd, write_fd = os.pipe() + try: + selector = windows_pipes.PipeSelector(ClockedStopEvent()) + selector.register(SimpleNamespace(fileno=lambda: read_fd), 1, "stdout") + assert selector.select(timeout) == [] + assert elapsed[0] == pytest.approx(timeout) + assert waits == pytest.approx([0.02] * math.ceil(timeout / 0.02)) + assert len(scans) == math.ceil(timeout / 0.02) + finally: + os.close(read_fd) + os.close(write_fd) + + +def test_closed_stream_selector_blocks_and_cancellation_wakes_it(): + from penguin.tools.windows_pipes import PipeSelector + + stop = threading.Event() + selector = PipeSelector(stop) + started = time.monotonic() + assert selector.select(0.05) == [] + assert time.monotonic() - started >= 0.04 + waiter = threading.Thread(target=selector.select, args=(30,)) + waiter.start() + stop.set() + waiter.join(0.5) + assert not waiter.is_alive() + + +def test_stdin_is_utf8_and_backpressure_is_bounded(runtime, command): + pid = runtime.start( + command("import sys; print(sys.stdin.buffer.readline().hex())") + )["process_id"] + written = runtime.write_stdin(pid, "\u96ea\n") + assert written["bytes_written"] == 4 + assert "e99baa0a" in finish(runtime, pid)["output"] + pid = runtime.start(command("import time; time.sleep(30)"))["process_id"] + started = time.monotonic() + result = runtime.write_stdin(pid, "x" * 1000000, timeout_ms=100) + assert time.monotonic() - started < 2 + assert result["error"] == "stdin_write_timeout" + assert 0 < result["bytes_written"] < result["bytes_total"] + cancelled = threading.Event() + cancelled.set() + assert ( + runtime.write_stdin(pid, "ignored", cancel_event=cancelled)["error"] + == "stdin_write_cancelled" + ) + assert runtime.status(pid)["process_status"] == "running" + + +@pytest.mark.parametrize("mode", ["terminate", "kill", "interrupt"]) +def test_stop_modes_terminate_process_and_reader(runtime, command, mode): + pid = runtime.start(command("import time; print('started'); time.sleep(30)"))[ + "process_id" + ] + assert "started" in runtime.poll(pid, wait_ms=2000)["output"] + stopped = runtime.stop(pid, mode=mode, timeout=1) + assert stopped["process_status"] == "exited" + assert stopped["returncode"] is not None + assert stopped["completion_reason"] == "cancelled" + assert not runtime._processes[pid].reader.is_alive() + + +def test_interrupt_without_console_falls_back_to_tree_stop( + runtime, command, monkeypatch +): + pid = runtime.start(command("import time; time.sleep(30)"))["process_id"] + + def no_console(_signal): + raise OSError("No console attached") + + monkeypatch.setattr(runtime._processes[pid].process, "send_signal", no_console) + assert runtime.stop(pid, mode="interrupt")["process_status"] == "exited" + + +def test_hosted_cleanup_filters_helper_environment(runtime, command, monkeypatch): + """The actual Windows helper works with the restricted environment.""" + monkeypatch.setenv("PENGUIN_TOOL_ENVIRONMENT", "hosted") + monkeypatch.setenv("SYNTHETIC_CLEANUP_SECRET", "synthetic-secret") + real_run = subprocess.run + environments = [] + + def capture_run(*args, **kwargs): + environments.append(kwargs.get("env")) + return real_run(*args, **kwargs) + + monkeypatch.setattr(subprocess, "run", capture_run) + pid = runtime.start(command("import time; print('started'); time.sleep(30)"))[ + "process_id" + ] + assert "started" in runtime.poll(pid, wait_ms=2000)["output"] + assert runtime.stop(pid)["process_status"] == "exited" + assert environments + assert all(isinstance(environment, dict) for environment in environments) + assert all("SYNTHETIC_CLEANUP_SECRET" not in env for env in environments) + + +def test_timeout_stops_child_tree(runtime, command, tmp_path): + import ctypes + from ctypes import wintypes + + kernel = ctypes.WinDLL("kernel32", use_last_error=True) + kernel.OpenProcess.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD] + kernel.OpenProcess.restype = wintypes.HANDLE + kernel.WaitForSingleObject.argtypes = [wintypes.HANDLE, wintypes.DWORD] + kernel.WaitForSingleObject.restype = wintypes.DWORD + kernel.CloseHandle.argtypes = [wintypes.HANDLE] + kernel.CloseHandle.restype = wintypes.BOOL + marker = tmp_path / "leaked-child.txt" + child = ( + "import time; from pathlib import Path; time.sleep(30); " + f"Path({str(marker)!r}).write_text('leak')" + ) + source = ( + "import subprocess,sys,time; " + f"child=subprocess.Popen([sys.executable,'-c',{child!r}]); " + "print(child.pid); time.sleep(30)" + ) + pid = runtime.start(command(source), timeout_seconds=3)["process_id"] + output = runtime.poll(pid, wait_ms=2000)["output"] + child_pid = int(output.split()[-1]) + handle = kernel.OpenProcess(0x00100000, False, child_pid) # SYNCHRONIZE + assert handle, ctypes.WinError(ctypes.get_last_error()) + try: + result = finish(runtime, pid) + assert result["completion_reason"] == "timeout" + assert result["error"] == "timeout" + assert kernel.WaitForSingleObject(handle, 2000) == 0 # signalled: exited + finally: + kernel.CloseHandle(handle) + assert not marker.exists() + + +def test_capture_fault_is_reported_and_process_is_reaped(runtime, command, monkeypatch): + from penguin.tools.windows_pipes import PipeSelector + + def broken_pipe(self, timeout): + raise OSError("injected capture failure") + + monkeypatch.setattr(PipeSelector, "select", broken_pipe) + pid = runtime.start(command("import time; time.sleep(30)"))["process_id"] + result = runtime.poll(pid, wait_ms=5000, wait_for_exit=True) + assert result["process_status"] == "exited" + assert "injected capture failure" in result["reader_error"] + assert result["returncode"] is not None + + +def test_public_process_service_emits_after_start_returns(runtime, command): + service = ProcessTools(runtime) + owner = {"session_id": "session", "agent_id": "parent"} + try: + started = service.execute( + "process_start", + { + "command": command( + "import time; print('first'); time.sleep(.2); print('last')" + ) + }, + owner, + ) + pid = started["process_id"] + result = finish(runtime, pid) + assert result["returncode"] == 0 + assert runtime._processes[pid].cursors == {} + notices = service.pending(("session", "parent")) + assert len(notices) == 1 + assert notices[0]["returncode"] == 0 + assert "last" in notices[0]["output_tail"] + assert ( + service.execute("process_status", {"process_id": pid}, owner)["output"] + == "" + ) + assert "first" in runtime.poll(pid)["output"] + finally: + service.cleanup() + + +@pytest.mark.asyncio +async def test_bound_process_events_arrive_after_start_returns(runtime, command): + service = ProcessTools(runtime) + completed = asyncio.Event() + events = [] + + async def receive(kind, payload): + events.append((kind, payload)) + if kind == "completed": + completed.set() + + owner = {"session_id": "session", "agent_id": "parent"} + service.bind(("session", "parent"), receive) + try: + started = await asyncio.to_thread( + service.execute, + "process_start", + { + "command": command( + "import time; print('first'); time.sleep(.2); print('last')" + ) + }, + owner, + ) + await asyncio.wait_for(completed.wait(), 5) + pid = started["process_id"] + assert runtime._processes[pid].cursors == {} + assert any(kind == "output" for kind, _ in events) + result = next(payload for kind, payload in events if kind == "completed") + assert result["session_id"] == "session" + assert result["agent_id"] == "parent" + assert result["returncode"] == 0 + assert "last" in result["output_tail"] + assert "first" in runtime.poll(pid)["output"] + finally: + service.cleanup()