From 5ce63978b4cff5b6b75982bbcdc4301155b71e87 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 12 Nov 2026 04:35:19 +0800 Subject: [PATCH 1/2] Keep checkpoint capture on the admitted continuation clock Periodic and terminal checkpoint capture must read the same phase clock after scheduler continuation. Accept the decimal epoch published by the scheduler without rounding it or changing the receiver budget; preserve the legacy integer phase file and reject invalid clocks before native capture. Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/edgebench/checkpoint_controller.py | 10 +++--- benchmark/edgebench/terminal_capture.py | 4 +-- benchmark/runtime/scheduler.py | 12 +++++++ benchmark/tests/test_checkpoint_controller.py | 35 +++++++++++++------ benchmark/tests/test_checkpoint_worker.py | 9 +++-- .../tests/test_scheduler_phase_resume.py | 17 ++++++--- 6 files changed, 62 insertions(+), 25 deletions(-) diff --git a/benchmark/edgebench/checkpoint_controller.py b/benchmark/edgebench/checkpoint_controller.py index 859a59c7b2..a842f1882f 100644 --- a/benchmark/edgebench/checkpoint_controller.py +++ b/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,17 +109,14 @@ 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. + """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. """ - 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) + 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)} diff --git a/benchmark/edgebench/terminal_capture.py b/benchmark/edgebench/terminal_capture.py index 320775df5b..3d9ed03914 100644 --- a/benchmark/edgebench/terminal_capture.py +++ b/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 diff --git a/benchmark/runtime/scheduler.py b/benchmark/runtime/scheduler.py index 0723d449e3..7a39919f2d 100644 --- a/benchmark/runtime/scheduler.py +++ b/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)) diff --git a/benchmark/tests/test_checkpoint_controller.py b/benchmark/tests/test_checkpoint_controller.py index 3f99714d96..cf016eae40 100644 --- a/benchmark/tests/test_checkpoint_controller.py +++ b/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,7 +273,7 @@ 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 @@ -275,12 +281,17 @@ def host_phase_command(tmp_path, monkeypatch): 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 # The provider's own phase-file/admission shell executes. Its following # model command is excluded explicitly; candidate writing below is synthetic. @@ -295,17 +306,20 @@ 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_phase.write_text(f"{int(time.time()) + 7200}\n") + source_deadline = int(time.time()) + 7200.25 + source_phase.write_text(f"{source_deadline!r}\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"] + assert budget["source_deadline_epoch"] == source_deadline + retained = budget["remaining_seconds"] + assert 7190 < retained <= 7200.25 prefix = host_phase_command(tmp_path, monkeypatch) control = tmp_path / "host-control" control.mkdir() @@ -331,7 +345,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 diff --git a/benchmark/tests/test_checkpoint_worker.py b/benchmark/tests/test_checkpoint_worker.py index 2d540c1855..27001e2abb 100644 --- a/benchmark/tests/test_checkpoint_worker.py +++ b/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() diff --git a/benchmark/tests/test_scheduler_phase_resume.py b/benchmark/tests/test_scheduler_phase_resume.py index e535689490..6ff48ba5cb 100644 --- a/benchmark/tests/test_scheduler_phase_resume.py +++ b/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: From 70f0930b1c9fcbe3980e1c29e52bf3c21ed858f2 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 13 Nov 2026 02:08:38 +0800 Subject: [PATCH 2/2] test(benchmark): seal expired fractional phase deadlines Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/tests/test_checkpoint_controller.py | 20 ++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) diff --git a/benchmark/tests/test_checkpoint_controller.py b/benchmark/tests/test_checkpoint_controller.py index daec8e28df..0b18557e4e 100644 --- a/benchmark/tests/test_checkpoint_controller.py +++ b/benchmark/tests/test_checkpoint_controller.py @@ -638,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) @@ -652,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)