fix(benchmark): preserve failed workers before terminal teardown

benchmark/edgebench/README.md

@@ -711,6 +711,54 @@ 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 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.
+
+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
+**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

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:

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

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

benchmark/edgebench/terminal_capture.py

@@ -0,0 +1,152 @@
+"""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 logging
+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 _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.
+        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._persist_receipt()
+            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._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"},
+                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:
+            # 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._persist_receipt()
+
+    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._persist_receipt()