Retain Official benchmark workers after confirmed failure

benchmark/README.md

@@ -1397,6 +1397,20 @@ and synthetic grading. It does **not** qualify scientific research utility,
 matched strategy results, or the frontend restore journey. Actual task/profile
 admission and measured continuation outcomes remain required.
 
+### Preserve an interrupted Official control
+
+For `--worker official --feedback native --eval-interval 0`, the explicit
+`--preserve-failed-worker` option keeps the original Docker worker after a
+confirmed final solver failure. Native auto-resume, scoring, and final archive
+collection keep their ordinary behavior. The retained worker is stopped and
+verified by exact identity; a failed stop observation or receipt write never
+falls through to deletion. Without the option, ordinary cleanup still applies.
+
+The private terminal receipt records source retention only. It does not
+export an authentication file, create a joint checkpoint, or authorize
+a resume. Recover the original evaluator, session, and finite remaining
+budget separately before treating later work as part of this control.
+
 ### Continue an interrupted Host in its retained filesystem
 
 The explicit `benchmark.runtime.scheduler` budget-continuation entrypoint can

benchmark/edgebench/run.py

@@ -165,7 +165,7 @@ def main(argv=None):
     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")
+                        help="Retain an Official worker after confirmed failure, or capture and retain a failed heartbeat worker with 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,
@@ -274,8 +274,9 @@ 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")
+    retain_official_failure = args.preserve_failed_worker and args.worker == "official" and args.feedback == "native"
+    if args.preserve_failed_worker and not (handoff or retain_official_failure):
+        parser.error("Failed-worker preservation requires native Official feedback or 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:
@@ -480,6 +481,7 @@ def main(argv=None):
                                      blind_api_endpoint=blind_endpoint, feedback=feedback,
                                      docker_host_bridge=args.docker_host_bridge,
                                      research_sidecar=research_sidecar,
+                                     retain_failed_native_worker=retain_official_failure,
                                      **({"checkpoint": checkpoint} if checkpoint is not None else {}))
     for image in (task.work_image_key, task.judge_image_key):
         if not backend.image_exists(image):

benchmark/edgebench/terminal_capture.py

@@ -128,25 +128,47 @@ def finish(self, execution):
 
     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
+            _stop_retained_worker(container, self.receipt)
         finally:
             self._persist_receipt()
+
+
+def _stop_retained_worker(container, receipt):
+    if container.id != 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)
+        receipt.update(source_retained=True, source_stopped=stopped)
+        if not stopped:
+            raise RuntimeError("Preserved worker is not confirmed stopped")
+    except Exception as error:
+        receipt["stop_error_kind"] = type(error).__name__
+        raise
+
+
+def retain_returned_native_worker(container, *, evidence: dict, receipt_path):
+    """Preserve the original Official worker after its confirmed final failure."""
+    receipt = {**evidence, "observed_at_epoch": time.time(),
+               "capture_attempted": False, "capture_succeeded": False,
+               "restore_qualified": False, "source_retained": None, "source_stopped": False}
+    logger = logging.getLogger(__name__)
+    try:
+        _stop_retained_worker(container, receipt)
+    except Exception as error:
+        receipt.setdefault("stop_error_kind", type(error).__name__)
+        logger.warning("Native failed worker stop is unconfirmed (%s); retaining the source.", type(error).__name__)
+    finally:
+        try:
+            write_json_atomic(receipt_path, receipt, mode=0o600, fsync_directory=True)
+        except OSError as error:
+            logger.warning("Native failed worker receipt unavailable (%s); retaining the source.", type(error).__name__)

benchmark/runtime/sforge_backend.py

@@ -19,14 +19,21 @@
 class RecordingDockerBackend(DockerBackend):
     def __init__(self, *, log_dir: Path, logger, oauth_proxy: bool = False,
                  blind_api_endpoint: tuple[str, int] | None = None, feedback=None, checkpoint=None, client=None,
-                 docker_host_bridge: bool = False, research_sidecar: bool = False):
+                 docker_host_bridge: bool = False, research_sidecar: bool = False,
+                 retain_failed_native_worker: bool = False):
         if not isinstance(docker_host_bridge, bool):
             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")
+        if type(retain_failed_native_worker) is not bool or (
+            retain_failed_native_worker and (checkpoint is not None or feedback is not None or blind_api_endpoint is not None)
+        ):
+            raise ValueError("Native failure retention requires an explicit native-only worker")
         self.research_sidecar = research_sidecar
+        self.retain_failed_native_worker = retain_failed_native_worker
+        self.native_failure_evidence = None
         super().__init__(client)
         self.docker_host_bridge = docker_host_bridge
         self.log_dir, self.logger = log_dir, logger
@@ -93,6 +100,9 @@ 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)]
         preservation = getattr(self.checkpoint, "preservation", None)
+        retain_native = is_solver and getattr(self, "retain_failed_native_worker", False)
+        if retain_native:
+            self.native_failure_evidence = None  # A new dispatch supersedes the previous return.
         if is_solver and self.checkpoint is not None and preservation is None:
             self.checkpoint.start(handle.raw)
         try:
@@ -108,6 +118,18 @@ def exec_run_with_timeout(self, handle, cmd, timeout=60, **kwargs):
                 self.execution_receipt = dict(exit_code=result.exit_code,
                     execution_started=execution_started,
                     timed_out=result.timed_out, elapsed_seconds=result.elapsed_seconds)
+                if retain_native and execution_started:
+                    native_code = getattr(result, "native_exit_code", None)
+                    stop = kwargs.get("shutdown_event")
+                    if (getattr(result, "terminal_observed", None) is True
+                            and type(native_code) is int and native_code > 0
+                            and result.exit_code == native_code and result.timed_out is False
+                            and (stop is None or not stop.is_set())):
+                        self.native_failure_evidence = {
+                            "container_id": handle.id,
+                            "execution": {**self.execution_receipt,
+                                          "native_exit_code": native_code, "terminal_observed": True},
+                        }
                 target = self.log_dir.parent / "execution-receipt.json"
                 target.parent.mkdir(parents=True, exist_ok=True)
                 pending = target.with_suffix(".tmp")
@@ -154,7 +176,15 @@ def start_feedback(self, handle):
         self.feedback.start(self, handle)
 
     def cleanup_container(self, handle, logger=None):
+        native_failure = getattr(self, "native_failure_evidence", None) if handle is not None else None
         try:
+            if native_failure is not None:
+                from benchmark.edgebench.terminal_capture import retain_returned_native_worker
+
+                # Native cleanup runs once after auto-resumes and grading drain.
+                # Do not stop or preserve on any intermediate segment return.
+                retain_returned_native_worker(handle.raw, evidence=native_failure,
+                    receipt_path=self.log_dir.parent / "terminal-preservation.json")
             preservation = getattr(self.checkpoint, "preservation", None)
             if preservation is not None and preservation.retain_source:
                 preservation.stop_source(handle.raw)
@@ -221,7 +251,7 @@ def cleanup_container(self, handle, logger=None):
                 preservation = getattr(self.checkpoint, "preservation", None)
                 if preservation is not None and preservation.retain_source:
                     preservation.stop_source(handle.raw)
-                else:
+                elif native_failure is None:
                     super().cleanup_container(handle, logger)
             finally:
                 if self.feedback is not None:

benchmark/tests/test_official_failure_retention.py

@@ -0,0 +1,105 @@
+"""Native Official retention runs at the final provider cleanup boundary."""
+
+from contextlib import suppress
+from dataclasses import replace
+from importlib import import_module
+import io
+import json
+import os
+import tarfile
+import uuid
+
+import pytest
+
+from benchmark.tests import test_checkpoint_controller as native_fixtures
+
+controllers = native_fixtures.controllers
+
+
+@pytest.mark.skipif(os.environ.get("LOOPX_EDGEBENCH_DOCKER_SMOKE") != "1",
+                    reason="Requires isolated Docker and native HTTP evaluator")
+@pytest.mark.parametrize("retain,final_exit,receipt_failure", [
+    (True, 23, False), (False, 23, False), (True, 0, False), (True, 23, True),
+])
+def test_official_final_failure_preserves_only_the_owned_stopped_source(
+    controllers, monkeypatch, retain, final_exit, receipt_failure,
+):
+    import docker
+    from benchmark.runtime.sforge_backend import RecordingDockerBackend
+    from sforge.harness.agent import Agent
+    from sforge.harness.backend.docker_backend import DockerContainerHandle
+    from sforge.harness.benchmark import load_benchmark
+    from sforge.harness.task_spec import make_task_spec
+
+    native_runner = import_module("sforge.harness.run_agent")
+    source, _, judge_url, _, _, grades = controllers
+    config = source.state.judge.config
+    trial = config.log_dir / "runs/source/fixture"
+    client = docker.from_env(timeout=15)
+    image = os.environ.get("LOOPX_CHECKPOINT_WRITER_IMAGE", "python:3.12-slim")
+    client.images.get(image)  # Never pull an unqualified test image.
+    container = client.containers.run(image, ["tail", "-f", "/dev/null"],
+        detach=True, name="loopx-official-retention-" + uuid.uuid4().hex[:12],
+        network_disabled=True, mem_limit="64m", nano_cpus=250_000_000)
+    try:
+        assert container.exec_run(["mkdir", "-p", "/data"]).exit_code == 0
+        backend = RecordingDockerBackend(log_dir=trial / "collected", logger=None,
+            oauth_proxy=True, client=client, retain_failed_native_worker=retain)
+        handle = DockerContainerHandle(container)
+        if receipt_failure:
+            def fail_receipt(*args, **kwargs):
+                raise OSError("synthetic receipt storage failure")
+            monkeypatch.setattr("benchmark.edgebench.terminal_capture.write_json_atomic", fail_receipt)
+        # Only container creation and tool setup are preinstalled. Use the
+        # provider's real exec, auto-resume, grading drain and final 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)
+
+        class SyntheticOfficial(Agent):
+            name, run_cmd, resume_cmd, api_key_env = "fixture", "solver", "resume", "FIXTURE_API_KEY"
+            install_cmds = []
+
+            def format_run_cmd(self, prompt_path, *, resume=False, **kwargs):
+                cmd = (f"test -f /data/unsubmitted && exit {final_exit}" if resume else
+                       "printf 'solver-state' > /data/unsubmitted; printf 'candidate' > /data/candidate; sleep 1.15; exit 9")
+                backend.execution_command = cmd
+                return cmd
+
+        def registered(token):
+            assert native_fixtures.submit(judge_url, 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"]),
+            SyntheticOfficial(config), config, backend, "source", timeout=60, judge_url=judge_url,
+            eval_interval=0, disable_stop_hook=True, disable_auto_eval=True, disable_auto_resume=False,
+            internet=True, submission_cooldown=3600, host_registration_callback=registered)
+        segments = result.solver_execution["segments"]
+        assert [(s["resume"], s["native_exit_code"]) for s in segments] == [(False, 9), (True, final_exit)]
+        assert result.best_score == 7 and result.agent_submissions == 1 and len(grades) == 1
+        assert json.loads((trial / "grading_status.json").read_text())["status"] == "complete"
+        receipt_path = trial / "terminal-preservation.json"
+        if retain and final_exit:
+            container.reload()
+            assert container.attrs["State"]["Running"] is False and container.attrs["State"]["Pid"] == 0
+            if receipt_failure:
+                assert not receipt_path.exists()
+            else:
+                receipt = json.loads(receipt_path.read_text())
+                assert receipt["container_id"] == container.id and receipt["source_stopped"] and receipt["source_retained"]
+                assert receipt["execution"]["native_exit_code"] == final_exit and receipt["execution"]["terminal_observed"]
+                assert not receipt["capture_attempted"] and not receipt["restore_qualified"]
+            stream, _ = container.get_archive("/data/unsubmitted")
+            with tarfile.open(fileobj=io.BytesIO(b"".join(stream))) as archive:
+                assert archive.extractfile("unsubmitted").read() == b"solver-state"
+        else:
+            assert not receipt_path.exists()
+            with pytest.raises(docker.errors.NotFound):
+                container.reload()
+    finally:
+        with suppress(docker.errors.NotFound):
+            container.remove(force=True)
+        client.close()

benchmark/tests/test_terminal_preservation.py

@@ -176,15 +176,17 @@ def test_receipt_failure_never_masks_an_unconfirmed_stop(tmp_path, monkeypatch):
     assert preservation.receipt["stop_error_kind"] == "RuntimeError"
 
 
-@pytest.mark.parametrize("profile", ["official", "native-goal", "heartbeat-explore"])
-def test_cli_preservation_requires_full_native_controller_before_trial(tmp_path, profile):
+@pytest.mark.parametrize("profile,feedback", [
+    ("official", "best-only"), ("native-goal", "native"), ("heartbeat-explore", "native"),
+])
+def test_cli_preservation_rejects_unowned_native_feedback_and_checkpoint_routes(tmp_path, profile, feedback):
     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"])
+              "--feedback", feedback, "--judge-url", "http://127.0.0.1:9999", "--preserve-failed-worker"])
     assert error.value.code == 2 and not (tmp_path / "runs").exists()