fix(benchmark): preserve native process evidence before cleanup

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

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",
 )
 

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":

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

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():