diff --git a/tests/control_plane/runtime_retirement_diagnostics.py b/tests/control_plane/runtime_retirement_diagnostics.py index 31717253a2..b4bce531c1 100644 --- a/tests/control_plane/runtime_retirement_diagnostics.py +++ b/tests/control_plane/runtime_retirement_diagnostics.py @@ -1,10 +1,11 @@ -"""Failure-only observations for copied runtime drain and retirement fixtures.""" +"""Owned child cleanup and failure observations for copied runtime fixtures.""" from __future__ import annotations import json import os from pathlib import Path import shutil +import subprocess from loopx.control_plane import effect_runtime @@ -88,7 +89,9 @@ def instrument_copied_runtime(root: Path) -> None: errors.write_text(text) -def observe_copied_runtime(tmp_path: Path, monkeypatch, request) -> tuple[Path, list[int]]: +def observe_copied_runtime( + tmp_path: Path, monkeypatch, request, *, processes: list[subprocess.Popen] | None = None, +) -> tuple[Path, list[int]]: """Reuse the isolated source copy and retain stderr until failure readback.""" source_root = effect_runtime._control_plane_root() root = tmp_path / "control_plane" @@ -111,13 +114,31 @@ def close_captures(): def capture_stderr(args, *positional, **options): retained_stderr(args, options, root, captures) - return popen(args, *positional, **options) + process = popen(args, *positional, **options) + if processes is not None and str(root / "effect_runtime_server.ts") in args: + processes.append(process) + return process monkeypatch.setattr(effect_runtime.subprocess, "Popen", capture_stderr) monkeypatch.setattr(effect_runtime, "_control_plane_root", lambda: root) return root, captures +def reap_copied_runtime(processes: list[subprocess.Popen], pid: int) -> None: + """Dispose only the captured child; PID liveness cannot prove ownership.""" + matches = [process for process in processes if process.pid == pid] + assert len(matches) == 1, "fixture must own exactly one matching runtime child" + process = matches[0] + try: + process.wait(timeout=3) + except subprocess.TimeoutExpired: + # Popen retains the Windows process handle and handles exit racing + # termination. Reaping also prevents a later PID from being mistaken + # for this child. Real termination/wait errors still fail the fixture. + process.terminate() + process.wait(timeout=3) + + def trace_events(raw: bytes) -> list[dict]: """Never publish raw stderr or unknown fields, event names, or error text.""" result = [] diff --git a/tests/control_plane/test_effect_runtime_integration.py b/tests/control_plane/test_effect_runtime_integration.py index 707aba5b74..681d147346 100644 --- a/tests/control_plane/test_effect_runtime_integration.py +++ b/tests/control_plane/test_effect_runtime_integration.py @@ -14,7 +14,7 @@ import pytest from runtime_retirement_diagnostics import ( - failure_trace, file_state, observe_copied_runtime, trace_events, + failure_trace, file_state, observe_copied_runtime, reap_copied_runtime, trace_events, ) from loopx.control_plane import effect_runtime @@ -1400,7 +1400,8 @@ def test_runtime_drains_admitted_write_before_exit( cache_fault: str | None, ) -> None: """Socket lifetime must not define the lifetime of an accepted effect.""" - root, captures = observe_copied_runtime(tmp_path, monkeypatch, request) + processes = [] + root, captures = observe_copied_runtime(tmp_path, monkeypatch, request, processes=processes) if cache_fault: # Exercise cache failure even under an outer coverage/disabled-cache run, # using only this fixture's private cache rather than an inherited one. @@ -1512,14 +1513,7 @@ def write() -> dict[str, object]: raise finally: lock_path.unlink(missing_ok=True) - cleanup_deadline = time.monotonic() + 3 - while time.monotonic() < cleanup_deadline: - effect_runtime._reap_exited_runtime_child(info) - if not effect_runtime._pid_is_alive(original["pid"]): - break - time.sleep(0.025) - if effect_runtime._pid_is_alive(original["pid"]): - os.kill(int(original["pid"]), signal.SIGTERM) + reap_copied_runtime(processes, int(original["pid"])) @pytest.mark.parametrize("diagnostics_available", [True, False]) diff --git a/tests/control_plane/test_runtime_retirement_cleanup.py b/tests/control_plane/test_runtime_retirement_cleanup.py new file mode 100644 index 0000000000..4b0d406785 --- /dev/null +++ b/tests/control_plane/test_runtime_retirement_cleanup.py @@ -0,0 +1,93 @@ +"""Fixture cleanup retains the real write result when a PID probe is stale.""" +from __future__ import annotations + +import errno +import os +from pathlib import Path +import signal +import subprocess +import sys + +import pytest +import test_effect_runtime_integration as integration +from runtime_retirement_diagnostics import reap_copied_runtime +from loopx.control_plane import effect_runtime + + +def test_drain_cleanup_keeps_io_error_after_stale_pid_probe(tmp_path, monkeypatch, request, capsys): + processes = [] + popen, kill = effect_runtime.subprocess.Popen, os.kill + observe, alive = integration.failure_trace, effect_runtime._pid_is_alive + + def launch(args, *positional, **options): + process = popen(args, *positional, **options) + if any(Path(str(arg)).name == "effect_runtime_server.ts" for arg in args): + processes.append(process) + return process + + def stale_probe_after_error(captures): + events = observe(captures) + assert len(processes) == 1 + pid = processes[0].pid + # The hosted failure occurs after the original IO error: a liveness + # sample can outlive the process, and Windows refuses the later signal. + monkeypatch.setattr(effect_runtime, "_pid_is_alive", lambda value: True if value == pid else alive(value)) + return events + + def refuse_stale_signal(pid, sig): + if sig == signal.SIGTERM and any(process.pid == pid for process in processes): + raise PermissionError(errno.EACCES, "synthetic exited-process signal refusal") + return kill(pid, sig) + + monkeypatch.setattr(effect_runtime.subprocess, "Popen", launch) + monkeypatch.setattr(integration, "failure_trace", stale_probe_after_error) + monkeypatch.setattr(os, "kill", refuse_stale_signal) + try: + integration.test_runtime_drain_io_failure_reports_cause_without_suppressing_error( + tmp_path, monkeypatch, request, capsys, diagnostics_available=True, + ) + assert len(processes) == 1 and processes[0].poll() is not None + finally: + for process in processes: + if process.poll() is None: + process.kill() + process.wait(timeout=5) + + +@pytest.mark.parametrize("already_exited", [False, True]) +def test_reap_owned_child_preserves_neighbor(already_exited): + command = [sys.executable, "-c", "import time; print('ready',flush=True); time.sleep(60)"] + with subprocess.Popen(command, stdout=subprocess.PIPE, text=True) as child, \ + subprocess.Popen(command, stdout=subprocess.PIPE, text=True) as neighbor: + try: + assert child.stdout.readline().strip() == neighbor.stdout.readline().strip() == "ready" + if already_exited: + child.terminate() + child.wait(timeout=5) + reap_copied_runtime([child, neighbor], child.pid) + assert child.poll() is not None + assert neighbor.poll() is None + finally: + for process in (child, neighbor): + if process.poll() is None: + process.kill() + process.wait(timeout=5) + + +def test_reap_rejects_uncaptured_pid_and_preserves_termination_failure(monkeypatch): + with subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"]) as child: + try: + with pytest.raises(AssertionError, match="exactly one matching runtime child"): + reap_copied_runtime([], child.pid) + assert child.poll() is None + + def denied(): + raise PermissionError("synthetic live-child termination refusal") + + monkeypatch.setattr(child, "terminate", denied) + with pytest.raises(PermissionError, match="live-child termination refusal"): + reap_copied_runtime([child], child.pid) + assert child.poll() is None + finally: + child.kill() + child.wait(timeout=5)