diff --git a/benchmark/README.md b/benchmark/README.md index 1fb7f74167..a9420407af 100644 --- a/benchmark/README.md +++ b/benchmark/README.md @@ -1533,6 +1533,11 @@ including any later failure; it cannot clear the failure or extend its deadline. A later explicit repair must bind that new failure separately. Existing operator and account authority remain required; a digest is provenance, not permission. +For in-place continuation, bind the checkpoint supervisor's `phase_deadline_path` +to the same new immutable clock supplied to the resumed scheduler. Both held +and terminal captures use that clock. Preserve the stopped phase's clock at +its original path. Fresh workers retain the standard path. + The explicit failed-phase path additionally uses the installed optional native Codex provider's existing availability policy (`loopx-codex-provider-routing`). It reads `account/read(refreshToken=false)` and `account/rateLimits/read` in a diff --git a/benchmark/edgebench/checkpoint_supervisor.py b/benchmark/edgebench/checkpoint_supervisor.py index 4341ac3132..55d060310e 100644 --- a/benchmark/edgebench/checkpoint_supervisor.py +++ b/benchmark/edgebench/checkpoint_supervisor.py @@ -12,15 +12,18 @@ from pathlib import Path from loopx.control_plane.effect_runtime import EffectRuntimeRemoteError +from benchmark.runtime.layout import _PHASE_DEADLINE from .checkpoint_controller import _controller -from .checkpoint_worker import _archive, capture_worker, qualify_export_storage +from .checkpoint_worker import _archive, _guest_path, _validate_request, capture_worker, qualify_export_storage class CheckpointSupervisor: def __init__(self, controller_file: Path, destination: Path, retention_policy: Path | None = None, *, retention_enabled: bool = True, terminal_components: list[dict[str, str]] | None = None, - terminal_receipt: Path | None = None): + terminal_receipt: Path | None = None, phase_deadline_path: str = _PHASE_DEADLINE): + _guest_path(phase_deadline_path) + self.phase_deadline_path = phase_deadline_path _controller(controller_file) # Fail before native registration. qualify_export_storage(destination) if not retention_enabled and retention_policy is not None: @@ -31,7 +34,10 @@ def __init__(self, controller_file: Path, destination: Path, retention_policy: P 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) + _validate_request({"container_id": "0" * 64, "components": terminal_components, + "phase_deadline_path": phase_deadline_path}) + self.preservation = FailurePreservation(self, terminal_components, terminal_receipt, + phase_deadline_path=phase_deadline_path) self.token = None self.stop = threading.Event() self.thread = None @@ -171,7 +177,7 @@ def _run(self, container): request = stage / "request.json" request.write_text(json.dumps({"container_id": container.id, "components": value["components"], - "phase_deadline_path": "/opt/loopx-benchmark/control/phase-deadline", + "phase_deadline_path": self.phase_deadline_path, "handoff": {"source_path": remote, "sha256": digest}})) remaining = value["deadline_epoch"] - time.time() if remaining <= 0: diff --git a/benchmark/edgebench/terminal_capture.py b/benchmark/edgebench/terminal_capture.py index 5d6f62f959..3e87b517d3 100644 --- a/benchmark/edgebench/terminal_capture.py +++ b/benchmark/edgebench/terminal_capture.py @@ -11,7 +11,7 @@ from pathlib import PurePosixPath from loopx.file_lock import write_json_atomic -from benchmark.runtime.layout import _CODEX_HOME, _CONTROL, _LOOPX_RUNTIME, _REGISTRY +from benchmark.runtime.layout import _CODEX_HOME, _CONTROL, _LOOPX_RUNTIME, _PHASE_DEADLINE, _REGISTRY def terminal_components(declaration: dict, workspace: str) -> list[dict[str, str]]: @@ -43,16 +43,17 @@ def terminal_components(declaration: dict, workspace: str) -> list[dict[str, str # 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"}) + "phase_deadline_path": _PHASE_DEADLINE}) return components class FailurePreservation: """One terminal attempt; a stopped source remains until explicit disposal.""" - def __init__(self, supervisor, components, receipt_path): + def __init__(self, supervisor, components, receipt_path, *, phase_deadline_path=_PHASE_DEADLINE): self.supervisor = supervisor self.components = components self.receipt_path = receipt_path + self.phase_deadline_path = phase_deadline_path self.receipt = None self.retain_source = False @@ -106,7 +107,7 @@ def finish(self, execution): 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"}, + "components": self.components, "phase_deadline_path": self.phase_deadline_path}, mode=0o600, fsync_directory=True) point = capture_worker(supervisor.controller_file, token=supervisor.token, capture_id=self.receipt["capture_id"], diff --git a/benchmark/runtime/layout.py b/benchmark/runtime/layout.py index f8dec8ceee..66df49713f 100644 --- a/benchmark/runtime/layout.py +++ b/benchmark/runtime/layout.py @@ -3,6 +3,7 @@ _ROOT = "/opt/loopx-benchmark" _SRC = f"{_ROOT}/source" _CONTROL = f"{_ROOT}/control" +_PHASE_DEADLINE = f"{_CONTROL}/phase-deadline" _CODEX_HOME = f"{_ROOT}/codex-home" _REGISTRY = f"{_CONTROL}/registry.json" _LOOPX_RUNTIME = f"{_ROOT}/state/runtime" diff --git a/benchmark/tests/test_checkpoint_worker.py b/benchmark/tests/test_checkpoint_worker.py index d0160a92da..2d540c1855 100644 --- a/benchmark/tests/test_checkpoint_worker.py +++ b/benchmark/tests/test_checkpoint_worker.py @@ -115,7 +115,7 @@ def test_archive_keeps_private_directory_modes_without_special_bits(writer, tmp_ @pytest.mark.parametrize("fault", [None, "source_moved", "expired", "read_race", "late_native", - "retention", "retention_default", "retention_off", "retention_pin", "retention_failure"]) + "retention", "retention_default", "retention_off", "retention_pin", "retention_failure", "continued_clock"]) def test_supervisor_captures_only_the_prepared_worker_boundary(controllers, writer, tmp_path, monkeypatch, fault): from loopx.control_plane.effect_runtime import effect_runtime_result from benchmark.edgebench.checkpoint_worker import _archive @@ -130,6 +130,11 @@ def test_supervisor_captures_only_the_prepared_worker_boundary(controllers, writ "Path('/logs/agent/checkpoint-controller').mkdir(parents=True); " "Path('/opt/loopx-benchmark/control').mkdir(parents=True); " "Path('/opt/loopx-benchmark/control/phase-deadline').write_bytes(Path('/phase').read_bytes())"]).exit_code == 0 + continued_clock = "/opt/loopx-benchmark/control/continued-phase-deadline" + if fault == "continued_clock": + assert container.exec_run(["python", "-c", "from pathlib import Path; " + f"Path({continued_clock!r}).write_bytes(Path('/phase').read_bytes()); " + "Path('/opt/loopx-benchmark/control/phase-deadline').write_text('1')"]).exit_code == 0 prepared = _archive(container, "/candidate", tmp_path / "prepared") probe = effect_runtime_result("runtime.checkpoint_artifacts", {"action": "probe", "components": [{"id": "code", "source_path": str(prepared)}]}) @@ -186,7 +191,8 @@ def with_old_point(*args, **kwargs): return original_capture(*args, **kwargs) monkeypatch.setattr(checkpoint_supervisor, "capture_worker", with_old_point) supervisor = CheckpointSupervisor(files[0], tmp_path / "points", retention_policy=policy_file, - retention_enabled=fault != "retention_off") + retention_enabled=fault != "retention_off", + **({"phase_deadline_path": continued_clock} if fault == "continued_clock" else {})) supervisor.bind({"SFORGE_TOKEN": token}) supervisor.start(container) try: @@ -196,15 +202,19 @@ def with_old_point(*args, **kwargs): supervisor.close() response = json.loads((tmp_path / "points" / (nonce + ".json")).read_bytes()) assert response["ok"] is (fault in {None, "read_race", "retention", "retention_default", - "retention_off", "retention_pin", "retention_failure"}) + "retention_off", "retention_pin", "retention_failure", "continued_clock"}) unit = tmp_path / "points" / nonce - if fault in {None, "read_race", "retention", "retention_default", "retention_off", "retention_pin", "retention_failure"}: + if response["ok"]: manifest = json.loads((unit / "checkpoint.json").read_bytes()) metadata = manifest["point_metadata"] assert metadata["synthetic_boundary"] is True assert metadata["worker_file_snapshot"]["prepared_source_verified"] is True assert (unit / "components/code").read_text() == "retained" assert response["capture"]["research_point_qualified"] is False + if fault == "continued_clock": + capsule = unit / "components" / metadata["worker_file_snapshot"]["native_capsule_component_id"] + assert (capsule / "components/host_phase_deadline").read_bytes() == container.exec_run(["cat", "/phase"]).output + assert container.exec_run(["cat", "/opt/loopx-benchmark/control/phase-deadline"]).output == b"1" if fault in {"retention", "retention_default"}: assert not (tmp_path / "points/old").exists() receipt = json.loads((tmp_path / "points/retention-last.json").read_bytes()) diff --git a/benchmark/tests/test_terminal_preservation.py b/benchmark/tests/test_terminal_preservation.py index 99fa0df04b..2227f4838b 100644 --- a/benchmark/tests/test_terminal_preservation.py +++ b/benchmark/tests/test_terminal_preservation.py @@ -2,7 +2,9 @@ from __future__ import annotations import errno +import io import json +import tarfile import os from types import SimpleNamespace @@ -176,6 +178,12 @@ def test_receipt_failure_never_masks_an_unconfirmed_stop(tmp_path, monkeypatch): assert preservation.receipt["stop_error_kind"] == "RuntimeError" +def _read_retained_clock(container, name): + with io.BytesIO(b"".join(container.get_archive("/opt/loopx-benchmark/control/" + name)[0])) as raw: + with tarfile.open(fileobj=raw) as archive: + return archive.extractfile(name).read() + + @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") @@ -208,7 +216,8 @@ def test_real_failed_solver_preserves_tail_and_native_opportunity(controllers, w 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/'phase-deadline').write_text('1') +(control/'continued-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) @@ -229,7 +238,8 @@ def test_real_failed_solver_preserves_tail_and_native_opportunity(controllers, w 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") + terminal_receipt=tmp_path / "terminal-preservation.json", + phase_deadline_path="/opt/loopx-benchmark/control/continued-phase-deadline") if receipt_failure: def fail_receipt(): raise OSError(errno.ENOSPC, "synthetic receipt disk full") @@ -292,6 +302,8 @@ def registered(token): 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"] + assert (capsule / "components/host_phase_deadline").read_bytes() == _read_retained_clock(container, "continued-phase-deadline") + assert _read_retained_clock(container, "phase-deadline") == b"1" 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