Keep checkpoint capture on the admitted continuation clock

benchmark/edgebench/checkpoint_controller.py

@@ -24,6 +24,7 @@
 from urllib.request import HTTPRedirectHandler, ProxyHandler, Request, build_opener
 
 from loopx.control_plane.effect_runtime import effect_runtime_result
+from benchmark.runtime.scheduler import read_phase_deadline
 from benchmark.runtime.worker import PHASE_WORK_RESERVE_SECONDS
 
 
@@ -108,15 +109,14 @@ def _bound_bank(folder: Path, reference: dict) -> dict:
 
 
 def _phase_budget(raw: bytes, native: dict) -> dict:
-    """Read the physical Host clock at the native capture time."""
-    if len(raw) > 64 or not re.fullmatch(rb"[0-9]+(?:\.[0-9]+)?\n?", raw):
-        raise ValueError("Host phase deadline must contain epoch seconds")
-    # Fresh SForge phases write integer seconds. The continuation scheduler
-    # publishes its existing floating-point deadline, including fractions.
-    epoch = float(raw) if b"." in raw else int(raw)
+    """Preserve the admitted Host phase clock without granting new authority.
+
+    Native capture owns this time. The declared physical Host file owns its
+    deadline. Neither fact qualifies complete Host/research-state capture.
+    """
+    epoch = read_phase_deadline(raw)
     captured = native.get("captured_at_epoch")
-    if (not 0 < epoch < 2 ** 53 or type(captured) not in (int, float) or
-            not math.isfinite(captured) or captured <= 0):
+    if type(captured) not in (int, float) or not math.isfinite(captured) or captured <= 0:
         raise ValueError("Host/native phase clocks cannot be represented exactly")
     return {"component_id": "host_phase_deadline", "source_deadline_epoch": epoch,
             "captured_at_epoch": captured, "remaining_seconds": max(0, epoch - captured)}

benchmark/edgebench/terminal_capture.py

@@ -53,7 +53,7 @@ def _scheduler_components_at_barrier(container, components, stage, *, deadline):
     import re
     import docker
     from benchmark.runtime.layout import _ACTIVE_PHASE_DEADLINE
-    from benchmark.runtime.scheduler import phase_state_name
+    from benchmark.runtime.scheduler import phase_state_name, read_phase_deadline
     from .checkpoint_worker import _archive
 
     root = PurePosixPath(_CONTROL, "scheduler-state.json")
@@ -63,7 +63,7 @@ def _scheduler_components_at_barrier(container, components, stage, *, deadline):
         clock = _archive(container, _ACTIVE_PHASE_DEADLINE, stage / "active-clock", deadline=deadline)
     except docker.errors.NotFound:
         return components  # A fresh worker has no admitted continuation.
-    current = root.with_name(phase_state_name(root, float(clock.read_bytes())))
+    current = root.with_name(phase_state_name(root, read_phase_deadline(clock.read_bytes())))
     chain = []
     while current != root:
         if (current in chain or current.parent != root.parent

benchmark/runtime/scheduler.py

@@ -23,6 +23,18 @@ def phase_work_reserve(mode: str) -> int:
     return 0 if mode == "heartbeat" else PHASE_WORK_RESERVE_SECONDS
 
 
+def read_phase_deadline(raw: bytes) -> int | float:
+    """Read the initial integer epoch or the scheduler's continued phase clock."""
+    if len(raw) > 64 or not re.fullmatch(rb"[0-9]+(?:\.[0-9]+)?\n?", raw):
+        raise ValueError("Host phase deadline must be an integer or canonical decimal epoch file")
+    body = raw.removesuffix(b"\n")
+    deadline = float(body) if b"." in body else int(body)
+    if not 0 < deadline < 2 ** 53 or (
+            isinstance(deadline, float) and body != repr(deadline).encode("ascii")):
+        raise ValueError("Host phase deadline cannot be represented exactly")
+    return deadline
+
+
 def phase_state_name(root: PurePath, phase_deadline: float) -> str:
     """Name a scheduler phase by the finite clock that owns its allowance."""
     if (isinstance(phase_deadline, bool) or not isinstance(phase_deadline, (int, float))

benchmark/tests/test_checkpoint_controller.py

@@ -23,6 +23,12 @@
 from benchmark.edgebench.source import active_source_receipt
 
 
+@pytest.mark.parametrize("clock", [b"nan\n", b"9007199254740992.0\n", b"2000.25000000000000001\n"])
+def test_phase_budget_rejects_invalid_or_rounded_clock(clock):
+    with pytest.raises(ValueError, match="Host phase deadline"):
+        transport._phase_budget(clock, {"captured_at_epoch": 1000})
+
+
 @contextmanager
 def server(app):
     import uvicorn
@@ -267,20 +273,25 @@ def reject_non_json_constant(value):
 
 
 def host_phase_command(tmp_path, monkeypatch):
-    """Real command builder; only installation/inference are outside this oracle."""
+    """Real phase admission shell, isolated from installation and model access."""
     pytest.importorskip("harbor")
     from sforge.harness.config import SForgeConfig
     from benchmark.runtime.sforge import BenchmarkCodex, SForgeWorker
 
     async def installed(self, environment):
         pass
     monkeypatch.setattr(BenchmarkCodex, "install", installed)
-    monkeypatch.setenv("CODEX_AUTH_JSON_PATH", "/synthetic-not-read")
     logs = tmp_path / "phase-worker"
     logs.mkdir()
-    worker = SForgeWorker(SForgeConfig(agent_model="fixture", agent_effort="xhigh"),
+    # The model-free prefix must not try to inspect a desktop login. The
+    # custom endpoint uses the same phase shell and is never contacted.
+    config = SForgeConfig(agent_model="fixture", agent_effort="xhigh",
+        agent_api_base_url="http://127.0.0.1:1/v1", agent_api_key="fixture-not-used")
+    backend = Mock()
+    backend.exec_run.return_value = SimpleNamespace(exit_code=0, output=config.agent_api_base_url)
+    worker = SForgeWorker(config,
         profile="heartbeat-resume", cwd="/task", timeout_seconds=43200, task_entry="seeded-todo")
-    worker.install_stop_hook(None, None, logs, None)
+    worker.install_stop_hook(backend, None, logs, None)
     worker.prepared = True
     # Execute the provider's actual phase shell between subscription readiness
     # and model dispatch. Both surrounding operations are outside this fixture.
@@ -298,8 +309,8 @@ def test_host_phase_and_native_opportunities_continue_without_offline_refill(
     token = register(source, first, max_agent_submissions=3, submission_cooldown=2)
     assert submit(first, token, 7).status_code == 200
     before = copy.deepcopy((source.state.judge.tokens, source.state.judge.submissions, source.state.judge.run_history))
-    # An already admitted Host has 2h left from its original 12h configured
-    # trial. Fixture seeding represents that prior elapsed research, not a new run.
+    # An admitted continuation retains about 2h from its original 12h trial.
+    # The scheduler publishes a decimal deadline for this phase.
     source_phase = tmp_path / "source-phase"
     source_epoch = int(time.time()) + 7200
     if fraction is not None:
@@ -341,7 +352,8 @@ def test_host_phase_and_native_opportunities_continue_without_offline_refill(
     result = subprocess.run(["/bin/bash", "-c", shell +
         f"printf '8' > '{candidate}'; printf '%s' \"$remaining\""],
         env={"PATH": str(bin) + ":/usr/bin:/bin"}, capture_output=True, text=True, check=True)
-    assert 7180 < int(result.stdout) <= 7200 and candidate.read_bytes() == b"8"
+    assert int(result.stdout) == int(phase.read_bytes()) - int(clock)
+    assert candidate.read_bytes() == b"8"
     history = requests.get(second + "/api/v1/history", params={"token": fresh, "admin_secret": destination.state.judge.admin_secret}, timeout=5).json()
     assert history["agent_submissions"] == 1 and history["best_score"] == 7
     assert submit(second, fresh, 8).status_code == 429
@@ -626,7 +638,7 @@ def lost(controller, route, body, timeout):
         assert calls == ["/api/v1/checkpoint/restore"]
 
 
-@pytest.mark.parametrize("raw", [b"", b"nan\n", b"1.5.0\n", b"0.0\n", b"9007199254740992\n"])
+@pytest.mark.parametrize("raw", [b"", b"nan\n", b"1.50\n", b"1.5.0\n", b"0.0\n", b"9007199254740992\n"])
 def test_invalid_host_epoch_cannot_create_native_point(controllers, tmp_path, raw):
     source, destination, first, second, files, grades = controllers
     token = register(source, first)
@@ -640,6 +652,24 @@ def test_invalid_host_epoch_cannot_create_native_point(controllers, tmp_path, ra
     assert destination.state.judge.tokens == {} and grades == []
 
 
+def test_expired_fractional_epoch_seals_without_work_allowance(controllers, tmp_path):
+    source, destination, first, _, files, grades = controllers
+    token = register(source, first)
+    phase = tmp_path / "source-phase"
+    phase.write_bytes(b"1.5\n")
+    unit = tmp_path / "unit"
+    point = transport.capture(files[0], token=token, capture_id="f" * 32,
+                              destination=unit, phase_deadline_file=phase)
+    manifest = json.loads((unit / "checkpoint.json").read_bytes())
+    budget = manifest["point_metadata"]["host_phase_budget"]
+    assert budget["source_deadline_epoch"] == 1.5 and budget["remaining_seconds"] == 0
+    assert (unit / "components" / "host_phase_deadline").read_bytes() == phase.read_bytes()
+    with pytest.raises(ValueError, match="no work allowance"):
+        transport.restore(files[1], root=unit, expected_sha256=point["sha256"], run_id="expired",
+                          phase_deadline_file=tmp_path / "destination-phase")
+    assert destination.state.judge.tokens == {} and grades == []
+
+
 def test_native_only_point_cannot_acquire_a_later_host_budget(controllers, tmp_path):
     source, destination, first, second, files, grades = controllers
     token = register(source, first)

benchmark/tests/test_checkpoint_worker.py

@@ -133,7 +133,7 @@ def test_supervisor_captures_only_the_prepared_worker_boundary(controllers, writ
     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()); "
+            f"Path({continued_clock!r}).write_text(str(int(Path('/phase').read_text()) + .25) + '\\n'); "
             "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",
@@ -213,7 +213,12 @@ def with_old_point(*args, **kwargs):
         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
+            raw = container.exec_run(["cat", continued_clock]).output
+            assert (capsule / "components/host_phase_deadline").read_bytes() == raw
+            budget = json.loads((capsule / "checkpoint.json").read_bytes())["point_metadata"]["host_phase_budget"]
+            assert budget["source_deadline_epoch"] == float(raw)
+            assert budget["source_deadline_epoch"] % 1 == .25
+            assert budget["remaining_seconds"] == float(raw) - budget["captured_at_epoch"]
             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()

benchmark/tests/test_scheduler_phase_resume.py

@@ -20,6 +20,7 @@ def test_scheduler_continuation_clock_reaches_both_host_capture_paths(tmp_path,
     """Actual guest scheduler + archive IO; only native sealing is substituted."""
     import docker
     from benchmark.edgebench import checkpoint_supervisor as host
+    from benchmark.edgebench.checkpoint_controller import _phase_budget
     from benchmark.edgebench.checkpoint_worker import _archive
     from benchmark.runtime.layout import _PHASE_DEADLINE
 
@@ -45,7 +46,7 @@ def test_scheduler_continuation_clock_reaches_both_host_capture_paths(tmp_path,
 receipt=root/'wakes/scheduler/receipt.json';receipt.parent.mkdir(parents=True)
 receipt.write_text(json.dumps({'budget_exhausted':True,'host_invoked':False,'mode':'heartbeat','stage':'execute'}))
 if not continued:state.unlink()
-os.environ.update(LOOPX_PHASE_DEADLINE_EPOCH=str(int(time.time())+800 if continued else original),LOOPX_AGENT_ID='agent',
+os.environ.update(LOOPX_PHASE_DEADLINE_EPOCH=str(int(time.time())+800.25 if continued else original),LOOPX_AGENT_ID='agent',
  LOOPX_WAKE_LOG_DIR=str(root/'wakes'),LOOPX_EXECUTION_MODE='heartbeat',LOOPX_CLI='/bin/true',
  LOOPX_REGISTRY='/registry.json',LOOPX_RUNTIME_ROOT='/runtime',LOOPX_GOAL_ID='fixture')
 sys.argv=['scheduler','--python',sys.executable,'--source','/source','--state-file',str(state),'--host-timeout','30']
@@ -71,18 +72,24 @@ def test_scheduler_continuation_clock_reaches_both_host_capture_paths(tmp_path,
         def capture(*args, **kwargs):
             request = json.loads(kwargs["request_file"].read_text())
             path = _archive(container, request["phase_deadline_path"], tmp_path / f"read-{len(observed)}")
-            observed.append((request["phase_deadline_path"], float(path.read_text()) - time.time()))
-            supervisor.stop.set()
+            try:
+                budget = _phase_budget(path.read_bytes(), {"captured_at_epoch": time.time()})
+                observed.append((request["phase_deadline_path"], budget))
+            finally:
+                supervisor.stop.set()
             return {"research_point_qualified": False}
 
         monkeypatch.setattr(host, "capture_worker", capture)
         monkeypatch.setattr("benchmark.edgebench.checkpoint_worker.capture_worker", capture)
         supervisor._run(container)
+        reply = json.loads((destination / ("b" * 32 + ".json")).read_bytes())
+        assert reply["ok"], reply
         supervisor.finish_solver({"execution_started": True, "exit_code": 2})
         expected_path = "/opt/loopx-benchmark/control/active-phase-deadline" if continued else _PHASE_DEADLINE
         assert [path for path, _ in observed] == [expected_path, expected_path]
-        limit = 800 if continued else 400
-        assert all(limit - 100 < remaining <= limit for _, remaining in observed)
+        limit = 800.25 if continued else 400
+        assert all(limit - 100 < budget["remaining_seconds"] <= limit for _, budget in observed)
+        assert all(budget["source_deadline_epoch"] % 1 == (.25 if continued else 0) for _, budget in observed)
         original = _archive(container, _PHASE_DEADLINE, tmp_path / "original")
         assert original.read_bytes() == original_bytes
     finally: