From 0441b38584487b268747cfe2ba27f8e6d6d7441d Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sat, 7 Nov 2026 05:49:31 +0800
Subject: [PATCH 1/3] fix(benchmark): preserve failed workers before terminal
 teardown

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
(cherry picked from commit 18684daa62db2f40905037044a3acd2987009d75)
Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 benchmark/edgebench/README.md                 |  37 +++
 benchmark/edgebench/checkpoint_supervisor.py  |  17 +-
 benchmark/edgebench/checkpoint_worker.py      |   5 +-
 benchmark/edgebench/run.py                    |  15 +-
 benchmark/edgebench/terminal_capture.py       | 133 ++++++++++
 benchmark/runtime/sforge_backend.py           |  27 +-
 benchmark/tests/test_terminal_preservation.py | 233 ++++++++++++++++++
 7 files changed, 460 insertions(+), 7 deletions(-)
 create mode 100644 benchmark/edgebench/terminal_capture.py
 create mode 100644 benchmark/tests/test_terminal_preservation.py

diff --git a/benchmark/edgebench/README.md b/benchmark/edgebench/README.md
index 9573f37cb1..0220493ce5 100644
--- a/benchmark/edgebench/README.md
+++ b/benchmark/edgebench/README.md
@@ -711,6 +711,43 @@ their existing owners. No worker path, scoring, deadline or capture opportunity
 changes. Select this option only for a new run; it does not relocate a running
 bank or recover a missed point.
 
+## Preserve a failed worker before teardown
+
+For a new heartbeat run with full native controller capture, add
+`--preserve-failed-worker` to preserve the last source after a nonzero or uncertain
+solver exit. This is opt-in and requires agent-only submissions (`--eval-interval 0`);
+ordinary runs, successful exits and Native Goal completion keep their previous
+cleanup behavior. Removing the option disables the policy for later runs; it does
+not delete a previously retained worker.
+
+At the exact solver-return boundary, before the native provider stops its work
+container, the adapter closes the periodic observer and attempts one bounded joint
+capture through the existing Docker writer barrier and native checkpoint owner.
+It captures the complete task workspace, declared external research components,
+Goal state, scheduler/wake evidence, native session/configuration and native SQLite
+participants (including committed WAL data). It never exports the native credential
+home or `auth.json`. The native capsule retains its own submission history,
+incumbent, opportunity counter, cooldown and original phase. No new submission,
+model request, successful settlement or extended clock is created.
+
+The trial's private `terminal-preservation.json` records the original execution,
+export outcome/digest, preservation time and confirmed stopped-container state.
+The file export explicitly remains **unqualified for restoration**: a failed native
+Turn is not a completed conversation checkpoint. Both successful and failed exports
+retain the original stopped container until an operator separately qualifies a
+receiver or authorizes disposal. Compute is stopped before diagnostic collection;
+a stop failure is reported without deleting the source. Missing participants, full
+storage or an uncertain writer barrier cannot fall through to force-removal.
+The provider still performs its ordinary network/proxy cleanup and grader drain.
+
+Inspect this receipt, the exported manifest and the exact retained container before
+planning recovery. Do not restart it, treat a file digest as receiver adoption, or
+roll back to an older point to fill a tail gap. Retained containers still occupy
+**disk** and may contain their original private credentials; reserve storage and
+explicitly dispose of the exact owned container after recovery/disposal approval.
+Terminal preservation does not rotate away the last qualified regular checkpoint,
+and it does not increase the daily/fresh-replan capture frequency.
+
 ## Default research checkpoint retention
 
 **Default change:** new controller-enabled native runs automatically keep only
diff --git a/benchmark/edgebench/checkpoint_supervisor.py b/benchmark/edgebench/checkpoint_supervisor.py
index 15d27d94f0..4341ac3132 100644
--- a/benchmark/edgebench/checkpoint_supervisor.py
+++ b/benchmark/edgebench/checkpoint_supervisor.py
@@ -19,12 +19,19 @@
 
 class CheckpointSupervisor:
     def __init__(self, controller_file: Path, destination: Path, retention_policy: Path | None = None,
-                 *, retention_enabled: bool = True):
+                 *, retention_enabled: bool = True, terminal_components: list[dict[str, str]] | None = None,
+                 terminal_receipt: Path | None = None):
         _controller(controller_file)  # Fail before native registration.
         qualify_export_storage(destination)
         if not retention_enabled and retention_policy is not None:
             raise ValueError("A retention policy cannot be combined with disabled retention")
         self.controller_file, self.destination = controller_file, destination
+        self.preservation = None
+        if terminal_components is not None:
+            from .terminal_capture import FailurePreservation
+            if terminal_receipt is None or terminal_receipt.exists():
+                raise ValueError("Terminal preservation requires a fresh private receipt")
+            self.preservation = FailurePreservation(self, terminal_components, terminal_receipt)
         self.token = None
         self.stop = threading.Event()
         self.thread = None
@@ -47,6 +54,7 @@ def bind(self, environment):
     def start(self, container):
         if self.thread is not None or self.token is None:
             raise ValueError("Checkpoint supervisor requires one bound worker execution")
+        self.container = container
         self.destination.mkdir(mode=0o700, parents=True, exist_ok=False)
         if self.retention_enabled:
             from loopx.hosts.checkpoint_retention import _state
@@ -58,10 +66,15 @@ def start(self, container):
                                   mode=0o600, fsync_directory=True)
             self.maintenance = threading.Thread(target=self._maintain, daemon=True)
             self.maintenance.start()
-        self.container = container
         self.thread = threading.Thread(target=self._run, args=(container,), daemon=True)
         self.thread.start()
 
+    def finish_solver(self, execution):
+        if self.preservation is None:
+            self.close()
+        else:
+            self.preservation.finish(execution)
+
     def close(self):
         self.stop.set()
         if self.thread is not None:
diff --git a/benchmark/edgebench/checkpoint_worker.py b/benchmark/edgebench/checkpoint_worker.py
index 3ae6fd5879..98f894bcfa 100644
--- a/benchmark/edgebench/checkpoint_worker.py
+++ b/benchmark/edgebench/checkpoint_worker.py
@@ -57,7 +57,10 @@ def qualify_export_storage(destination: Path) -> None:
 
 
 def _request(path: Path) -> dict:
-    value = json.loads(_physical(path).read_bytes())
+    return _validate_request(json.loads(_physical(path).read_bytes()))
+
+
+def _validate_request(value: dict) -> dict:
     if (not isinstance(value, dict) or
             set(value) not in ({"container_id", "components", "phase_deadline_path"},
                               {"container_id", "components", "phase_deadline_path", "handoff"}) or
diff --git a/benchmark/edgebench/run.py b/benchmark/edgebench/run.py
index 0d3e9c477d..e22b7feabe 100644
--- a/benchmark/edgebench/run.py
+++ b/benchmark/edgebench/run.py
@@ -153,6 +153,8 @@ def main(argv=None):
                         help="Host-private native checkpoint controller; requires full fresh-replan handoff")
     parser.add_argument("--checkpoint-export-root", type=Path,
                         help="Existing physical host volume for full worker checkpoints; must preserve Linux filenames")
+    parser.add_argument("--preserve-failed-worker", action="store_true",
+                        help="Opt-in terminal joint capture and stopped-worker retention after a failed heartbeat execution; requires full controller capture")
     parser.add_argument("--checkpoint-request", type=Path,
                         help="Explicit host JSON declaration staged into a heartbeat worker; omitted disables capture")
     parser.add_argument("--restore-checkpoint", type=Path,
@@ -235,6 +237,10 @@ def main(argv=None):
         parser.error("controller file and full controller_handoff must be enabled together with native feedback")
     if (args.checkpoint_retention_policy is not None or args.no_checkpoint_retention) and not handoff:
         parser.error("Checkpoint retention requires the explicit native controller capture path")
+    if args.preserve_failed_worker and not handoff:
+        parser.error("Failed-worker preservation requires full native controller capture")
+    if args.preserve_failed_worker and args.eval_interval != 0:
+        parser.error("Failed-worker preservation requires agent-only native submissions")
     if args.checkpoint_export_root is not None:
         if not handoff:
             parser.error("Checkpoint export storage requires full controller capture")
@@ -411,9 +417,15 @@ def main(argv=None):
         from benchmark.edgebench.checkpoint_supervisor import CheckpointSupervisor
         destination = (args.checkpoint_export_root / args.run_id / args.task / "checkpoints"
                        if args.checkpoint_export_root is not None else trial / "checkpoints")
+        terminal = None
+        if args.preserve_failed_worker:
+            from benchmark.edgebench.terminal_capture import terminal_components
+            terminal = terminal_components(declaration, task.cwd)
         checkpoint = CheckpointSupervisor(args.checkpoint_controller_file, destination,
                                           retention_policy=args.checkpoint_retention_policy,
-                                          retention_enabled=not args.no_checkpoint_retention)
+                                          retention_enabled=not args.no_checkpoint_retention,
+                                          terminal_components=terminal,
+                                          terminal_receipt=(trial / "terminal-preservation.json") if terminal else None)
     backend = RecordingDockerBackend(log_dir=trial / "collected", logger=logger,
                                      oauth_proxy=bool(args.api_proxy_url or api_base_url),
                                      blind_api_endpoint=blind_endpoint, feedback=feedback,
@@ -483,6 +495,7 @@ def main(argv=None):
         **({"checkpoint_request_sha256": hashlib.sha256(checkpoint_request).hexdigest()}
            if checkpoint_request is not None else {}),
         **({"checkpoint_export_directory": str(checkpoint.destination)} if checkpoint is not None else {}),
+        **({"preserve_failed_worker": True} if args.preserve_failed_worker else {}),
     }
     receipt_path = trial / "runtime-receipt.json"
     receipt_path.write_text(json.dumps(receipt, indent=2))
diff --git a/benchmark/edgebench/terminal_capture.py b/benchmark/edgebench/terminal_capture.py
new file mode 100644
index 0000000000..9329a4cf9b
--- /dev/null
+++ b/benchmark/edgebench/terminal_capture.py
@@ -0,0 +1,133 @@
+"""Opt-in failure preservation at the native solver's terminal boundary.
+
+This Docker adapter reuses the existing joint-file/native-capsule capture owner.
+It does not qualify a failed conversation, settle a Turn, or authorize recovery.
+"""
+from __future__ import annotations
+
+import time
+import uuid
+from pathlib import PurePosixPath
+
+from loopx.file_lock import write_json_atomic
+from benchmark.runtime.layout import _CODEX_HOME, _CONTROL, _LOOPX_RUNTIME, _REGISTRY
+
+
+def terminal_components(declaration: dict, workspace: str) -> list[dict[str, str]]:
+    """Explicit full-capture participants, without the credential-bearing home."""
+    from benchmark.runtime.research_artifacts import validate_research_scope
+
+    validate_research_scope(workspace, ["."])
+    native_home = PurePosixPath(_CODEX_HOME)
+    paths = [PurePosixPath(workspace), *(PurePosixPath(row["source_path"]) for row in declaration["components"])]
+    if any(path == native_home or path in native_home.parents or native_home in path.parents for path in paths):
+        raise ValueError("Terminal native participants are allowlisted separately; never capture the credential home")
+    components = [{"id": "terminal_workspace", "source_path": workspace}]
+    for entry in declaration["components"]:
+        path = PurePosixPath(entry["source_path"])
+        if path == PurePosixPath(workspace) or PurePosixPath(workspace) in path.parents:
+            continue  # The complete workspace owns a declared submitted subset too.
+        components.append(dict(entry))
+    components.extend({"id": name, "source_path": path} for name, path in (
+        ("terminal_registry", _REGISTRY),
+        ("terminal_authority", _LOOPX_RUNTIME),
+        ("terminal_scheduler", f"{_CONTROL}/scheduler-state.json"),
+        ("terminal_wakes", "/logs/agent/wakes"),
+        ("terminal_sessions", f"{_CODEX_HOME}/sessions"),
+        ("terminal_native_state", f"{_CODEX_HOME}/state_5.sqlite"),
+        ("terminal_native_history", f"{_CODEX_HOME}/thread_history_1.sqlite"),
+        ("terminal_native_config", f"{_CODEX_HOME}/config.toml"),
+    ))
+    # Let the shared transport reject aliases/overlaps before launch, rather
+    # than silently dropping an explicit external research participant.
+    from benchmark.edgebench.checkpoint_worker import _validate_request
+    _validate_request({"container_id": "0" * 64, "components": components,
+                       "phase_deadline_path": f"{_CONTROL}/phase-deadline"})
+    return components
+
+
+class FailurePreservation:
+    """One terminal attempt; a stopped source remains until explicit disposal."""
+    def __init__(self, supervisor, components, receipt_path):
+        self.supervisor = supervisor
+        self.components = components
+        self.receipt_path = receipt_path
+        self.receipt = None
+        self.retain_source = False
+
+    def _write(self):
+        write_json_atomic(self.receipt_path,
+                          self.receipt, mode=0o600, fsync_directory=True)
+
+    def finish(self, execution):
+        if self.receipt is not None:
+            return  # Repeated cleanup cannot capture later state under this point.
+        if execution is not None and execution.get("exit_code") == 0 and not execution.get("timed_out"):
+            try:
+                self.supervisor.close()
+                return
+            except Exception:
+                # An unreleased observer barrier is itself an uncertain terminal
+                # outcome. Never delete its source via the ordinary finally.
+                pass
+        # Arm retention before any IO, observer join or optional export can fail.
+        self.retain_source = True
+        started = time.monotonic()
+        supervisor = self.supervisor
+        self.receipt = {"container_id": self.supervisor.container.id,
+            "observed_at_epoch": time.time(),
+            "capture_id": uuid.uuid4().hex, "execution": execution,
+            "capture_attempted": False, "capture_succeeded": False,
+            "restore_qualified": False, "source_retained": None, "source_stopped": False}
+        try:
+            self._write()
+            self.supervisor.close()
+            maintenance = getattr(self.supervisor, "maintenance", None)
+            if maintenance is not None and maintenance.is_alive():
+                raise RuntimeError("Checkpoint retention has not settled")
+            if execution is None or execution.get("execution_started") is not True:
+                return  # A failed setup has no original phase/native session to seal.
+            from .checkpoint_worker import capture_worker
+            from benchmark.runtime.checkpoint_handoff import CHECKPOINT_CAPTURE_SECONDS
+            self.receipt["capture_attempted"] = True
+            self._write()
+            request = self.supervisor.destination / "terminal-request.json"
+            write_json_atomic(request, {"container_id": self.supervisor.container.id,
+                "components": self.components, "phase_deadline_path": f"{_CONTROL}/phase-deadline"},
+                mode=0o600, fsync_directory=True)
+            point = capture_worker(supervisor.controller_file, token=supervisor.token,
+                capture_id=self.receipt["capture_id"],
+                destination=self.supervisor.destination / self.receipt["capture_id"],
+                request_file=request, timeout=CHECKPOINT_CAPTURE_SECONDS)
+            self.receipt.update(capture_succeeded=True, capture=point)
+        except Exception as error:
+            # No remote message, output, account information or token is retained.
+            self.receipt["error_kind"] = type(error).__name__
+        finally:
+            self.receipt["preservation_elapsed_seconds"] = time.monotonic() - started
+            self._write()
+
+    def stop_source(self, container):
+        """Release compute, never remove the only source after an uncertain save."""
+        if container.id != self.receipt["container_id"]:
+            raise ValueError("Failure preservation requires the original worker")
+        try:
+            container.reload()
+            if container.attrs["State"].get("Running"):
+                try:
+                    container.stop(timeout=15)
+                except Exception:
+                    container.kill()  # Bounded daemon operation; never kill a guessed host PID.
+            container.reload()
+            state = container.attrs["State"]
+            stopped = (state.get("Running") is False and state.get("Paused") is False
+                       and state.get("Restarting") is False and state.get("Pid") == 0)
+            self.receipt["source_retained"] = True
+            self.receipt["source_stopped"] = stopped
+            if not stopped:
+                raise RuntimeError("Preserved worker is not confirmed stopped")
+        except Exception as error:
+            self.receipt["stop_error_kind"] = type(error).__name__
+            raise
+        finally:
+            self._write()
diff --git a/benchmark/runtime/sforge_backend.py b/benchmark/runtime/sforge_backend.py
index 0edc0652f7..1c0e63f740 100644
--- a/benchmark/runtime/sforge_backend.py
+++ b/benchmark/runtime/sforge_backend.py
@@ -24,6 +24,8 @@ def __init__(self, *, log_dir: Path, logger, oauth_proxy: bool = False,
             raise ValueError("docker_host_bridge must be an explicit boolean")
         if type(research_sidecar) is not bool:
             raise ValueError("research_sidecar requires an explicit boolean")
+        if feedback is not None and getattr(checkpoint, "preservation", None) is not None:
+            raise ValueError("Failure preservation requires native feedback without an online sampler")
         self.research_sidecar = research_sidecar
         super().__init__(client)
         self.docker_host_bridge = docker_host_bridge
@@ -90,9 +92,12 @@ def _agent_environment(self, environment):
     def exec_run_with_timeout(self, handle, cmd, timeout=60, **kwargs):
         kwargs["environment"] = self._agent_environment(kwargs.get("environment"))
         is_solver = cmd == ["/bin/bash", "-c", getattr(self, "execution_command", None)]
-        if is_solver and self.checkpoint is not None:
+        preservation = getattr(self.checkpoint, "preservation", None)
+        if is_solver and self.checkpoint is not None and preservation is None:
             self.checkpoint.start(handle.raw)
         try:
+            if is_solver and preservation is not None:
+                self.checkpoint.start(handle.raw)
             result = super().exec_run_with_timeout(handle, cmd, timeout, **kwargs)
             if is_solver:
                 # SForge's RunResult discards the process exit code. Preserve
@@ -109,9 +114,18 @@ def exec_run_with_timeout(self, handle, cmd, timeout=60, **kwargs):
                 pending.write_text(json.dumps(self.execution_receipt))
                 pending.replace(target)
             return result
+        except Exception as error:
+            if (is_solver and self.execution_receipt is None
+                    and getattr(self.checkpoint, "preservation", None) is not None):
+                self.execution_receipt = {"transport_error_kind": type(error).__name__,
+                                          "execution_started": None}
+            raise
         finally:
             if is_solver and self.checkpoint is not None:
-                self.checkpoint.close()
+                if getattr(self.checkpoint, "preservation", None) is None:
+                    self.checkpoint.close()
+                else:
+                    self.checkpoint.finish_solver(self.execution_receipt)
             if is_solver and self.feedback is not None:
                 self.feedback.pause()
 
@@ -141,6 +155,9 @@ def start_feedback(self, handle):
 
     def cleanup_container(self, handle, logger=None):
         try:
+            preservation = getattr(self.checkpoint, "preservation", None)
+            if preservation is not None and preservation.retain_source:
+                preservation.stop_source(handle.raw)
             if self.checkpoint is not None:
                 self.checkpoint.close()
             if self.feedback is not None:
@@ -201,7 +218,11 @@ def cleanup_container(self, handle, logger=None):
                     self.native_goal_receipt = {"native_goal_outcome": "unknown"}
         finally:
             try:
-                super().cleanup_container(handle, logger)
+                preservation = getattr(self.checkpoint, "preservation", None)
+                if preservation is not None and preservation.retain_source:
+                    preservation.stop_source(handle.raw)
+                else:
+                    super().cleanup_container(handle, logger)
             finally:
                 if self.feedback is not None:
                     # Native cleanup logs some failures instead of raising. Verify the
diff --git a/benchmark/tests/test_terminal_preservation.py b/benchmark/tests/test_terminal_preservation.py
new file mode 100644
index 0000000000..382ca45d71
--- /dev/null
+++ b/benchmark/tests/test_terminal_preservation.py
@@ -0,0 +1,233 @@
+"""Failed native workers retain their last source, without granting recovery."""
+from __future__ import annotations
+
+import json
+import os
+from types import SimpleNamespace
+
+import pytest
+
+from benchmark.edgebench.terminal_capture import FailurePreservation, terminal_components
+from benchmark.tests import test_checkpoint_controller as native_fixtures
+from benchmark.tests import test_checkpoint_worker as docker_fixtures
+
+controllers = native_fixtures.controllers
+writer = docker_fixtures.writer
+
+
+def fixture_preservation(tmp_path, monkeypatch, fault=None):
+    calls = []
+    def close():
+        calls.append("close")
+        if fault == "barrier":
+            raise RuntimeError("private transport details")
+    supervisor = SimpleNamespace(destination=tmp_path, controller_file=tmp_path / "controller.json",
+        token="dummy-token", container=SimpleNamespace(id="a" * 64), close=close)
+    def capture(*args, **kwargs):
+        calls.append("capture")
+        if fault == "capture":
+            raise TimeoutError("dummy-token in remote message")
+        assert "handoff" not in json.loads(kwargs["request_file"].read_text())
+        return {"sha256": "b" * 64, "research_point_qualified": False}
+    monkeypatch.setattr("benchmark.edgebench.checkpoint_worker.capture_worker", capture)
+    return FailurePreservation(supervisor, [{"id": "workspace", "source_path": "/task"}],
+                               tmp_path / "terminal-preservation.json"), calls
+
+
+@pytest.mark.parametrize("fault", [None, "capture", "barrier"])
+def test_failed_capture_preserves_source_and_never_replays(tmp_path, monkeypatch, fault):
+    preservation, calls = fixture_preservation(tmp_path, monkeypatch, fault)
+    execution = {"exit_code": 2, "execution_started": True, "timed_out": False}
+    preservation.finish(execution)
+    preservation.finish(execution)
+    receipt = json.loads((tmp_path / "terminal-preservation.json").read_text())
+    assert calls == (["close"] if fault == "barrier" else ["close", "capture"])
+    assert receipt["execution"] == execution and receipt["restore_qualified"] is False
+    assert receipt["capture_succeeded"] is (fault is None)
+    assert preservation.retain_source and receipt["source_retained"] is None
+    assert "private" not in json.dumps(receipt)
+    box = SimpleNamespace(id="a" * 64, attrs={"State": {"Running": True, "Paused": False, "Restarting": False, "Pid": 17}})
+    box.reload = lambda: None
+    def stop(timeout):
+        assert timeout == 15
+        box.attrs["State"].update(Running=False, Pid=0)
+    box.stop = stop
+    preservation.stop_source(box)
+    assert json.loads((tmp_path / "terminal-preservation.json").read_text())["source_stopped"] is True
+
+
+@pytest.mark.parametrize("execution", [None, {"exit_code": 1, "execution_started": False}])
+def test_failed_setup_or_unknown_transport_does_not_invent_phase(tmp_path, monkeypatch, execution):
+    preservation, calls = fixture_preservation(tmp_path, monkeypatch)
+    preservation.finish(execution)
+    assert calls == ["close"] and preservation.retain_source
+    assert preservation.receipt["capture_attempted"] is False
+
+
+def test_success_retains_normal_cleanup_unless_writer_barrier_is_uncertain(tmp_path, monkeypatch):
+    for fault in [None, "barrier"]:
+        preservation, calls = fixture_preservation(tmp_path, monkeypatch, fault)
+        preservation.finish({"exit_code": 0, "execution_started": True, "timed_out": False})
+        assert preservation.retain_source is (fault == "barrier")
+        assert "capture" not in calls
+
+
+def test_component_declaration_captures_full_workspace_and_no_auth():
+    original = {"components": [{"id": "submitted", "source_path": "/task/solution.py"},
+                                {"id": "notes", "source_path": "/private-notes"}]}
+    components = terminal_components(original, "/task")
+    paths = [row["source_path"] for row in components]
+    assert "/task" in paths and "/task/solution.py" not in paths and "/private-notes" in paths
+    assert "/opt/loopx-benchmark/codex-home" not in paths
+    assert not any(path.endswith("auth.json") for path in paths)
+    assert original["components"][0]["source_path"] == "/task/solution.py"
+    with pytest.raises(ValueError, match="credential home"):
+        terminal_components({"components": [{"id": "unsafe", "source_path": "/opt/loopx-benchmark"}]}, "/task")
+    with pytest.raises(ValueError, match="credential home"):
+        terminal_components({"components": [{"id": "secret", "source_path": "/opt/loopx-benchmark/codex-home/auth.json"}]}, "/task")
+
+
+@pytest.mark.parametrize("failure", [False, True])
+def test_backend_terminal_failure_replaces_remove_with_verified_stop(tmp_path, monkeypatch, failure):
+    pytest.importorskip("sforge")
+    pytest.importorskip("harbor")
+    from benchmark.runtime.sforge_backend import RecordingDockerBackend, DockerBackend
+    backend = object.__new__(RecordingDockerBackend)
+    events = []
+    preservation = SimpleNamespace(retain_source=True, stop_source=lambda raw: events.append("stop"))
+    def close():
+        if failure:
+            raise RuntimeError("uncertain observer barrier")
+    backend.checkpoint = SimpleNamespace(close=close, preservation=preservation)
+    backend.feedback = None
+    backend.log_dir = tmp_path
+    backend.research_sidecar = False
+    monkeypatch.setattr(backend, "copy_from_container", lambda *args: (_ for _ in ()).throw(OSError()))
+    monkeypatch.setattr(DockerBackend, "cleanup_container", lambda *args: events.append("remove"))
+    handle = SimpleNamespace(raw=SimpleNamespace(id="a" * 64))
+    if failure:
+        with pytest.raises(RuntimeError, match="uncertain"):
+            backend.cleanup_container(handle)
+    else:
+        backend.cleanup_container(handle)
+    assert events == ["stop", "stop"]
+
+
+@pytest.mark.parametrize("enabled", [False, True])
+def test_partial_observer_start_preserves_only_opted_in_failure(tmp_path, monkeypatch, enabled):
+    pytest.importorskip("sforge")
+    pytest.importorskip("harbor")
+    from benchmark.runtime.sforge_backend import RecordingDockerBackend, DockerBackend
+    events = []
+    class Observer:
+        preservation = SimpleNamespace(retain_source=False) if enabled else None
+        def start(self, container):
+            events.append("start")
+            raise OSError("private setup text")
+        def close(self):
+            events.append("close")
+        def finish_solver(self, execution):
+            events.append("finish")
+            assert execution == {"transport_error_kind": "OSError", "execution_started": None}
+    backend = RecordingDockerBackend(log_dir=tmp_path / "collected", logger=None,
+        oauth_proxy=True, checkpoint=Observer(), client=SimpleNamespace())
+    backend.execution_command = "solver"
+    monkeypatch.setattr(DockerBackend, "exec_run_with_timeout", lambda *args, **kwargs: pytest.fail("must not start solver"))
+    with pytest.raises(OSError, match="private setup"):
+        backend.exec_run_with_timeout(SimpleNamespace(raw="owned"), ["/bin/bash", "-c", "solver"])
+    assert events == (["start", "finish"] if enabled else ["start"])
+    assert (backend.execution_receipt is None) is not enabled
+
+
+def test_unknown_stop_state_does_not_claim_released_compute(tmp_path, monkeypatch):
+    preservation, _ = fixture_preservation(tmp_path, monkeypatch)
+    preservation.finish({"exit_code": 2, "execution_started": True})
+    box = SimpleNamespace(id="a" * 64, attrs={"State": {}}, reload=lambda: None)
+    with pytest.raises(RuntimeError, match="not confirmed stopped"):
+        preservation.stop_source(box)
+    assert preservation.receipt["source_stopped"] is False
+
+
+def test_receipt_disk_failure_still_arms_source_retention(tmp_path, monkeypatch):
+    preservation, calls = fixture_preservation(tmp_path, monkeypatch)
+    monkeypatch.setattr(preservation, "_write", lambda: (_ for _ in ()).throw(OSError("full")))
+    with pytest.raises(OSError):
+        preservation.finish({"exit_code": 2, "execution_started": True})
+    assert preservation.retain_source and calls == []
+
+
+@pytest.mark.parametrize("profile", ["official", "native-goal", "heartbeat-explore"])
+def test_cli_preservation_requires_full_native_controller_before_trial(tmp_path, profile):
+    pytest.importorskip("sforge")
+    pytest.importorskip("harbor")
+    from benchmark.edgebench.run import main
+    with pytest.raises(SystemExit) as error:
+        main(["--task", "fixture", "--tasks-dir", str(tmp_path), "--log-dir", str(tmp_path),
+              "--run-id", "fixture", "--worker", profile, "--model", "fixture", "--effort", "xhigh",
+              "--judge-url", "http://127.0.0.1:9999", "--preserve-failed-worker"])
+    assert error.value.code == 2 and not (tmp_path / "runs").exists()
+
+
+@pytest.mark.skipif(os.environ.get("LOOPX_EDGEBENCH_DOCKER_SMOKE") != "1",
+                    reason="Requires isolated Docker and native HTTP checkpoint provider")
+@pytest.mark.parametrize("capture_failure", [False, True])
+def test_real_failed_solver_preserves_tail_and_native_opportunity(controllers, writer, tmp_path, capture_failure):
+    from sforge.harness.backend.docker_backend import DockerContainerHandle
+    from benchmark.runtime.sforge_backend import RecordingDockerBackend
+    from benchmark.edgebench.checkpoint_supervisor import CheckpointSupervisor
+    from benchmark.tests.test_checkpoint_controller import register, submit
+    source, _, first, _, files, grades = controllers
+    container, _ = writer
+    token = register(source, first, submission_cooldown=3600)
+    assert submit(first, token, 7).status_code == 200
+    script = '''from pathlib import Path
+import sqlite3,time
+root=Path('/opt/loopx-benchmark'); control=root/'control'; control.mkdir(parents=True)
+(control/'phase-deadline').write_text(str(int(time.time())+600))
+(control/'registry.json').write_text('{"fixture":true}')
+(control/'scheduler-state.json').write_text('{"host_attempt":{"status":"failed"}}')
+(root/'state/runtime').mkdir(parents=True)
+(root/'state/runtime/goal.json').write_text('{"todo":"open"}')
+home=root/'codex-home'; (home/'sessions').mkdir(parents=True)
+(home/'sessions/tail.jsonl').write_text('{"status":"failed","turn":"latest"}\\n')
+(home/'config.toml').write_text('model="synthetic"\\n')
+(home/'auth.json').write_text('PRIVATE-DO-NOT-EXPORT')
+for name in ['state_5.sqlite','thread_history_1.sqlite']:
+ db=sqlite3.connect(home/name); db.execute('CREATE TABLE retained (turn TEXT)'); db.execute("INSERT INTO retained VALUES ('latest')"); db.commit(); db.close()
+Path('/logs/agent/wakes/latest').mkdir(parents=True)
+Path('/logs/agent/wakes/latest/receipt.json').write_text('{"ok":false}')
+Path('/data/candidate').write_text('newer than the last graded candidate')
+'''
+    assert container.exec_run(["python", "-c", script]).exit_code == 0
+    components = terminal_components({"components": []}, "/data")
+    if capture_failure:
+        components.append({"id": "missing", "source_path": "/missing-participant"})
+    supervisor = CheckpointSupervisor(files[0], tmp_path / "terminal-points",
+        retention_enabled=False, terminal_components=components,
+        terminal_receipt=tmp_path / "terminal-preservation.json")
+    backend = RecordingDockerBackend(log_dir=tmp_path / "collected", logger=None,
+                                     oauth_proxy=True, checkpoint=supervisor, client=container.client)
+    backend.bind_registration(token)
+    backend.execution_command = "exit 2"
+    handle = DockerContainerHandle(container)
+    result = backend.exec_run_with_timeout(handle, ["/bin/bash", "-c", "exit 2"], timeout=60)
+    assert result.exit_code == 2
+    backend.cleanup_container(handle)
+    container.reload()
+    assert not container.attrs["State"]["Running"] and container.attrs["State"]["Pid"] == 0
+    receipt = json.loads((tmp_path / "terminal-preservation.json").read_text())
+    assert receipt["source_retained"] and receipt["source_stopped"]
+    assert not receipt["restore_qualified"]
+    assert receipt["capture_succeeded"] is not capture_failure
+    assert len(grades) == 1  # Saving the tail did not create an evaluation opportunity.
+    if not capture_failure:
+        point = supervisor.destination / receipt["capture_id"]
+        manifest = json.loads((point / "checkpoint.json").read_text())
+        metadata = manifest["point_metadata"]["worker_file_snapshot"]
+        assert metadata["native_cursor_qualified"] is False
+        assert (point / "components/terminal_workspace/candidate").read_text() == 'newer than the last graded candidate'
+        capsule = point / "components" / metadata["native_capsule_component_id"]
+        native = json.loads((capsule / "checkpoint.json").read_text())["point_metadata"]["native_reference"]
+        assert native["opportunity"]["next_agent"] == 2
+        assert native["opportunity"]["remaining_cooldown_seconds"] > 3500
+        assert not list(point.rglob("auth.json"))

From 410506a5748e7a0a29c8c63d69764bc72729bb8b Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sat, 7 Nov 2026 05:58:25 +0800
Subject: [PATCH 2/3] fix(benchmark): stop preserved workers before provider
 result drain

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
(cherry picked from commit 6097e14220e03d8ceb2b1df4555e59e62a64883c)
Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 benchmark/edgebench/README.md                 |  7 ++-
 benchmark/edgebench/terminal_capture.py       | 10 +++-
 benchmark/tests/test_terminal_preservation.py | 59 +++++++++++++++----
 3 files changed, 62 insertions(+), 14 deletions(-)

diff --git a/benchmark/edgebench/README.md b/benchmark/edgebench/README.md
index 0220493ce5..becdb20c59 100644
--- a/benchmark/edgebench/README.md
+++ b/benchmark/edgebench/README.md
@@ -735,10 +735,15 @@ export outcome/digest, preservation time and confirmed stopped-container state.
 The file export explicitly remains **unqualified for restoration**: a failed native
 Turn is not a completed conversation checkpoint. Both successful and failed exports
 retain the original stopped container until an operator separately qualifies a
-receiver or authorizes disposal. Compute is stopped before diagnostic collection;
+receiver or authorizes disposal. Compute is stopped before returning to provider final-archive extraction or
+grading drain, and cleanup verifies that stop again;
 a stop failure is reported without deleting the source. Missing participants, full
 storage or an uncertain writer barrier cannot fall through to force-removal.
 The provider still performs its ordinary network/proxy cleanup and grader drain.
+Its best-effort final submission archive may be absent because its extraction
+requires a running container; latest files remain in the stopped source and, on
+successful export, the terminal point. Native graded history/results remain
+independent and are not discarded with that optional archive.
 
 Inspect this receipt, the exported manifest and the exact retained container before
 planning recovery. Do not restart it, treat a file digest as receiver adoption, or
diff --git a/benchmark/edgebench/terminal_capture.py b/benchmark/edgebench/terminal_capture.py
index 9329a4cf9b..2fdd533495 100644
--- a/benchmark/edgebench/terminal_capture.py
+++ b/benchmark/edgebench/terminal_capture.py
@@ -104,8 +104,14 @@ def finish(self, execution):
             # No remote message, output, account information or token is retained.
             self.receipt["error_kind"] = type(error).__name__
         finally:
-            self.receipt["preservation_elapsed_seconds"] = time.monotonic() - started
-            self._write()
+            # Native final-archive extraction and grading drain occur before its
+            # cleanup callback. Stop here so orphaned work cannot run through
+            # that interval; later cleanup only verifies this same source.
+            try:
+                self.stop_source(supervisor.container)
+            finally:
+                self.receipt["preservation_elapsed_seconds"] = time.monotonic() - started
+                self._write()
 
     def stop_source(self, container):
         """Release compute, never remove the only source after an uncertain save."""
diff --git a/benchmark/tests/test_terminal_preservation.py b/benchmark/tests/test_terminal_preservation.py
index 382ca45d71..dc8f336eb5 100644
--- a/benchmark/tests/test_terminal_preservation.py
+++ b/benchmark/tests/test_terminal_preservation.py
@@ -21,8 +21,10 @@ def close():
         calls.append("close")
         if fault == "barrier":
             raise RuntimeError("private transport details")
+    container = SimpleNamespace(id="a" * 64, attrs={"State": {
+        "Running": False, "Paused": False, "Restarting": False, "Pid": 0}}, reload=lambda: None)
     supervisor = SimpleNamespace(destination=tmp_path, controller_file=tmp_path / "controller.json",
-        token="dummy-token", container=SimpleNamespace(id="a" * 64), close=close)
+        token="dummy-token", container=container, close=close)
     def capture(*args, **kwargs):
         calls.append("capture")
         if fault == "capture":
@@ -44,7 +46,7 @@ def test_failed_capture_preserves_source_and_never_replays(tmp_path, monkeypatch
     assert calls == (["close"] if fault == "barrier" else ["close", "capture"])
     assert receipt["execution"] == execution and receipt["restore_qualified"] is False
     assert receipt["capture_succeeded"] is (fault is None)
-    assert preservation.retain_source and receipt["source_retained"] is None
+    assert preservation.retain_source and receipt["source_retained"] is True
     assert "private" not in json.dumps(receipt)
     box = SimpleNamespace(id="a" * 64, attrs={"State": {"Running": True, "Paused": False, "Restarting": False, "Pid": 17}})
     box.reload = lambda: None
@@ -171,15 +173,19 @@ def test_cli_preservation_requires_full_native_controller_before_trial(tmp_path,
 @pytest.mark.skipif(os.environ.get("LOOPX_EDGEBENCH_DOCKER_SMOKE") != "1",
                     reason="Requires isolated Docker and native HTTP checkpoint provider")
 @pytest.mark.parametrize("capture_failure", [False, True])
-def test_real_failed_solver_preserves_tail_and_native_opportunity(controllers, writer, tmp_path, capture_failure):
+def test_real_failed_solver_preserves_tail_and_native_opportunity(controllers, writer, tmp_path, monkeypatch, capture_failure):
     from sforge.harness.backend.docker_backend import DockerContainerHandle
     from benchmark.runtime.sforge_backend import RecordingDockerBackend
     from benchmark.edgebench.checkpoint_supervisor import CheckpointSupervisor
-    from benchmark.tests.test_checkpoint_controller import register, submit
+    from benchmark.tests.test_checkpoint_controller import submit
+    from sforge.harness.agent import Agent
+    from sforge.harness.benchmark import load_benchmark
+    from sforge.harness.task_spec import make_task_spec
+    from dataclasses import replace
+    import importlib
+    native_runner = importlib.import_module("sforge.harness.run_agent")
     source, _, first, _, files, grades = controllers
     container, _ = writer
-    token = register(source, first, submission_cooldown=3600)
-    assert submit(first, token, 7).status_code == 200
     script = '''from pathlib import Path
 import sqlite3,time
 root=Path('/opt/loopx-benchmark'); control=root/'control'; control.mkdir(parents=True)
@@ -205,14 +211,45 @@ def test_real_failed_solver_preserves_tail_and_native_opportunity(controllers, w
     supervisor = CheckpointSupervisor(files[0], tmp_path / "terminal-points",
         retention_enabled=False, terminal_components=components,
         terminal_receipt=tmp_path / "terminal-preservation.json")
-    backend = RecordingDockerBackend(log_dir=tmp_path / "collected", logger=None,
+    config = source.state.judge.config
+    trial = config.log_dir / "runs/source/fixture"
+    backend = RecordingDockerBackend(log_dir=trial / "collected", logger=None,
                                      oauth_proxy=True, checkpoint=supervisor, client=container.client)
-    backend.bind_registration(token)
     backend.execution_command = "exit 2"
     handle = DockerContainerHandle(container)
-    result = backend.exec_run_with_timeout(handle, ["/bin/bash", "-c", "exit 2"], timeout=60)
-    assert result.exit_code == 2
-    backend.cleanup_container(handle)
+    # Only setup uses a precreated test-owned box. Keep the production solver,
+    # final-archive exception handling, HTTP drain, result publication and cleanup.
+    monkeypatch.setattr(backend, "image_exists", lambda image: True)
+    monkeypatch.setattr(backend, "remove_container_by_name", lambda name: None)
+    monkeypatch.setattr(backend, "create_container", lambda *args, **kwargs: handle)
+    monkeypatch.setattr(backend, "start_container", lambda handle: None)
+    monkeypatch.setattr(native_runner, "_install_tools", lambda *args, **kwargs: None)
+    monkeypatch.setattr("sforge.harness.network_isolation.cleanup_stale_chains", lambda *args: None)
+    original_drain = native_runner._wait_for_grading
+    drained = []
+    def drain(*args, **kwargs):
+        container.reload()
+        assert container.attrs["State"]["Running"] is False
+        value = original_drain(*args, **kwargs)
+        drained.append(value.status)
+        return value
+    monkeypatch.setattr(native_runner, "_wait_for_grading", drain)
+    class SyntheticAgent(Agent):
+        name, run_cmd, api_key_env = "fixture", "exit 2", "SYNTHETIC_API_KEY"
+        install_cmds = []
+    def registered(token):
+        backend.bind_registration(token)
+        assert submit(first, token, 7).status_code == 200
+    task = make_task_spec(config.tasks_dir / "fixture.json", load_benchmark(config.tasks_dir))
+    result = native_runner.run_agent(replace(task, cwd="/data", submit_paths=["candidate"]),
+        SyntheticAgent(config), config, backend, "source", timeout=60, judge_url=first,
+        eval_interval=0, disable_stop_hook=True, disable_auto_eval=True, disable_auto_resume=True,
+        internet=True, submission_cooldown=3600, host_registration_callback=registered)
+    assert result.solver_execution["segments"][0]["native_exit_code"] == 2
+    assert result.solver_execution["status"] == "nonzero_exit"
+    assert result.best_score == 7 and result.agent_submissions == 1
+    assert drained == ["complete"] and result.archive == b""
+    assert json.loads((trial / "run_history.json").read_text())["best_score"] == 7
     container.reload()
     assert not container.attrs["State"]["Running"] and container.attrs["State"]["Pid"] == 0
     receipt = json.loads((tmp_path / "terminal-preservation.json").read_text())

From 6fafca628a5d538d2aad35fbf99716fbdffbc5db Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sat, 7 Nov 2026 17:36:38 +0800
Subject: [PATCH 3/3] fix(benchmark): preserve terminal outcomes when
 diagnostic storage fails

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 benchmark/edgebench/README.md                 |  6 +++
 benchmark/edgebench/terminal_capture.py       | 21 ++++++++--
 benchmark/tests/test_terminal_preservation.py | 40 ++++++++++++++++---
 3 files changed, 57 insertions(+), 10 deletions(-)

diff --git a/benchmark/edgebench/README.md b/benchmark/edgebench/README.md
index becdb20c59..004e4e0a91 100644
--- a/benchmark/edgebench/README.md
+++ b/benchmark/edgebench/README.md
@@ -745,6 +745,12 @@ requires a running container; latest files remain in the stopped source and, on
 successful export, the terminal point. Native graded history/results remain
 independent and are not discarded with that optional archive.
 
+If the optional preservation receipt cannot be written, a redacted warning reports
+that gap; observer shutdown, one capture attempt and verified source stopping
+still proceed. The known solver exit and native grading remain authoritative.
+An unconfirmed stop still fails explicitly, even if receipt storage also fails.
+Missing receipt files never establish successful export or recovery eligibility.
+
 Inspect this receipt, the exported manifest and the exact retained container before
 planning recovery. Do not restart it, treat a file digest as receiver adoption, or
 roll back to an older point to fill a tail gap. Retained containers still occupy
diff --git a/benchmark/edgebench/terminal_capture.py b/benchmark/edgebench/terminal_capture.py
index 2fdd533495..5d6f62f959 100644
--- a/benchmark/edgebench/terminal_capture.py
+++ b/benchmark/edgebench/terminal_capture.py
@@ -5,6 +5,7 @@
 """
 from __future__ import annotations
 
+import logging
 import time
 import uuid
 from pathlib import PurePosixPath
@@ -59,6 +60,18 @@ def _write(self):
         write_json_atomic(self.receipt_path,
                           self.receipt, mode=0o600, fsync_directory=True)
 
+    def _persist_receipt(self):
+        """Optional diagnosis cannot replace the solver or stop outcome."""
+        try:
+            self._write()
+        except OSError as error:
+            self.receipt["receipt_write_error_kind"] = type(error).__name__
+            logging.getLogger(__name__).warning(
+                "Terminal preservation receipt could not be saved (%s); "
+                "source retention and verified solver/stop outcomes remain authoritative.",
+                type(error).__name__,
+            )
+
     def finish(self, execution):
         if self.receipt is not None:
             return  # Repeated cleanup cannot capture later state under this point.
@@ -80,7 +93,7 @@ def finish(self, execution):
             "capture_attempted": False, "capture_succeeded": False,
             "restore_qualified": False, "source_retained": None, "source_stopped": False}
         try:
-            self._write()
+            self._persist_receipt()
             self.supervisor.close()
             maintenance = getattr(self.supervisor, "maintenance", None)
             if maintenance is not None and maintenance.is_alive():
@@ -90,7 +103,7 @@ def finish(self, execution):
             from .checkpoint_worker import capture_worker
             from benchmark.runtime.checkpoint_handoff import CHECKPOINT_CAPTURE_SECONDS
             self.receipt["capture_attempted"] = True
-            self._write()
+            self._persist_receipt()
             request = self.supervisor.destination / "terminal-request.json"
             write_json_atomic(request, {"container_id": self.supervisor.container.id,
                 "components": self.components, "phase_deadline_path": f"{_CONTROL}/phase-deadline"},
@@ -111,7 +124,7 @@ def finish(self, execution):
                 self.stop_source(supervisor.container)
             finally:
                 self.receipt["preservation_elapsed_seconds"] = time.monotonic() - started
-                self._write()
+                self._persist_receipt()
 
     def stop_source(self, container):
         """Release compute, never remove the only source after an uncertain save."""
@@ -136,4 +149,4 @@ def stop_source(self, container):
             self.receipt["stop_error_kind"] = type(error).__name__
             raise
         finally:
-            self._write()
+            self._persist_receipt()
diff --git a/benchmark/tests/test_terminal_preservation.py b/benchmark/tests/test_terminal_preservation.py
index dc8f336eb5..99fa0df04b 100644
--- a/benchmark/tests/test_terminal_preservation.py
+++ b/benchmark/tests/test_terminal_preservation.py
@@ -1,6 +1,7 @@
 """Failed native workers retain their last source, without granting recovery."""
 from __future__ import annotations
 
+import errno
 import json
 import os
 from types import SimpleNamespace
@@ -150,12 +151,29 @@ def test_unknown_stop_state_does_not_claim_released_compute(tmp_path, monkeypatc
     assert preservation.receipt["source_stopped"] is False
 
 
-def test_receipt_disk_failure_still_arms_source_retention(tmp_path, monkeypatch):
+def test_receipt_disk_failure_keeps_solver_and_stopped_source_outcomes(tmp_path, monkeypatch, caplog):
     preservation, calls = fixture_preservation(tmp_path, monkeypatch)
-    monkeypatch.setattr(preservation, "_write", lambda: (_ for _ in ()).throw(OSError("full")))
-    with pytest.raises(OSError):
+    monkeypatch.setattr(preservation, "_write", lambda: (_ for _ in ()).throw(OSError(errno.ENOSPC, "private path")))
+    execution = {"exit_code": 2, "execution_started": True}
+    preservation.finish(execution)
+    assert preservation.retain_source and calls == ["close", "capture"]
+    assert preservation.receipt["execution"] == execution
+    assert preservation.receipt["source_stopped"] and preservation.receipt["capture_succeeded"]
+    assert preservation.receipt["receipt_write_error_kind"] == "OSError"
+    assert "could not be saved" in caplog.text and "private path" not in caplog.text
+    assert not preservation.receipt_path.exists()
+    preservation.finish(execution)
+    assert calls == ["close", "capture"]
+
+
+def test_receipt_failure_never_masks_an_unconfirmed_stop(tmp_path, monkeypatch):
+    preservation, _ = fixture_preservation(tmp_path, monkeypatch)
+    preservation.supervisor.container.attrs["State"] = {}
+    monkeypatch.setattr(preservation, "_write", lambda: (_ for _ in ()).throw(OSError(errno.ENOSPC, "full")))
+    with pytest.raises(RuntimeError, match="not confirmed stopped"):
         preservation.finish({"exit_code": 2, "execution_started": True})
-    assert preservation.retain_source and calls == []
+    assert preservation.retain_source and preservation.receipt["source_stopped"] is False
+    assert preservation.receipt["stop_error_kind"] == "RuntimeError"
 
 
 @pytest.mark.parametrize("profile", ["official", "native-goal", "heartbeat-explore"])
@@ -173,7 +191,8 @@ def test_cli_preservation_requires_full_native_controller_before_trial(tmp_path,
 @pytest.mark.skipif(os.environ.get("LOOPX_EDGEBENCH_DOCKER_SMOKE") != "1",
                     reason="Requires isolated Docker and native HTTP checkpoint provider")
 @pytest.mark.parametrize("capture_failure", [False, True])
-def test_real_failed_solver_preserves_tail_and_native_opportunity(controllers, writer, tmp_path, monkeypatch, capture_failure):
+@pytest.mark.parametrize("receipt_failure", [False, True])
+def test_real_failed_solver_preserves_tail_and_native_opportunity(controllers, writer, tmp_path, monkeypatch, capture_failure, receipt_failure):
     from sforge.harness.backend.docker_backend import DockerContainerHandle
     from benchmark.runtime.sforge_backend import RecordingDockerBackend
     from benchmark.edgebench.checkpoint_supervisor import CheckpointSupervisor
@@ -211,6 +230,10 @@ def test_real_failed_solver_preserves_tail_and_native_opportunity(controllers, w
     supervisor = CheckpointSupervisor(files[0], tmp_path / "terminal-points",
         retention_enabled=False, terminal_components=components,
         terminal_receipt=tmp_path / "terminal-preservation.json")
+    if receipt_failure:
+        def fail_receipt():
+            raise OSError(errno.ENOSPC, "synthetic receipt disk full")
+        monkeypatch.setattr(supervisor.preservation, "_write", fail_receipt)
     config = source.state.judge.config
     trial = config.log_dir / "runs/source/fixture"
     backend = RecordingDockerBackend(log_dir=trial / "collected", logger=None,
@@ -252,7 +275,12 @@ def registered(token):
     assert json.loads((trial / "run_history.json").read_text())["best_score"] == 7
     container.reload()
     assert not container.attrs["State"]["Running"] and container.attrs["State"]["Pid"] == 0
-    receipt = json.loads((tmp_path / "terminal-preservation.json").read_text())
+    if receipt_failure:
+        assert not (tmp_path / "terminal-preservation.json").exists()
+        receipt = supervisor.preservation.receipt
+        assert receipt["receipt_write_error_kind"] == "OSError"
+    else:
+        receipt = json.loads((tmp_path / "terminal-preservation.json").read_text())
     assert receipt["source_retained"] and receipt["source_stopped"]
     assert not receipt["restore_qualified"]
     assert receipt["capture_succeeded"] is not capture_failure