gpt-6.1-sol: reap stalled Xvfb children after replay captures

Author: N0zoM1z0

docs/RE_HANDOFF.md

@@ -31,6 +31,15 @@ Each port has its own build and runtime validation alongside the VC7 path.
 
 ## Native replay parity investigation
 
+A fresh cold gate is in progress under `build/replay-cold-20261011/`.
+Cold normal/bugfix links and all 1,106 accepted authored units passed. The
+new 108-case suite has 46 complete retail/native pairs with equal traces.
+The next retail capture completed all 113,660 calculations, then failed on
+an Xvfb shutdown timeout. Capture cleanup now kills and reaps a stalled
+Xvfb child; process tests cover graceful, stalled, and completed children.
+Resume with the archived candidate in this batch. Merge into main and sync
+`port/portable-64bit` after the full replay and ordinary-demo gates pass.
+
 The current playable VC7 build passes all three bundled demos against a fresh
 muted retail capture with ordinary pacing and rasterization: 16,080 frames,
 with all 22 recorded fields equal. Its executable SHA-256 is

scripts/capture-replay.py

@@ -20,6 +20,18 @@
 TARGET_SHA = "330fbdbf58a710829d65277b4f312cfbb38d5448b3df523e79350b879213d924"
 
 
+def stop_process(process, timeout=5):
+    """Terminate and reap an owned child, killing it if shutdown stalls."""
+    if process.poll() is not None:
+        return
+    process.terminate()
+    try:
+        process.wait(timeout=timeout)
+    except subprocess.TimeoutExpired:
+        process.kill()
+        process.wait(timeout=timeout)
+
+
 def observer_config(candidate, map_path):
     if candidate is None:
         return dict(gameManager=0x0160F508, player=0x017D5EF8, rng=0x0164D520, input=0x0164D52C,
@@ -388,9 +400,7 @@ def command(command_text):
                 shutil.copyfile(runtime / "log.txt", args.output_dir / "game.log")
             shutil.rmtree(runtime, ignore_errors=True)
         if xvfb:
-            if xvfb.poll() is None:
-                xvfb.terminate()
-                xvfb.wait(timeout=5)
+            stop_process(xvfb)
         if xvfb_log:
             xvfb_log.close()
 

scripts/ci.py

@@ -59,6 +59,7 @@ def main() -> int:
         run("Test local relocation validation", [sys.executable, "scripts/test-match-local-relocations.py"])
         run("Test replay trace comparison", [sys.executable, "scripts/test-replay-trace-comparison.py"])
         run("Test replay file validation", [sys.executable, "scripts/test-replay-file.py"])
+        run("Test replay capture cleanup", [sys.executable, "scripts/test-replay-capture-cleanup.py"])
         run("Validate replay route coverage", [sys.executable, "scripts/validate-replay-fixtures.py"])
         run(
             "Test semantic protocol guards",

scripts/test-replay-capture-cleanup.py

@@ -0,0 +1,50 @@
+#!/usr/bin/env python3
+"""Verify that replay capture cleanup reaps responsive and stalled children."""
+import importlib.util
+import os
+from pathlib import Path
+import signal
+import subprocess
+import sys
+import unittest
+
+
+spec = importlib.util.spec_from_file_location("capture", Path(__file__).with_name("capture-replay.py"))
+capture = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(capture)
+
+
+@unittest.skipUnless(os.name == "posix", "Process cleanup tests require POSIX signals")
+class CaptureCleanupTests(unittest.TestCase):
+    def check_child(self, ignore_termination, expected_signal):
+        script = (
+            "import signal\n"
+            + ("signal.signal(signal.SIGTERM, signal.SIG_IGN)\n" if ignore_termination else "")
+            + "print('ready', flush=True)\nsignal.pause()\n"
+        )
+        process = subprocess.Popen([sys.executable, "-c", script], stdout=subprocess.PIPE, text=True)
+        try:
+            self.assertEqual(process.stdout.readline(), "ready\n")
+            capture.stop_process(process, timeout=0.1)
+            self.assertEqual(process.returncode, -expected_signal)
+        finally:
+            if process.poll() is None:
+                process.kill()
+            process.wait(timeout=5)
+            process.stdout.close()
+
+    def test_responsive_child_exits_on_termination(self):
+        self.check_child(False, signal.SIGTERM)
+
+    def test_stalled_child_is_killed_and_reaped(self):
+        self.check_child(True, signal.SIGKILL)
+
+    def test_completed_child_keeps_its_exit_status(self):
+        process = subprocess.Popen([sys.executable, "-c", "pass"])
+        process.wait(timeout=5)
+        capture.stop_process(process, timeout=0.1)
+        self.assertEqual(process.returncode, 0)
+
+
+if __name__ == "__main__":
+    unittest.main()