From e44f90e0c24847081804f247bc3862adae7f66c8 Mon Sep 17 00:00:00 2001 From: Tartar Date: Thu, 29 Oct 2026 02:26:26 +0800 Subject: [PATCH] fix(host): contain Windows Host descendants in an owned Job Signed-off-by: Tartar --- .../turn_driver/host_process_transport.py | 249 ++++++++++-------- .../turn_driver/windows_host_job.py | 126 +++++++++ .../test_windows_host_process.py | 124 +++++++++ 3 files changed, 390 insertions(+), 109 deletions(-) create mode 100644 loopx/control_plane/turn_driver/windows_host_job.py create mode 100644 tests/control_plane/test_windows_host_process.py diff --git a/loopx/control_plane/turn_driver/host_process_transport.py b/loopx/control_plane/turn_driver/host_process_transport.py index bd9a1c646a..91374306c5 100644 --- a/loopx/control_plane/turn_driver/host_process_transport.py +++ b/loopx/control_plane/turn_driver/host_process_transport.py @@ -127,6 +127,15 @@ def _host_process_drain(record: dict[str, Any] | None, *, supervises: str, return (HOST_PROCESS_NOT_LAUNCHED if record.get("process_group") is None and "bridge_pid" not in record else HOST_PROCESS_UNATTRIBUTABLE) if not hasattr(os, "killpg"): + if sys.platform == "win32" and "windows_job" in record: + from .windows_host_job import active_job_processes, valid_job_name + if not valid_job_name(record["windows_job"]): + return HOST_PROCESS_UNATTRIBUTABLE + try: + active = active_job_processes(record["windows_job"]) + except OSError: + return HOST_PROCESS_UNATTRIBUTABLE + return HOST_PROCESS_DRAINING if active else HOST_PROCESS_DRAINED # A launched Host on a platform without process groups is never proven # drained. This is a platform boundary, not an attribution failure. return HOST_PROCESS_UNSUPPORTED_PLATFORM @@ -177,7 +186,11 @@ def execution_host_drain(record_path: Path, *, launch_possible: bool = False) -> host = _host_process_drain(owned_record, supervises=HOST_PROCESS_SUPERVISES_HOST, expected=True) drain = next(state for state in _DRAIN_PRECEDENCE if state in {supervisor, host}) - if not host_process_drain_supported() and (launch_possible or drain != HOST_PROCESS_NOT_LAUNCHED): + assigned_windows_job = sys.platform == "win32" and any( + "windows_job" in item for item in (owned_record or {}, supervisor_record or {}) + ) + if (not assigned_windows_job and not host_process_drain_supported() + and (launch_possible or drain != HOST_PROCESS_NOT_LAUNCHED)): return HOST_PROCESS_UNSUPPORTED_PLATFORM return drain @@ -283,6 +296,131 @@ def run_host_process( if delegated_lease is not None: request["delegated_lease"] = delegated_lease bridge = Path(__file__).with_name("host_process_bridge.ts") + bridge_environment, record_value, record_path, supervision, supervises = _bridge_record_scope(environment, delegated_lease) + launch_budget = max(1.0, timeout_seconds) if timeout_seconds is not None else 30.0 + launch_deadline = time.monotonic() + launch_budget if launch_guard is not None else None + launch_arguments = (["--guarded-launch", str((time.time() + launch_budget) * 1000)] + if launch_guard is not None else []) + with ExitStack() as containment: + windows_job = None + if sys.platform == "win32": + from .windows_host_job import WindowsHostJob + windows_job = containment.enter_context(WindowsHostJob()) + proc = subprocess.Popen( + [ + _node_executable(), + "--no-warnings", + "--experimental-strip-types", + str(bridge), + *launch_arguments, + ], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + encoding="utf-8", + errors="strict", + start_new_session=True, + env=bridge_environment, + ) + assert proc.stdin is not None and proc.stdout is not None + result: dict[str, Any] | None = None + record = None + launch_scope = ExitStack() + guarded = (GuardedHostBridge(proc, deadline=launch_deadline, + cleanup_timeout=8 if delegated_lease is not None else 5) + if launch_deadline is not None else None) + try: + if windows_job is not None: + windows_job.assign(proc) + if launch_guard is not None: + launch_scope.enter_context(launch_guard()) + record = _begin_host_process_record(record_path, record_value=record_value, + supervises=supervises, supervision=supervision, bridge_pid=proc.pid) + if record is not None and windows_job is not None: + record["windows_job"] = windows_job.name + _write_host_process_record(record_path, record) + frame = json.dumps(request, ensure_ascii=False, separators=(",", ":")) + "\n" + if guarded is not None: + guarded.send(frame) + else: + proc.stdin.write(frame) + proc.stdin.flush() + lines = iter(proc.stdout) + while True: + event = (guarded.next(deadline=launch_deadline) if guarded is not None + else json.loads(line) if (line := next(lines, None)) is not None else None) + if event is None: + break + kind = event.get("kind") + if kind == "spawned" and isinstance(event.get("pid"), int): + if record is not None: + assert record_path is not None + group = event.get("process_group") + record.update(phase="spawned", host_pid=event["pid"], + process_group=group if isinstance(group, int) else None) + _write_host_process_record(record_path, record) + launch_deadline = None + launch_scope.close() + elif kind in {"stdout", "stderr"} and isinstance(event.get("text"), str): + consume = on_stdout if kind == "stdout" else on_stderr + if consume is not None: + consume(event["text"]) + elif kind == "result" and event.get("outcome") in { + "exited", + "timeout", + "cancelled", + "output_limit", + "spawn_failed", + }: + result = event + launch_deadline = None + launch_scope.close() + elif kind == "not_launched" and guarded is not None: + guarded.not_launched = True + raise TimeoutError("managed Host bridge launch deadline expired") + else: + raise RuntimeError("Invalid managed Host process observation") + finally: + try: + if guarded is not None and (launch_deadline is not None or sys.exc_info()[0] is not None): + # A failed handoff still owns the Goal fence. The TS bridge must + # report no launch or finish its Host cleanup before it releases. + guarded.cancel() + if record is not None and guarded.not_launched: + assert record_path is not None + record.update(phase=HOST_PROCESS_NOT_LAUNCHED, process_group=None) + record.pop("bridge_pid", None) + _write_host_process_record(record_path, record) + else: + # EOF asks the TS supervisor to clean its Host, also on a + # callback failure. Leased cleanup retains its original budget. + proc.stdin.close() + try: + proc.wait(timeout=8 if delegated_lease is not None else 5) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait() + finally: + if windows_job is not None: + windows_job.finish() + launch_scope.close() + if guarded is not None: + guarded.close_output() + else: + proc.stdout.close() + if record is not None and result is not None and proc.returncode == 0: + assert record_path is not None + # The supervisor returned only after cleaning its group; the group is still re-read. + record["phase"] = "finished" + _write_host_process_record(record_path, record) + if proc.returncode != 0 or result is None: + raise RuntimeError("Managed Host process supervision returned no result") + return result + + +def _bridge_record_scope(environment, delegated_lease): + """Separate the supervisor's private record namespace from its child Host.""" # The bridge runs on the caller's environment, which may pin the release the # Host must run and name the record its supervisor owns. Copy it instead of # replacing it with the ambient one, and consume the record env from the @@ -302,111 +440,4 @@ def run_host_process( record_path = host_process_supervisor_record(record_path) supervises = HOST_PROCESS_SUPERVISES_NESTED_HOST supervision = "leased" - launch_budget = max(1.0, timeout_seconds) if timeout_seconds is not None else 30.0 - launch_deadline = time.monotonic() + launch_budget if launch_guard is not None else None - launch_arguments = (["--guarded-launch", str((time.time() + launch_budget) * 1000)] - if launch_guard is not None else []) - proc = subprocess.Popen( - [ - _node_executable(), - "--no-warnings", - "--experimental-strip-types", - str(bridge), - *launch_arguments, - ], - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.DEVNULL, - text=True, - encoding="utf-8", - errors="strict", - start_new_session=True, - env=bridge_environment, - ) - assert proc.stdin is not None and proc.stdout is not None - result: dict[str, Any] | None = None - record = None - launch_scope = ExitStack() - guarded = (GuardedHostBridge(proc, deadline=launch_deadline, - cleanup_timeout=8 if delegated_lease is not None else 5) - if launch_deadline is not None else None) - try: - if launch_guard is not None: - launch_scope.enter_context(launch_guard()) - record = _begin_host_process_record(record_path, record_value=record_value, - supervises=supervises, supervision=supervision, bridge_pid=proc.pid) - frame = json.dumps(request, ensure_ascii=False, separators=(",", ":")) + "\n" - if guarded is not None: - guarded.send(frame) - else: - proc.stdin.write(frame) - proc.stdin.flush() - lines = iter(proc.stdout) - while True: - event = (guarded.next(deadline=launch_deadline) if guarded is not None - else json.loads(line) if (line := next(lines, None)) is not None else None) - if event is None: - break - kind = event.get("kind") - if kind == "spawned" and isinstance(event.get("pid"), int): - if record is not None: - assert record_path is not None - group = event.get("process_group") - record.update(phase="spawned", host_pid=event["pid"], - process_group=group if isinstance(group, int) else None) - _write_host_process_record(record_path, record) - launch_deadline = None - launch_scope.close() - elif kind in {"stdout", "stderr"} and isinstance(event.get("text"), str): - consume = on_stdout if kind == "stdout" else on_stderr - if consume is not None: - consume(event["text"]) - elif kind == "result" and event.get("outcome") in { - "exited", - "timeout", - "cancelled", - "output_limit", - "spawn_failed", - }: - result = event - launch_deadline = None - launch_scope.close() - elif kind == "not_launched" and guarded is not None: - guarded.not_launched = True - raise TimeoutError("managed Host bridge launch deadline expired") - else: - raise RuntimeError("Invalid managed Host process observation") - finally: - try: - if guarded is not None and (launch_deadline is not None or sys.exc_info()[0] is not None): - # A failed handoff still owns the Goal fence. The TS bridge must - # report no launch or finish its Host cleanup before it releases. - guarded.cancel() - if record is not None and guarded.not_launched: - assert record_path is not None - record.update(phase=HOST_PROCESS_NOT_LAUNCHED, process_group=None) - record.pop("bridge_pid", None) - _write_host_process_record(record_path, record) - else: - # EOF asks the TS supervisor to clean its Host, also on a - # callback failure. Leased cleanup retains its original budget. - proc.stdin.close() - try: - proc.wait(timeout=8 if delegated_lease is not None else 5) - except subprocess.TimeoutExpired: - proc.kill() - proc.wait() - finally: - launch_scope.close() - if guarded is not None: - guarded.close_output() - else: - proc.stdout.close() - if record is not None and result is not None and proc.returncode == 0: - assert record_path is not None - # The supervisor returned only after cleaning its group; the group is still re-read. - record["phase"] = "finished" - _write_host_process_record(record_path, record) - if proc.returncode != 0 or result is None: - raise RuntimeError("Managed Host process supervision returned no result") - return result + return bridge_environment, record_value, record_path, supervision, supervises diff --git a/loopx/control_plane/turn_driver/windows_host_job.py b/loopx/control_plane/turn_driver/windows_host_job.py new file mode 100644 index 0000000000..3a23e7cd84 --- /dev/null +++ b/loopx/control_plane/turn_driver/windows_host_job.py @@ -0,0 +1,126 @@ +"""Windows-only kernel containment IO for the existing typed Host supervisor. + +No task, stop, lease or deadline decision belongs here. The parent assigns the +control bridge before sending its first request; the bridge cannot launch a +Host before that input. Neither breakaway nor inherited Job handles is allowed. +""" +from __future__ import annotations + +import ctypes +import re +import time +import uuid +from ctypes import wintypes + + +class _BasicLimit(ctypes.Structure): + _fields_ = [("per_process_time", ctypes.c_longlong), ("per_job_time", ctypes.c_longlong), + ("flags", wintypes.DWORD), ("min_working_set", ctypes.c_size_t), + ("max_working_set", ctypes.c_size_t), ("active_limit", wintypes.DWORD), + ("affinity", ctypes.c_size_t), ("priority", wintypes.DWORD), ("scheduling", wintypes.DWORD)] + + +class _IoCounters(ctypes.Structure): + _fields_ = [(name, ctypes.c_ulonglong) for name in + ("read_ops", "write_ops", "other_ops", "read_bytes", "write_bytes", "other_bytes")] + + +class _ExtendedLimit(ctypes.Structure): + _fields_ = [("basic", _BasicLimit), ("io", _IoCounters), + ("process_memory", ctypes.c_size_t), ("job_memory", ctypes.c_size_t), + ("peak_process_memory", ctypes.c_size_t), ("peak_job_memory", ctypes.c_size_t)] + + +class _Accounting(ctypes.Structure): + _fields_ = [("user", ctypes.c_longlong), ("kernel", ctypes.c_longlong), + ("period_user", ctypes.c_longlong), ("period_kernel", ctypes.c_longlong), + ("page_faults", wintypes.DWORD), ("total", wintypes.DWORD), + ("active", wintypes.DWORD), ("terminated", wintypes.DWORD)] + + +_kernel = ctypes.WinDLL("kernel32", use_last_error=True) +for _name, _arguments, _result in ( + ("CreateJobObjectW", [ctypes.c_void_p, wintypes.LPCWSTR], wintypes.HANDLE), + ("OpenJobObjectW", [wintypes.DWORD, wintypes.BOOL, wintypes.LPCWSTR], wintypes.HANDLE), + ("SetInformationJobObject", [wintypes.HANDLE, ctypes.c_int, ctypes.c_void_p, wintypes.DWORD], wintypes.BOOL), + ("AssignProcessToJobObject", [wintypes.HANDLE, wintypes.HANDLE], wintypes.BOOL), + ("QueryInformationJobObject", [wintypes.HANDLE, ctypes.c_int, ctypes.c_void_p, wintypes.DWORD, ctypes.c_void_p], wintypes.BOOL), + ("TerminateJobObject", [wintypes.HANDLE, wintypes.UINT], wintypes.BOOL), + ("CloseHandle", [wintypes.HANDLE], wintypes.BOOL), +): + _function = getattr(_kernel, _name) + _function.argtypes, _function.restype = _arguments, _result + + +def _checked(ok): + if not ok: + raise ctypes.WinError(ctypes.get_last_error()) + return ok + + +def _active(handle) -> int: + info = _Accounting() + _checked(_kernel.QueryInformationJobObject(handle, 1, ctypes.byref(info), ctypes.sizeof(info), None)) + return info.active + + +def valid_job_name(value: object) -> bool: + return isinstance(value, str) and re.fullmatch(r"Local\\LoopXHost-[0-9a-f]{32}", value) is not None + + +def active_job_processes(name: str) -> int | None: + """Fresh kernel observation. None means the named Job no longer exists. + + An inaccessible/unreadable Job is never absence. Closing this temporary + non-inherited query handle also avoids keeping a dead owner's Job alive. + """ + if not valid_job_name(name): + raise ValueError("invalid managed Host Job identity") + handle = _kernel.OpenJobObjectW(0x0004, False, name) # JOB_OBJECT_QUERY + if not handle: + error = ctypes.get_last_error() + if error == 2: # ERROR_FILE_NOT_FOUND, never access denied + return None + raise ctypes.WinError(error) + try: + return _active(handle) + finally: + _checked(_kernel.CloseHandle(handle)) + + +class WindowsHostJob: + """One execution's non-inherited, non-breakaway Job; no sandbox claims.""" + def __init__(self): + self.name = "Local\\LoopXHost-" + uuid.uuid4().hex + self.handle = _checked(_kernel.CreateJobObjectW(None, self.name)) + try: + if ctypes.get_last_error() == 183: # Never reuse another execution's Job. + raise RuntimeError("managed Host Job identity already exists") + limits = _ExtendedLimit() + limits.basic.flags = 0x2000 # JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE, no breakaway + _checked(_kernel.SetInformationJobObject(self.handle, 9, ctypes.byref(limits), ctypes.sizeof(limits))) + except BaseException: + _kernel.CloseHandle(self.handle) + raise + + def __enter__(self): + return self + + def assign(self, process) -> None: + # Popen retains the original process HANDLE, so PID reuse cannot redirect this assignment. + _checked(_kernel.AssignProcessToJobObject(self.handle, int(process._handle))) + + def finish(self) -> None: + if _active(self.handle): + _checked(_kernel.TerminateJobObject(self.handle, 1)) + deadline = time.monotonic() + 5 + while _active(self.handle): + if time.monotonic() >= deadline: + raise TimeoutError("managed Host Job still has active processes") + time.sleep(.01) + + def __exit__(self, *_exc): + try: + self.finish() + finally: + _checked(_kernel.CloseHandle(self.handle)) diff --git a/tests/control_plane/test_windows_host_process.py b/tests/control_plane/test_windows_host_process.py new file mode 100644 index 0000000000..0ea0d92225 --- /dev/null +++ b/tests/control_plane/test_windows_host_process.py @@ -0,0 +1,124 @@ +"""Real Windows descendant cleanup, including loss of the Python owner.""" +from __future__ import annotations + +import json +import os +import subprocess +import sys +import time +from pathlib import Path + +import pytest + +from loopx.control_plane.turn_driver import host_process_transport as transport + +pytestmark = pytest.mark.skipif(sys.platform != "win32", reason="Windows Job Object transport") + + +def host_source(marker: Path, pid: Path, *, exit_leader: bool = False) -> str: + child = ( + "import os,time;from pathlib import Path;" + f"Path({str(pid)!r}).write_text(str(os.getpid()));" + f"p=Path({str(marker)!r});" + "\nwhile True:\n with p.open('a') as f:f.write('x');f.flush()\n time.sleep(.02)" + ) + return ( + "import subprocess,sys,time;from pathlib import Path;" + f"subprocess.Popen([sys.executable,'-c',{child!r}],stdin=subprocess.DEVNULL," + "stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL,close_fds=True);" + f"p=Path({str(marker)!r});" + "\nwhile not p.exists():time.sleep(.01)\n" + "print('started',flush=True)\n" + ("" if exit_leader else "time.sleep(60)") + ) + + +def clean_fixture_child(pid: Path) -> None: + # Only a child this test actually launched and recorded; never enumerate processes. + if pid.exists(): + try: + os.kill(int(pid.read_text()), 9) + except (ProcessLookupError, PermissionError, OSError): + pass + + +@pytest.mark.parametrize("boundary", ["leader_exit", "timeout", "consumer_failure"]) +def test_windows_real_host_descendants_stop_before_return(tmp_path: Path, boundary: str) -> None: + marker, pid, record = (tmp_path / name for name in ("writes", "child.pid", "host.json")) + environment = {**os.environ, transport.HOST_PROCESS_RECORD_ENV: str(record)} + def consume(_text: str) -> None: + if boundary == "consumer_failure": + raise ValueError("consumer rejected") + try: + def run(): + return transport.run_host_process( + [sys.executable, "-c", host_source(marker, pid, exit_leader=boundary == "leader_exit")], + project=tmp_path, input_text="", timeout_seconds=2, + environment=environment, on_stdout=consume, + ) + if boundary == "consumer_failure": + with pytest.raises(ValueError, match="consumer rejected"): + run() + else: + result = run() + assert result["outcome"] == ("exited" if boundary == "leader_exit" else "timeout") + assert marker.exists(), "The real descendant must have executed" + assert transport.execution_host_drain(record) == transport.HOST_PROCESS_DRAINED + before = marker.read_bytes() + time.sleep(.2) + assert marker.read_bytes() == before, "A descendant continued writing after cleanup" + finally: + clean_fixture_child(pid) + + +def test_windows_cold_readback_after_owner_kill_sees_whole_job_exit(tmp_path: Path) -> None: + marker, pid, record = (tmp_path / name for name in ("writes", "child.pid", "host.json")) + argv = [sys.executable, "-c", host_source(marker, pid)] + source = ( + "from pathlib import Path;from loopx.control_plane.turn_driver.host_process_transport " + f"import run_host_process;run_host_process({argv!r},project=Path({str(tmp_path)!r})," + "input_text='',timeout_seconds=60)" + ) + owner = subprocess.Popen([sys.executable, "-c", source], + env={**os.environ, transport.HOST_PROCESS_RECORD_ENV: str(record)}, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + try: + deadline = time.monotonic() + 10 + while not marker.exists() and time.monotonic() < deadline: + time.sleep(.02) + assert marker.exists(), "Host never launched" + assert transport.execution_host_drain(record) == transport.HOST_PROCESS_DRAINING + owner.kill() + owner.wait(timeout=5) + deadline = time.monotonic() + 5 + while transport.execution_host_drain(record) != transport.HOST_PROCESS_DRAINED and time.monotonic() < deadline: + time.sleep(.02) + assert transport.execution_host_drain(record) == transport.HOST_PROCESS_DRAINED + before = marker.read_bytes() + time.sleep(.2) + assert marker.read_bytes() == before + finally: + if owner.poll() is None: + owner.kill() + owner.wait(timeout=5) + clean_fixture_child(pid) + + +def test_windows_legacy_finished_pid_does_not_prove_job_drain(tmp_path: Path) -> None: + record = tmp_path / "old.json" + from loopx.file_lock import lock_holder_host_label + record.write_text(json.dumps({"schema_version": transport.HOST_PROCESS_RECORD_SCHEMA_VERSION, + "host": lock_holder_host_label(), "supervises": "host", "supervision": "direct", + "phase": "finished", "bridge_pid": 9999999, "process_group": None})) + assert transport.execution_host_drain(record) == transport.HOST_PROCESS_UNSUPPORTED_PLATFORM + + +def test_windows_job_assignment_failure_launches_no_host(tmp_path: Path, monkeypatch) -> None: + from loopx.control_plane.turn_driver.windows_host_job import WindowsHostJob + marker = tmp_path / "must-not-execute" + def deny(_self, _process): + raise PermissionError("assignment rejected") + monkeypatch.setattr(WindowsHostJob, "assign", deny) + with pytest.raises(PermissionError, match="assignment rejected"): + transport.run_host_process([sys.executable, "-c", f"from pathlib import Path;Path({str(marker)!r}).write_text('bad')"], + project=tmp_path, input_text="", timeout_seconds=5) + assert not marker.exists()