diff --git a/benchmark/edgebench/checkpoint_controller.py b/benchmark/edgebench/checkpoint_controller.py index 06246ef8c5..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,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)} 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 659c8a9015..0b18557e4e 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 # 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) 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: