diff --git a/benchmark/README.md b/benchmark/README.md index 1fb7f74167..d0f89e8224 100644 --- a/benchmark/README.md +++ b/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 diff --git a/benchmark/edgebench/run.py b/benchmark/edgebench/run.py index 03115a5ad8..8f2dbabd14 100644 --- a/benchmark/edgebench/run.py +++ b/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): diff --git a/benchmark/edgebench/terminal_capture.py b/benchmark/edgebench/terminal_capture.py index 5d6f62f959..42a45c62e2 100644 --- a/benchmark/edgebench/terminal_capture.py +++ b/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__) diff --git a/benchmark/runtime/sforge_backend.py b/benchmark/runtime/sforge_backend.py index 1c0e63f740..71568873b7 100644 --- a/benchmark/runtime/sforge_backend.py +++ b/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: diff --git a/benchmark/tests/test_official_failure_retention.py b/benchmark/tests/test_official_failure_retention.py new file mode 100644 index 0000000000..1788272fca --- /dev/null +++ b/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() diff --git a/benchmark/tests/test_terminal_preservation.py b/benchmark/tests/test_terminal_preservation.py index 99fa0df04b..496e108d51 100644 --- a/benchmark/tests/test_terminal_preservation.py +++ b/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()