Bind benchmark checkpoints to the resumed phase clock

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

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:

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"],

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"

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

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