fix(benchmark): seal resumed worker phase deadlines

benchmark/README.md

@@ -1011,10 +1011,12 @@ An explicit **SForge phase-clock participant** preserves a declared Host file's
 remaining duration alongside native opportunities. Add
 `--phase-deadline-file /private/source-phase-deadline` to capture, and
 `--phase-deadline-file /private/new-host-control/phase-deadline` to restore.
-Supply the original physical integer deadline from the admitted source Host,
-through an authorized controller read route. The destination file must be new,
-outside the artifact and native bank, and installed at the Host's existing
-phase-file route before starting its ordinary entry command.
+Supply the original physical deadline from the admitted source Host through an
+authorized controller read route. Fresh SForge phases record integer seconds.
+The continuation scheduler may record fractional seconds in its active clock.
+The destination file must be new, outside the artifact and native bank, and
+installed at the Host's existing phase-file route before starting its ordinary
+entry command.
 
 Planning input no longer projects the Host's deadline, remaining time or
 execution reserve to the model. The original task-planning packet is preserved;

benchmark/edgebench/checkpoint_controller.py

@@ -108,14 +108,12 @@ def _bound_bank(folder: Path, reference: dict) -> dict:
 
 
 def _phase_budget(raw: bytes, native: dict) -> dict:
-    """Adapt the existing SForge integer phase clock, not quota authority.
-
-    Native capture owns this time. The declared physical Host file owns its
-    deadline. Neither fact qualifies complete Host/research-state capture.
-    """
-    if len(raw) > 64 or not re.fullmatch(rb"[0-9]+\n?", raw):
-        raise ValueError("Host phase deadline must be an integer epoch file")
-    epoch = int(raw)
+    """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)
     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):

benchmark/tests/test_checkpoint_controller.py

@@ -282,14 +282,17 @@ async def installed(self, environment):
         profile="heartbeat-resume", cwd="/task", timeout_seconds=43200, task_entry="seeded-todo")
     worker.install_stop_hook(None, None, logs, None)
     worker.prepared = True
-    # The provider's own phase-file/admission shell executes. Its following
-    # model command is excluded explicitly; candidate writing below is synthetic.
-    return worker.format_run_cmd("/task.md").split(
+    # Execute the provider's actual phase shell between subscription readiness
+    # and model dispatch. Both surrounding operations are outside this fixture.
+    _, readiness, phase = worker.format_run_cmd("/task.md").partition("-m benchmark.runtime.codex_readiness; ")
+    assert readiness
+    return "set -eu; " + phase.split(
         "set +e; timeout --signal=TERM --kill-after=30 ${remaining}s env", 1)[0]
 
 
+@pytest.mark.parametrize("fraction", [None, 0.0, 0.625], ids=["fresh", "resumed", "subsecond"])
 def test_host_phase_and_native_opportunities_continue_without_offline_refill(
-        controllers, tmp_path, monkeypatch, capsys):
+        controllers, tmp_path, monkeypatch, capsys, fraction):
     import requests
     source, destination, first, second, files, grades = controllers
     token = register(source, first, max_agent_submissions=3, submission_cooldown=2)
@@ -298,14 +301,21 @@ def test_host_phase_and_native_opportunities_continue_without_offline_refill(
     # An already admitted Host has 2h left from its original 12h configured
     # trial. Fixture seeding represents that prior elapsed research, not a new run.
     source_phase = tmp_path / "source-phase"
-    source_phase.write_text(f"{int(time.time()) + 7200}\n")
+    source_epoch = int(time.time()) + 7200
+    if fraction is not None:
+        source_epoch += fraction
+    source_phase.write_text(f"{source_epoch}\n")
     source_bytes = source_phase.read_bytes()
     unit = tmp_path / "unit"
     point = transport.capture(files[0], token=token, capture_id="c" * 32,
                               destination=unit, phase_deadline_file=source_phase)
     body = json.loads((unit / "checkpoint.json").read_bytes())
-    retained = body["point_metadata"]["host_phase_budget"]["remaining_seconds"]
-    assert 7190 < retained <= 7200
+    budget = body["point_metadata"]["host_phase_budget"]
+    retained = budget["remaining_seconds"]
+    assert budget["source_deadline_epoch"] == source_epoch
+    assert retained == source_epoch - budget["captured_at_epoch"]
+    assert (unit / "components" / "host_phase_deadline").read_bytes() == source_bytes
+    assert 7190 < retained <= 7200.625
     prefix = host_phase_command(tmp_path, monkeypatch)
     control = tmp_path / "host-control"
     control.mkdir()
@@ -616,7 +626,7 @@ def lost(controller, route, body, timeout):
         assert calls == ["/api/v1/checkpoint/restore"]
 
 
-@pytest.mark.parametrize("raw", [b"", b"nan\n", b"1.5\n", b"9007199254740992\n"])
+@pytest.mark.parametrize("raw", [b"", b"nan\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)