From a3d3403e8bb5ccbc2ed6e0e4be3e996c0145b947 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Fri, 30 Oct 2026 08:15:48 +0800 Subject: [PATCH 1/2] fix(benchmark): preserve native process evidence before cleanup Signed-off-by: huangruiteng --- benchmark/runtime/RUNTIME.md | 32 ++++ benchmark/runtime/worker.py | 53 ++++- .../tests/test_native_process_evidence.py | 181 ++++++++++++++++++ .../benchmark_toolkit/native_codex_process.py | 149 ++++++++++++++ 4 files changed, 407 insertions(+), 8 deletions(-) create mode 100644 benchmark/tests/test_native_process_evidence.py create mode 100644 loopx/capabilities/benchmark_toolkit/native_codex_process.py diff --git a/benchmark/runtime/RUNTIME.md b/benchmark/runtime/RUNTIME.md index 223f0e9a89..9f9746c662 100644 --- a/benchmark/runtime/RUNTIME.md +++ b/benchmark/runtime/RUNTIME.md @@ -432,6 +432,38 @@ Pending controlled Turns retain their identity across worker restarts. Core does not delete homes or session bindings, edit registries directly, or monkeypatch CLI internals. +### Native Goal process observations + +New `native-goal` and `loopx-goal` worker executions also write +`native-process.json` in the wake directory and include the same observation in +`receipt.json.native_process`. No additional setting is required. The existing +worker owns launch and cleanup; the benchmark capability observes its shared +stdio transport without restarting the process or replaying requests. + +The snapshot retains PID, observation time and Linux process-start ticks when +available; the first transport API failure (operation, exception type and message +digest); process return code/signal before cleanup; requested cleanup signals; +and the status observed after cleanup. Startup and failure observations are +written immediately, with atomic owner-only snapshot replacement. Earlier +edges stay in the snapshot. Task text, messages, command arguments, environment +and stderr are not copied into it. A write failure is disclosed separately and +does not replace the original runtime exception or prevent process cleanup. + +| Observation | Supported interpretation | +| --- | --- | +| Exit code or signal observed before cleanup | The child was already terminal; the code does not by itself establish root cause. | +| Stream failure while return code is null | The exchange failed while child termination was unobserved; EOF alone is not proof of process death. | +| Signal observed only after cleanup started | Cleanup may have contributed. Do not count this signal alone as a native crash. | +| Exit code 0 with an unfinished transaction | Process exit is not proof of Goal completion. Preserve the protocol failure. | +| Missing start ticks or missing final observation | Identity/terminal evidence remains incomplete; do not infer the missing fact from timestamps. | + +These records do not assert OOM, scorer validity, score countability or fault +ownership. An abrupt worker kill may leave only the last successfully written +snapshot. Final error accounting still needs the native transaction and run +evidence. Historical attempts remain frozen and gain no retrospective proof; +new experiment adoption requires the reviewed runner/source profile. Other +worker modes retain their existing receipts and process behavior. + ## Migration and qualification This is the native runtime bridge/adapter slice in the research program's diff --git a/benchmark/runtime/worker.py b/benchmark/runtime/worker.py index 9d99ec4ebf..9479fe3e27 100644 --- a/benchmark/runtime/worker.py +++ b/benchmark/runtime/worker.py @@ -130,8 +130,9 @@ def heartbeat_body(env: dict[str, str], turn_id: str, *, native_goal: bool, @contextmanager def child_process(command: list[str], *, env: dict[str, str], stdout, stderr, - goal_admission=None, check_current_launch=None): + goal_admission=None, check_current_launch=None, process_observation=None): process = None + scope_error = None try: if check_current_launch is not None and goal_admission is None: raise ValueError("current Host launch check requires its original Goal admission") @@ -154,10 +155,27 @@ def child_process(command: list[str], *, env: dict[str, str], stdout, stderr, # This affects the isolated child, never the parent or off mode. 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) yield process + except BaseException as exc: + scope_error = type(exc).__name__ + raise finally: if process is not None: - _reap_child_process(process) + cleanup_error = None + try: + if process_observation is not None: + process_observation.before_cleanup(process, scope_error) + finally: + try: + _reap_child_process(process, process_observation=process_observation) + except BaseException as exc: + cleanup_error = type(exc).__name__ + raise + finally: + if process_observation is not None: + process_observation.after_cleanup(process, cleanup_error) def heartbeat_launch_check(env, session_wake, *, original_quota=None): @@ -176,9 +194,11 @@ def check(registry_path): return check -def _reap_child_process(process): +def _reap_child_process(process, *, process_observation=None): """Reap the process group created by child_process, including descendants.""" try: + if process_observation is not None: + process_observation.signal_requested(signal.SIGINT) os.killpg(process.pid, signal.SIGINT) except ProcessLookupError: pass @@ -187,6 +207,8 @@ def _reap_child_process(process): except subprocess.TimeoutExpired: pass try: + if process_observation is not None: + process_observation.signal_requested(signal.SIGKILL) os.killpg(process.pid, signal.SIGKILL) except ProcessLookupError: pass @@ -304,16 +326,27 @@ def native_goal_rounds(run_round, *, budget_seconds, continue_after_complete, cl def run_native_goal( - env: dict[str, str], execution: Execution, body: str, receipt: dict, stderr + env: dict[str, str], execution: Execution, body: str, receipt: dict, stderr, + *, process_evidence_path: Path | None = None, ) -> None: from loopx.capabilities.benchmark_toolkit.native_codex_goal import ( NativeGoalConfig, - StdioNativeGoalTransport, compact_native_goal_receipt, native_goal_outcome, refresh_native_goal_status, run_native_goal_until_terminal, ) + from loopx.capabilities.benchmark_toolkit.native_codex_process import ( + NativeCodexProcessObservation, ObservedStdioNativeGoalTransport, + write_process_observation, + ) + + def publish_process(snapshot): + receipt["native_process"] = snapshot + if process_evidence_path is not None: + write_process_observation(process_evidence_path, snapshot) + + process_observation = NativeCodexProcessObservation(publish_process) required_skills = () if execution.uses_loopx: @@ -337,9 +370,12 @@ def run_native_goal( required_skill_ids=required_skills, ) with child_process( - [env["CODEX_BIN"], "app-server"], env=env, stdout=subprocess.PIPE, stderr=stderr + [env["CODEX_BIN"], "app-server"], env=env, stdout=subprocess.PIPE, stderr=stderr, + process_observation=process_observation, ) as process: - transport = StdioNativeGoalTransport(process, response_timeout_sec=120) + transport = ObservedStdioNativeGoalTransport( + process, observation=process_observation, response_timeout_sec=120, + ) def run_round(suffix, timeout_sec): turn = run_native_goal_until_terminal( @@ -572,7 +608,8 @@ def run_once(env: dict[str, str]) -> dict: return receipt with (wake / "stderr.log").open("w") as stderr: if execution.native_goal and stage == "execute": - run_native_goal(env, execution, body, receipt, stderr) + run_native_goal(env, execution, body, receipt, stderr, + process_evidence_path=wake / "native-process.json") else: pending = {} if execution.mode == "turn" and stage == "execute": diff --git a/benchmark/tests/test_native_process_evidence.py b/benchmark/tests/test_native_process_evidence.py new file mode 100644 index 0000000000..5532697549 --- /dev/null +++ b/benchmark/tests/test_native_process_evidence.py @@ -0,0 +1,181 @@ +"""Process observations must distinguish native exits from our own cleanup.""" + +import json +import os +import signal +import subprocess +import sys +from pathlib import Path + +import pytest + +from benchmark.tests.test_native_goal_outcomes import native_worker +from benchmark.tests.test_shared_codex_runtime import worker_env + + +def failed_worker(tmp_path, action, *, during_turn=False): + env = worker_env(tmp_path) + env.update(LOOPX_EXECUTION_MODE="native-goal", LOOPX_CODEX_TURN_TIMEOUT_SEC="5") + script = f'''#!{sys.executable} +import json, os, signal, sys, time +from pathlib import Path +objective = "" +for line in sys.stdin: + request = json.loads(line) + method = request["method"] + with Path("requests.jsonl").open("a") as log: + log.write(json.dumps(method) + "\\n") + if method == "initialized": + continue + if method == "initialize": + if not {during_turn!r}: + {action} + result = {{}} + elif method == "thread/start": + result = {{"thread": {{"id": "thread"}}}} + elif method == "thread/goal/set": + objective = request["params"]["objective"] + result = {{}} + elif method == "thread/goal/get": + result = {{"goal": {{"threadId": "thread", "objective": objective, "status": "active"}}}} + elif method == "turn/start": + result = {{"turn": {{"id": "turn", "status": "inProgress"}}}} + else: + raise AssertionError(method) + print(json.dumps({{"id": request["id"], "result": result}}), flush=True) + if method == "turn/start": + print(json.dumps({{"method": "turn/started", "params": {{"threadId": "thread", + "turn": {{"id": "turn", "status": "inProgress"}}}}}}), flush=True) + print(json.dumps({{"method": "item/started", "params": {{"threadId": "thread", + "turnId": "turn", "item": {{"type": "commandExecution", "id": "tool"}}}}}}), flush=True) + {action} +''' + Path(env["CODEX_BIN"]).write_text(script) + result = subprocess.run([sys.executable, "-m", "benchmark.runtime.worker"], + env=env, capture_output=True, text=True, timeout=20) + receipt_path, = (tmp_path / "logs/wakes").glob("*/receipt.json") + receipt = json.loads(receipt_path.read_text()) + assert not receipt["ok"] + assert result.returncode != 0 + observation = json.loads((receipt_path.parent / "native-process.json").read_text()) + assert observation == receipt["native_process"] + assert observation["pid"] > 0 + assert observation["started_at_unix"] > 0 + if sys.platform.startswith("linux"): + assert observation["linux_start_ticks"] > 0 + calls = [json.loads(line) for line in (Path(env["LOOPX_PROJECT"]) / "requests.jsonl").read_text().splitlines()] + assert calls.count("initialize") == 1 # The observer never replays a transaction. + return observation, receipt, calls + + +@pytest.mark.parametrize("code", [0, 23]) +def test_request_eof_retains_actual_exit_code(tmp_path, code): + observation, receipt, calls = failed_worker(tmp_path, f"os._exit({code})") + assert observation["first_failure"]["operation"] == "request" + assert observation["first_failure"]["error_type"] == "NativeGoalProtocolError" + assert observation["after_cleanup"]["return_code"] == code + assert observation["after_cleanup"]["signal"] is None + assert calls == ["initialize"] + # A zero process exit did not finish the outstanding native transaction. + assert receipt["error_kind"] == "NativeGoalProtocolError" + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX child signal semantics") +def test_signal_death_retains_signal_without_inferring_oom(tmp_path): + observation, _, _ = failed_worker(tmp_path, "os.kill(os.getpid(), signal.SIGKILL)") + assert observation["after_cleanup"]["return_code"] == -signal.SIGKILL + assert observation["after_cleanup"]["signal"] == signal.SIGKILL + assert "oom" not in json.dumps(observation).lower() + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX pipe and signal semantics") +def test_live_child_eof_does_not_attribute_cleanup_signal_to_native_failure(tmp_path): + observation, _, calls = failed_worker(tmp_path, "os.close(1); time.sleep(30)") + assert observation["first_failure"]["process"]["return_code"] is None + assert observation["before_cleanup"]["return_code"] is None + assert observation["cleanup_started_at_unix"] >= observation["first_failure"]["observed_at_unix"] + assert observation["after_cleanup"]["signal"] in {signal.SIGINT, signal.SIGKILL} + assert calls == ["initialize"] + + +def test_eof_with_outstanding_tool_keeps_event_failure_and_partial_turn(tmp_path): + observation, receipt, calls = failed_worker(tmp_path, "os._exit(17)", during_turn=True) + assert observation["first_failure"]["operation"] == "next_event" + assert observation["after_cleanup"]["return_code"] == 17 + assert not receipt["native_goal"]["terminal_event_observed"] + assert calls.count("turn/start") == 1 + + +def test_normal_goal_completion_records_intentional_cleanup_separately(tmp_path): + result, receipt = native_worker(tmp_path, "complete", "completed") + assert result.returncode == 0 + observation = receipt["native_process"] + assert observation["first_failure"] is None + assert observation["cleanup_trigger"] == "scope_completed" + assert observation["before_cleanup"]["return_code"] is None + assert receipt["native_goal_outcome"] == "complete" + path, = (tmp_path / "logs/wakes").glob("*/native-process.json") + assert json.loads(path.read_text()) == observation + assert path.stat().st_mode & 0o777 == 0o600 + + +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 ( + NativeCodexProcessObservation, write_process_observation, + ) + + path = tmp_path / "process.json" + observation = NativeCodexProcessObservation(lambda value: write_process_observation(path, value)) + original_reap = worker._reap_child_process + checked = [] + + def reap(process, **kwargs): + retained = json.loads(path.read_text()) + assert retained["first_failure"]["operation"] == "request" + assert retained["before_cleanup"]["return_code"] == 0 + assert retained["after_cleanup"] is None + checked.append(True) + return original_reap(process, **kwargs) + + monkeypatch.setattr(worker, "_reap_child_process", reap) + with worker.child_process([sys.executable, "-c", "pass"], + env={**os.environ, "LOOPX_PROJECT": str(tmp_path)}, + stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, + process_observation=observation) as process: + startup = json.loads(path.read_text()) + assert startup["pid"] == process.pid + assert startup["before_cleanup"] is None + process.wait(timeout=5) + observation.failure(process, "request", RuntimeError("PRIVATE-SENTINEL")) + first = observation.snapshot()["first_failure"] + observation.failure(process, "next_event", ValueError("different private text")) + assert observation.snapshot()["first_failure"] == first + assert checked == [True] + assert "PRIVATE-SENTINEL" not in path.read_text() + assert "different private text" not in path.read_text() + + +def test_observation_write_failure_keeps_original_exception_and_reaps_child(tmp_path): + from benchmark.runtime.worker import child_process + from loopx.capabilities.benchmark_toolkit.native_codex_process import NativeCodexProcessObservation + + retained = [] + + def unavailable(snapshot): + retained.append(snapshot) + raise OSError("private destination is unavailable") + + observation = NativeCodexProcessObservation(unavailable) + original = RuntimeError("original failure") + with pytest.raises(RuntimeError) as caught: + with child_process([sys.executable, "-c", "import time; time.sleep(30)"], + env={**os.environ, "LOOPX_PROJECT": str(tmp_path)}, + stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, + process_observation=observation) as process: + raise original + assert caught.value is original + assert process.poll() is not None + assert observation.snapshot()["persistence_error_type"] == "OSError" + assert retained[-1]["persistence_error_type"] == "OSError" + assert "private destination" not in json.dumps(observation.snapshot()) diff --git a/loopx/capabilities/benchmark_toolkit/native_codex_process.py b/loopx/capabilities/benchmark_toolkit/native_codex_process.py new file mode 100644 index 0000000000..e11c365210 --- /dev/null +++ b/loopx/capabilities/benchmark_toolkit/native_codex_process.py @@ -0,0 +1,149 @@ +"""Passive native-process evidence for the existing benchmark worker. + +The shared stdio transport owns JSON-RPC; the worker owns process cleanup. +These observations grant neither retry authority nor a root-cause verdict. +""" + +from __future__ import annotations + +import copy +import json +import os +import subprocess +import tempfile +import time +from collections.abc import Callable +from hashlib import sha256 +from pathlib import Path +from typing import Any + +from ...control_plane.turn_driver.codex_stdio import StdioNativeGoalTransport + + +def _process_status(process: subprocess.Popen[str]) -> dict[str, Any]: + code = process.poll() + return { + "observed_at_unix": time.time(), + "return_code": code, + "signal": -code if os.name == "posix" and code is not None and code < 0 else None, + } + + +class NativeCodexProcessObservation: + """Keep a bounded, content-free snapshot across process and cleanup edges.""" + + def __init__(self, publish: Callable[[dict[str, Any]], None]) -> None: + self._publish = publish + self._data: dict[str, Any] = { + "schema_version": "native_codex_process_observation_v1", + "pid": None, + "started_at_unix": None, + "linux_start_ticks": None, + "first_failure": None, + "before_cleanup": None, + "cleanup_started_at_unix": None, + "cleanup_trigger": None, + "scope_error_type": None, + "cleanup_signals": [], + "after_cleanup": None, + "cleanup_error_type": None, + "persistence_error_type": None, + } + + def snapshot(self) -> dict[str, Any]: + return copy.deepcopy(self._data) + + def _emit(self) -> None: + snapshot = self.snapshot() + try: + self._publish(snapshot) + except OSError as exc: + # Observation failure must neither replace the original exception + # nor prevent the worker from reaping its owned process group. + 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()) + try: + # A missing /proc (or a child that already exited) is unknown. The + # parent timestamp is not a substitute for a native start identity. + stat = Path(f"/proc/{process.pid}/stat").read_text() + self._data["linux_start_ticks"] = int(stat.rsplit(")", 1)[1].split()[19]) + except (OSError, ValueError, IndexError): + pass + self._emit() + + def failure(self, process: subprocess.Popen[str], operation: str, exc: Exception) -> None: + if self._data["first_failure"] is not None: + return + self._data["first_failure"] = { + "operation": operation, + "error_type": type(exc).__name__, + "error_sha256": sha256(str(exc).encode("utf-8", errors="replace")).hexdigest(), + "observed_at_unix": time.time(), + "process": _process_status(process), + } + self._emit() + + def before_cleanup(self, process: subprocess.Popen[str], scope_error: str | None) -> None: + self._data["before_cleanup"] = _process_status(process) + self._data["cleanup_started_at_unix"] = time.time() + self._data["cleanup_trigger"] = "scope_exception" if scope_error else "scope_completed" + self._data["scope_error_type"] = scope_error + self._emit() + + def signal_requested(self, number: int) -> None: + self._data["cleanup_signals"].append({"signal": int(number), "requested_at_unix": time.time()}) + self._emit() + + def after_cleanup(self, process: subprocess.Popen[str], error: str | None) -> None: + self._data["after_cleanup"] = _process_status(process) + self._data["cleanup_error_type"] = error + self._emit() + + +class ObservedStdioNativeGoalTransport(StdioNativeGoalTransport): + """Observe failures at the shared transport API, without replay or new IO.""" + + def __init__(self, process, *, observation, response_timeout_sec=30): + self.observation = observation + super().__init__(process, response_timeout_sec=response_timeout_sec) + + def request(self, method, params): + try: + return super().request(method, params) + except Exception as exc: + self.observation.failure(self.process, "request", exc) + raise + + def notify(self, method, params): + try: + return super().notify(method, params) + except Exception as exc: + self.observation.failure(self.process, "notify", exc) + raise + + def next_event(self, *, timeout_sec): + try: + return super().next_event(timeout_sec=timeout_sec) + except Exception as exc: + self.observation.failure(self.process, "next_event", exc) + raise + + +def write_process_observation(path: Path, observation: dict[str, Any]) -> None: + """Atomically replace one caller-owned snapshot, retaining every earlier edge.""" + temporary = None + try: + with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", dir=path.parent, + prefix=path.name + ".", delete=False) as stream: + temporary = Path(stream.name) + json.dump(observation, stream, sort_keys=True) + stream.write("\n") + stream.flush() + os.fsync(stream.fileno()) + temporary.replace(path) + finally: + if temporary is not None: + temporary.unlink(missing_ok=True) From 125d132affbe4bc44822d36826382884a5a898e2 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Sat, 31 Oct 2026 00:06:23 +0800 Subject: [PATCH 2/2] fix(benchmark): stage native process observer Signed-off-by: huangruiteng --- benchmark/runtime/harbor.py | 1 + benchmark/tests/test_shared_codex_runtime.py | 9 +++++++-- 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/benchmark/runtime/harbor.py b/benchmark/runtime/harbor.py index 3908043203..e1a3e2ada5 100644 --- a/benchmark/runtime/harbor.py +++ b/benchmark/runtime/harbor.py @@ -48,6 +48,7 @@ NATIVE_GOAL_SOURCE_PATHS = ( "benchmark/runtime", "loopx/capabilities/benchmark_toolkit/native_codex_goal.py", + "loopx/capabilities/benchmark_toolkit/native_codex_process.py", "loopx/control_plane/turn_driver/codex_stdio.py", ) diff --git a/benchmark/tests/test_shared_codex_runtime.py b/benchmark/tests/test_shared_codex_runtime.py index 26e6d5ce66..91393f54d0 100644 --- a/benchmark/tests/test_shared_codex_runtime.py +++ b/benchmark/tests/test_shared_codex_runtime.py @@ -190,11 +190,16 @@ def test_native_goal_staged_source_is_importable_without_loopx_control_plane(tmp result = subprocess.run( [sys.executable, "-I", "-S", "-c", f"import sys; sys.path.insert(0, {str(staged)!r}); " - "import loopx.capabilities.benchmark_toolkit.native_codex_goal as m; print(m.__name__)"], + "import loopx.capabilities.benchmark_toolkit.native_codex_goal as goal; " + "import loopx.capabilities.benchmark_toolkit.native_codex_process as process; " + "print(goal.__name__, process.__name__)"], cwd=tmp_path, env={"PATH": os.environ["PATH"]}, capture_output=True, text=True, ) assert result.returncode == 0, result.stderr - assert result.stdout.strip() == "loopx.capabilities.benchmark_toolkit.native_codex_goal" + assert result.stdout.strip() == ( + "loopx.capabilities.benchmark_toolkit.native_codex_goal " + "loopx.capabilities.benchmark_toolkit.native_codex_process" + ) def test_native_goal_rounds_graceful_deadline_and_opt_in_continuation():