fix(benchmark): capture the native process window before spawn

benchmark/runtime/interruption_recovery.py

@@ -40,23 +40,29 @@ def recovery_handoff(env: dict[str, str]) -> tuple[dict | None, str]:
 
 class ProcessObservers:
     """Preserve request-admission cleanup while retaining process observations."""
-    def __init__(self, *observers):
-        self.observers = [observer for observer in observers if observer is not None]
-
-    def started(self, process):
-        for observer in self.observers:
-            observer.started(process)
+    def __init__(self, native_observation, request_admission):
+        self._native_observation = native_observation
+        self._request_admission = request_admission
+        self._observers = tuple(
+            observer for observer in (native_observation, request_admission) if observer is not None
+        )
+
+    def started(self, process, *, launch_started_at_unix):
+        if self._native_observation is not None:
+            self._native_observation.started(process, launch_started_at_unix=launch_started_at_unix)
+        if self._request_admission is not None:
+            self._request_admission.started(process)
 
     def before_cleanup(self, process, error):
-        for observer in self.observers:
+        for observer in self._observers:
             observer.before_cleanup(process, error)
 
     def signal_requested(self, number):
-        for observer in self.observers:
+        for observer in self._observers:
             observer.signal_requested(number)
 
     def after_cleanup(self, process, error):
-        for observer in self.observers:
+        for observer in self._observers:
             observer.after_cleanup(process, error)
 
 

benchmark/runtime/worker.py

@@ -150,6 +150,8 @@ def child_process(command: list[str], *, env: dict[str, str], stdout, stderr,
         # launch with the existing Goal lifetime owner, never the whole model run.
         with (goal_admission.launch_lifetime(operation="codex_agent_host_launch")
               if goal_admission is not None else nullcontext()):
+            # The child can write a native event before Popen returns.
+            launch_started_at_unix = time.time()
             process = subprocess.Popen(
                 command,
                 cwd=env["LOOPX_PROJECT"],
@@ -164,7 +166,7 @@ def child_process(command: list[str], *, env: dict[str, str], stdout, stderr,
                 umask=0o077 if os.name == "posix" and env.get("LOOPX_CONTEXT_DELIVERY") == "on" else -1,
             )
         if process_observation is not None:
-            process_observation.started(process)
+            process_observation.started(process, launch_started_at_unix=launch_started_at_unix)
         yield process
     except BaseException as exc:
         scope_error = type(exc).__name__
@@ -214,12 +216,15 @@ def native_child_process(command, *, env, stdout, stderr, session_wake, receipt,
         guard = ProviderRequestAdmission(env, session_wake, receipt, wake,
             timeout=timeout, original_quota=original_quota)
         scope = guard.serve(command)
-    observation = guard
+    native_observation = None
     if env.get("LOOPX_HOST_ATTEMPT_ID") and session_wake is not None:
         from loopx.capabilities.benchmark_toolkit.native_codex_process import NativeCodexProcessObservation
-        from .interruption_recovery import ProcessObservers
 
         native_observation = NativeCodexProcessObservation(lambda value: receipt.update(native_process=value))
+    observation = None
+    if native_observation is not None or guard is not None:
+        from .interruption_recovery import ProcessObservers
+
         observation = ProcessObservers(native_observation, guard)
     with scope as routed:
         with child_process(routed, env=env, stdout=stdout, stderr=stderr,

benchmark/tests/test_native_process_evidence.py

@@ -119,6 +119,30 @@ def test_normal_goal_completion_records_intentional_cleanup_separately(tmp_path)
     assert path.stat().st_mode & 0o777 == 0o600
 
 
+@pytest.mark.skipif(os.name != "posix", reason="POSIX process-group cleanup")
+def test_process_window_starts_before_child_can_emit_events(tmp_path, monkeypatch):
+    from benchmark.runtime import worker
+    from loopx.capabilities.benchmark_toolkit.native_codex_process import NativeCodexProcessObservation
+
+    popen = worker.subprocess.Popen
+    child_started_at = []
+
+    def spawn_before_parent_observes(*args, **kwargs):
+        process = popen(*args, **kwargs)
+        stdout, _ = process.communicate(timeout=5)
+        child_started_at.append(float(stdout))
+        return process
+
+    monkeypatch.setattr(worker.subprocess, "Popen", spawn_before_parent_observes)
+    observation = NativeCodexProcessObservation(lambda _snapshot: None)
+    with worker.child_process(
+        [sys.executable, "-c", "import time; print(time.time())"],
+        env={**os.environ, "LOOPX_PROJECT": str(tmp_path)},
+        stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, process_observation=observation,
+    ):
+        assert observation.snapshot()["started_at_unix"] <= child_started_at[0]
+
+
 def test_first_failure_is_durable_before_cleanup_and_never_replaced(tmp_path, monkeypatch):
     from benchmark.runtime import worker
     from loopx.capabilities.benchmark_toolkit.native_codex_process import (

loopx/capabilities/benchmark_toolkit/native_codex_process.py

@@ -63,8 +63,8 @@ def _emit(self) -> None:
             self._data["persistence_error_type"] = type(exc).__name__
             snapshot["persistence_error_type"] = type(exc).__name__
 
-    def started(self, process: subprocess.Popen[str]) -> None:
-        self._data.update(pid=process.pid, started_at_unix=time.time())
+    def started(self, process: subprocess.Popen[str], *, launch_started_at_unix: float) -> None:
+        self._data.update(pid=process.pid, started_at_unix=launch_started_at_unix)
         try:
             # A missing /proc (or a child that already exited) is unknown. The
             # parent timestamp is not a substitute for a native start identity.