fix(benchmark): route admitted continuation clocks into automatic captures

benchmark/README.md

@@ -1555,6 +1555,16 @@ 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.
 
+After in-place continuation is admitted, the scheduler atomically publishes
+`active-phase-deadline` beside its state, rounded down from that phase's existing
+deadline. Replaying the same request republishes the same deadline; a refused
+continuation does not replace it. Both held and terminal captures automatically
+select this physical clock, including when the container is paused or stopped.
+They preserve the original `phase-deadline` file as a recovery witness. Fresh
+workers without a continuation clock keep the standard path. An explicitly
+configured supervisor `phase_deadline_path` still takes precedence; unreadable or
+invalid clocks fail capture instead of silently falling back or granting time.
+
 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 _ACTIVE_PHASE_DEADLINE, _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,6 +34,8 @@ 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")
+            _validate_request({"container_id": "0" * 64, "components": terminal_components,
+                               "phase_deadline_path": phase_deadline_path})
             self.preservation = FailurePreservation(self, terminal_components, terminal_receipt)
         self.token = None
         self.stop = threading.Event()
@@ -51,6 +56,20 @@ def bind(self, environment):
             raise ValueError("Checkpoint controller requires one native worker registration")
         self.token = environment["SFORGE_TOKEN"]
 
+    def capture_phase_deadline_path(self):
+        """Use a published continuation clock; fresh workers keep their default."""
+        if self.phase_deadline_path != _PHASE_DEADLINE:
+            return self.phase_deadline_path
+        from docker.errors import NotFound
+
+        with tempfile.TemporaryDirectory(prefix=".phase-clock-", dir=self.destination) as temporary:
+            try:
+                _archive(self.container, _ACTIVE_PHASE_DEADLINE, Path(temporary) / "clock",
+                         deadline=time.time() + 5)
+            except NotFound:
+                return _PHASE_DEADLINE
+        return _ACTIVE_PHASE_DEADLINE
+
     def start(self, container):
         if self.thread is not None or self.token is None:
             raise ValueError("Checkpoint supervisor requires one bound worker execution")
@@ -171,7 +190,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.capture_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,7 +43,7 @@ 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
 
 
@@ -106,7 +106,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": supervisor.capture_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,9 @@
 _ROOT = "/opt/loopx-benchmark"
 _SRC = f"{_ROOT}/source"
 _CONTROL = f"{_ROOT}/control"
+_PHASE_DEADLINE = f"{_CONTROL}/phase-deadline"
+_ACTIVE_PHASE_DEADLINE_NAME = "active-phase-deadline"
+_ACTIVE_PHASE_DEADLINE = f"{_CONTROL}/{_ACTIVE_PHASE_DEADLINE_NAME}"
 _CODEX_HOME = f"{_ROOT}/codex-home"
 _REGISTRY = f"{_CONTROL}/registry.json"
 _LOOPX_RUNTIME = f"{_ROOT}/state/runtime"

benchmark/runtime/scheduler.py

@@ -185,6 +185,18 @@ def hold(state: Path, reason: str) -> str:
             raise ValueError(f'Explicit idle handoff {reason}')
         return str(state)
 
+    def activate(state: Path) -> str:
+        # The retained clock is an immutable recovery witness. Publish only
+        # this already-admitted phase for both Host capture paths, with no
+        # rounding up or new scheduler/native allowance.
+        from .layout import _ACTIVE_PHASE_DEADLINE_NAME
+        from loopx.file_lock import write_bytes_atomic
+
+        clock = _physical_state(path.with_name(_ACTIVE_PHASE_DEADLINE_NAME))
+        write_bytes_atomic(clock, f"{math.floor(phase_deadline)}\n".encode(),
+                           mode=0o600, fsync_directory=True)
+        return str(state)
+
     with ExitStack() as locks:
         # The root serializes phase selection. Locking the existing phase files
         # as well keeps a running/failed phase from moving under that decision.
@@ -229,7 +241,7 @@ def hold(state: Path, reason: str) -> str:
                             != retained.get('source_proof_sha256')):
                     raise ValueError('Explicit idle handoff replay differs from its original phase')
                 verify_idle_host_replay(parent, retained, env=env)
-                return str(predecessor)  # Keep any later work or failure in this same phase.
+                return activate(predecessor)  # Keep any later work or failure in this same phase.
             if (same_phase or hashlib.sha256(predecessor.read_bytes()).hexdigest() != idle_host_state_sha256):
                 raise ValueError('Explicit idle handoff source changed; inspect the retained phase')
             from .interruption_recovery import read_idle_host
@@ -241,7 +253,7 @@ def hold(state: Path, reason: str) -> str:
             recovered.update(source_state_sha256=idle_host_state_sha256, phase_deadline_epoch=phase_deadline)
             _create_resume_state(target, predecessor, recovered=recovered,
                 replan_attempts=state.get('replan_attempts'), handoff=handoff, idle_recovery=True)
-            return str(target)
+            return activate(target)
         explicit_failed_phase = (retry_failed_state_sha256 is not None and not same_phase
             and hashlib.sha256(predecessor.read_bytes()).hexdigest() == retry_failed_state_sha256)
         if ((recovery is not None or state.get("resume_from"))
@@ -261,7 +273,7 @@ def hold(state: Path, reason: str) -> str:
                 _verify_failed_phase_replay(predecessor, state, failed_phase,
                                             states=states, digest=retry_failed_state_sha256,
                                             env=env, wake_log_dir=wake_log_dir)
-                return str(predecessor)  # Replay never erases a later failure in this phase.
+                return activate(predecessor)  # Replay never erases a later failure in this phase.
             if not explicit_failed_phase:
                 raise ValueError('Explicit failed continuation source changed; inspect the retained phase')
         if not isinstance(prior, dict):
@@ -302,7 +314,7 @@ def hold(state: Path, reason: str) -> str:
             _create_resume_state(target, predecessor, recovered=recovered,
                                  replan_attempts=state.get("replan_attempts"), handoff=handoff,
                                  failed_recovery=failed_recovery)
-            return str(target)
+            return activate(target)
         if (prior.get("status") != "stopped"
                 or prior.get("stop_exit_code") != BUDGET_EXHAUSTED_EXIT_CODE
                 or "host_failure" in prior or _host_pending(predecessor) is None
@@ -316,7 +328,7 @@ def hold(state: Path, reason: str) -> str:
         if target.is_symlink():
             raise ValueError("Resumed scheduler state must not be a symlink")
         if target.exists():
-            return str(target)  # Never replace a retry/failure in this phase.
+            return activate(target)  # Never replace a retry/failure in this phase.
         # Only a stop before native dispatch admits this shortcut. An
         # interrupted native Turn still needs its original recovery/settlement.
         # Restrict legacy receipt discovery to the source scheduler snapshot;
@@ -327,7 +339,7 @@ def hold(state: Path, reason: str) -> str:
                 or result.get("stage") != "execute"):
             return str(predecessor)
         _create_resume_state(target, predecessor, handoff=handoff)
-        return str(target)
+        return activate(target)
 
 
 def worker_command(env: dict[str, str], *, python: str, source: str,

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