test: reap owned runtime children during drain cleanup

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 = []

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])

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)