diff --git a/benchmark/runtime/interruption_recovery.py b/benchmark/runtime/interruption_recovery.py index 7d99e8ae8d..1c99f9917b 100644 --- a/benchmark/runtime/interruption_recovery.py +++ b/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) diff --git a/benchmark/runtime/worker.py b/benchmark/runtime/worker.py index 6f44ae5684..98e33f6109 100644 --- a/benchmark/runtime/worker.py +++ b/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, diff --git a/benchmark/tests/test_native_process_evidence.py b/benchmark/tests/test_native_process_evidence.py index 5532697549..292a014d7d 100644 --- a/benchmark/tests/test_native_process_evidence.py +++ b/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 ( diff --git a/loopx/capabilities/benchmark_toolkit/native_codex_process.py b/loopx/capabilities/benchmark_toolkit/native_codex_process.py index feb0da24ba..d24b3985e1 100644 --- a/loopx/capabilities/benchmark_toolkit/native_codex_process.py +++ b/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.