diff --git a/benchmark/README.md b/benchmark/README.md index d219238a4f..a97fcc0117 100644 --- a/benchmark/README.md +++ b/benchmark/README.md @@ -1364,6 +1364,25 @@ accepted round counts, cooldown and scoring remain unchanged. Omit the option for the original-budget behavior; a changed budget requires a fresh destination, not a rewrite of a running study or its original results. +When the audited remaining allowance is shorter than the captured allowance, +use `--restore-phase-remaining-cap-seconds N` (controller-only: +`--phase-remaining-cap-seconds N`). The receiving Host, native registration and +worker all use one deadline derived from the smaller of the captured allowance +plus any explicit extension, and this cap. A larger cap never grants time. +The first activation binds the cap; recovery must use the identical value and +does not refill elapsed time. Omission preserves existing receipts and behavior. +Zero or an allowance within the existing work reserve cannot start work. This +operator transport only limits time: it does not recover uncaptured work or +qualify a stale checkpoint as a complete continuation. Keep that integrity check +and the remaining-time audit separate, and use a fresh destination. + +补跑额度短于快照保留时间时，使用 `--restore-phase-remaining-cap-seconds N`； +controller 独立入口对应 `--phase-remaining-cap-seconds N`。接收端、原生注册和 +worker 共用同一个截止：取“快照余量加显式扩展”与上限的较小值。上限不能增加额度， +首次激活后重试必须保持原值和原截止；省略时保持既有行为与回执兼容。零额度或不足 +既有工作保留时间时拒绝启动。该参数只约束时间，不能补回未捕获的工作，也不能把旧 +checkpoint 认证成完整接续；仍须独立核验研究状态及剩余时间，并使用新接收目录。 + The earlier import-based worker passed a synthetic joined qualification: native session continuation, a new-Goal preparation/review, candidate work and released native feedback under the original remaining allowance. That fixture diff --git a/benchmark/edgebench/checkpoint_controller.py b/benchmark/edgebench/checkpoint_controller.py index 5b58460d7f..859a59c7b2 100644 --- a/benchmark/edgebench/checkpoint_controller.py +++ b/benchmark/edgebench/checkpoint_controller.py @@ -278,24 +278,27 @@ def _bank_import_lock(target: Path, timeout: float): def restore(controller_file: Path, *, root: Path, expected_sha256: str, run_id: str, timeout: float = 30, phase_deadline_file: Path | None = None, recover: bool = False, - phase_extension_seconds: int = 0) -> dict: + phase_extension_seconds: int = 0, phase_remaining_cap_seconds: int | None = None) -> dict: from loopx.hosts.checkpoint_retention import checkpoint_bank_use started = monotonic() with checkpoint_bank_use(root): return _restore(controller_file, root=root, expected_sha256=expected_sha256, run_id=run_id, timeout=timeout - (monotonic() - started), phase_deadline_file=phase_deadline_file, - recover=recover, phase_extension_seconds=phase_extension_seconds) + recover=recover, phase_extension_seconds=phase_extension_seconds, + phase_remaining_cap_seconds=phase_remaining_cap_seconds) def _restore(controller_file: Path, *, root: Path, expected_sha256: str, run_id: str, timeout: float, phase_deadline_file: Path | None, recover: bool, - phase_extension_seconds: int) -> dict: + phase_extension_seconds: int, phase_remaining_cap_seconds: int | None) -> dict: """Import verified inert bytes, then ask fresh native authority to admit them. An explicitly captured Host clock preserves its remaining phase duration. An operator can explicitly grant additional continuation time for a new study. The grant is bound to the first activation, never to recovery time; it changes no native cooldown, round limit, history or scoring rule. + An explicit remaining cap can only shorten that allowance; it does not + establish that an older checkpoint contains the latest research state. Does not restore a model, LoopX Goal or effect authority. An uncertain response is not replayed under a different run or capture id. """ @@ -303,6 +306,11 @@ def _restore(controller_file: Path, *, root: Path, expected_sha256: str, run_id: raise ValueError("Continuation phase extension must be a non-negative integer number of seconds") if phase_extension_seconds and phase_deadline_file is None: raise ValueError("Continuation phase extension requires the captured Host phase clock") + if phase_remaining_cap_seconds is not None: + if type(phase_remaining_cap_seconds) is not int or not 0 <= phase_remaining_cap_seconds < 2 ** 53: + raise ValueError("Continuation phase cap must be a non-negative integer number of seconds") + if phase_deadline_file is None: + raise ValueError("Continuation phase cap requires the captured Host phase clock") if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}", run_id): raise ValueError("Continuation requires one explicit native run component") controller = _controller(controller_file) @@ -327,7 +335,10 @@ def _restore(controller_file: Path, *, root: Path, expected_sha256: str, run_id: phase = _phase_budget((root / "components/host_phase_deadline").read_bytes(), native) if manifest["point_metadata"].get("host_phase_budget") != phase: raise ValueError("Host phase metadata differs from sealed deadline/native clock") - if phase["remaining_seconds"] + phase_extension_seconds <= PHASE_WORK_RESERVE_SECONDS: + remaining = phase["remaining_seconds"] + phase_extension_seconds + if phase_remaining_cap_seconds is not None: + remaining = min(remaining, phase_remaining_cap_seconds) + if remaining <= PHASE_WORK_RESERVE_SECONDS: raise ValueError("Captured Host phase has no work allowance beyond its existing reserve") phase_path = _phase_destination(phase_deadline_file, root=root, bank=bank, recover=recover) target = bank / reference["checkpoint_id"] @@ -382,7 +393,9 @@ def _restore(controller_file: Path, *, root: Path, expected_sha256: str, run_id: "bank_root": str(bank), "phase_path": str(phase_path), # Keep pre-existing zero-extension receipt bytes recoverable. **({"phase_extension_seconds": phase_extension_seconds} if phase_extension_seconds else {}), - }, phase["remaining_seconds"] + phase_extension_seconds, recover) + **({"phase_remaining_cap_seconds": phase_remaining_cap_seconds} + if phase_remaining_cap_seconds is not None else {}), + }, remaining, recover) if original_phase not in (b"", f"{phase_epoch}\n".encode()): raise ValueError("Host phase bytes differ from original activation") native_attempted = True @@ -444,6 +457,8 @@ def main(argv=None) -> None: help="Read the exact original native result and retain its original Host deadline") restore_parser.add_argument("--phase-extension-seconds", type=int, default=0, help="Explicit operator-authorized extra time for a new continuation study; recovery never refills it") + restore_parser.add_argument("--phase-remaining-cap-seconds", type=int, + help="Only reduce the captured/extended remaining allowance; bound to the first activation") args = parser.parse_args(argv) if args.command in {"capture", "capture-worker"}: source_token = _physical(args.token_file).read_text().strip() @@ -474,7 +489,8 @@ def main(argv=None) -> None: result = restore(args.controller_file, root=args.root, expected_sha256=args.expected_sha256, run_id=args.run_id, phase_deadline_file=args.phase_deadline_file, recover=args.recover, - phase_extension_seconds=args.phase_extension_seconds) + phase_extension_seconds=args.phase_extension_seconds, + phase_remaining_cap_seconds=args.phase_remaining_cap_seconds) if prior_result is not None and prior_result != result: raise ValueError("Existing private result differs from original native activation") result_file.seek(0) diff --git a/benchmark/edgebench/run.py b/benchmark/edgebench/run.py index a1feda16be..3f64199faf 100644 --- a/benchmark/edgebench/run.py +++ b/benchmark/edgebench/run.py @@ -173,6 +173,8 @@ def main(argv=None): help="Start a fresh conversation before the retained pending replan while preserving full Goal state") parser.add_argument("--restore-phase-extension-seconds", type=int, default=0, help="Explicit authorized extra continuation time; omitted preserves the checkpoint's remaining phase") + parser.add_argument("--restore-phase-remaining-cap-seconds", type=int, + help="Only reduce the warm continuation allowance to the audited remaining solver time") parser.add_argument("--eval-interval", type=int, help="Native default: 0 (no periodic evaluation); other modes use task defaults; explicit 0 disables") parser.add_argument("--submission-cooldown", type=int, @@ -243,6 +245,9 @@ def main(argv=None): if (not 0 <= args.restore_phase_extension_seconds < 2 ** 53 or (args.restore_phase_extension_seconds and not args.restore_checkpoint)): parser.error("A non-negative phase extension requires an explicit warm checkpoint continuation") + if args.restore_phase_remaining_cap_seconds is not None and ( + not 0 <= args.restore_phase_remaining_cap_seconds < 2 ** 53 or not args.restore_checkpoint): + parser.error("A non-negative phase cap requires an explicit warm checkpoint continuation") if args.restore_research_component and not args.restore_checkpoint: parser.error("Research component restoration requires an explicit warm checkpoint continuation") if args.restore_fresh_replan and not args.restore_checkpoint: @@ -410,7 +415,8 @@ def main(argv=None): if warm_start is not None: activated = restore(args.checkpoint_controller_file, root=capsule, expected_sha256=snapshot["native_capsule_sha256"], run_id=args.run_id, - phase_deadline_file=clock, phase_extension_seconds=args.restore_phase_extension_seconds) + phase_deadline_file=clock, phase_extension_seconds=args.restore_phase_extension_seconds, + phase_remaining_cap_seconds=args.restore_phase_remaining_cap_seconds) epoch = int(clock.read_text()) warm_start = replace(candidate, deadline_epoch=epoch) reference = json.loads((capsule / "checkpoint.json").read_bytes())["point_metadata"]["native_reference"] @@ -441,6 +447,8 @@ def main(argv=None): "status": "starting", "score_countable": False, **({"restore_phase_extension_seconds": args.restore_phase_extension_seconds} if args.restore_phase_extension_seconds else {}), + **({"restore_phase_remaining_cap_seconds": args.restore_phase_remaining_cap_seconds} + if args.restore_phase_remaining_cap_seconds is not None else {}), **({"restore_fresh_replan": True} if args.restore_fresh_replan else {}), **({"restored_checkpoint_sha256": warm_start.sha256, "original_deadline_epoch": warm_start.deadline_epoch, diff --git a/benchmark/tests/test_checkpoint_controller.py b/benchmark/tests/test_checkpoint_controller.py index fe6851b638..3f99714d96 100644 --- a/benchmark/tests/test_checkpoint_controller.py +++ b/benchmark/tests/test_checkpoint_controller.py @@ -348,8 +348,9 @@ def test_host_phase_and_native_opportunities_continue_without_offline_refill( assert len(destination.state.judge.tokens) == 1 and len(grades) == 2 +@pytest.mark.parametrize("cap", [None, 600, 100000]) def test_explicit_phase_extension_is_first_activation_bound_without_native_refill( - controllers, tmp_path, monkeypatch): + controllers, tmp_path, monkeypatch, cap): import requests source, destination, first, second, files, grades = controllers token = register(source, first, max_agent_submissions=3, submission_cooldown=3600) @@ -364,11 +365,17 @@ def test_explicit_phase_extension_is_first_activation_bound_without_native_refil target = tmp_path / "continued-phase" started = time.time() result = transport.restore(files[1], root=unit, expected_sha256=point["sha256"], - run_id="extended-branch", phase_deadline_file=target, phase_extension_seconds=64800) + run_id="extended-branch", phase_deadline_file=target, phase_extension_seconds=64800, + phase_remaining_cap_seconds=cap) deadline = int(target.read_text()) - assert started + retained + 64800 - 1 <= deadline <= time.time() + retained + 64800 + allowance = retained + 64800 if cap is None else min(retained + 64800, cap) + assert started + allowance - 1 <= deadline <= time.time() + allowance binding = json.loads(target.with_name(target.name + ".restore.json").read_text()) assert binding["binding"]["phase_extension_seconds"] == 64800 + if cap is None: + assert "phase_remaining_cap_seconds" not in binding["binding"] + else: + assert binding["binding"]["phase_remaining_cap_seconds"] == cap history = requests.get(second + "/api/v1/history", params={"token": result["token"], "admin_secret": destination.state.judge.admin_secret}, timeout=5).json() assert history["agent_submissions"] == 1 and history["best_score"] == 7 assert submit(second, result["token"], 8).status_code == 429 @@ -381,13 +388,18 @@ def test_explicit_phase_extension_is_first_activation_bound_without_native_refil monkeypatch.setattr(transport, "time", SimpleNamespace(time=lambda: clock)) assert transport.restore(files[1], root=unit, expected_sha256=point["sha256"], run_id="extended-branch", phase_deadline_file=target, recover=True, - phase_extension_seconds=64800) == result + phase_extension_seconds=64800, phase_remaining_cap_seconds=cap) == result assert int(target.read_text()) == deadline and info["last_agent_submit_at"] == accepted_clock for changed in (0, 64801): with pytest.raises(ValueError, match="original activation"): transport.restore(files[1], root=unit, expected_sha256=point["sha256"], run_id="extended-branch", phase_deadline_file=target, recover=True, - phase_extension_seconds=changed) + phase_extension_seconds=changed, phase_remaining_cap_seconds=cap) + for changed in ([600] if cap is None else [None, cap + 1]): + with pytest.raises(ValueError, match="original activation"): + transport.restore(files[1], root=unit, expected_sha256=point["sha256"], + run_id="extended-branch", phase_deadline_file=target, recover=True, + phase_extension_seconds=64800, phase_remaining_cap_seconds=changed) assert phase.read_bytes() == original and len(grades) == 1 assert len(destination.state.judge.tokens) == 1 @@ -407,6 +419,69 @@ def test_phase_extension_requires_a_captured_clock_before_controller_access(tmp_ expected_sha256="a" * 64, run_id="branch", phase_extension_seconds=64800) +@pytest.mark.parametrize("cap", [-1, True, 1.5, float("nan"), float("inf"), 2 ** 53]) +def test_invalid_phase_cap_rejects_before_controller_access(tmp_path, cap): + with pytest.raises(ValueError, match="non-negative integer"): + transport.restore(tmp_path / "absent-controller", root=tmp_path / "absent-point", + expected_sha256="a" * 64, run_id="branch", phase_deadline_file=tmp_path / "phase", + phase_remaining_cap_seconds=cap) + assert not (tmp_path / "phase").exists() + + +def test_phase_cap_requires_a_captured_clock_before_controller_access(tmp_path): + with pytest.raises(ValueError, match="captured Host phase clock"): + transport.restore(tmp_path / "absent-controller", root=tmp_path / "absent-point", + expected_sha256="a" * 64, run_id="branch", phase_remaining_cap_seconds=600) + + +@pytest.mark.parametrize("cap", [0, transport.PHASE_WORK_RESERVE_SECONDS]) +def test_empty_capped_allowance_cannot_activate_native(controllers, tmp_path, cap): + source, destination, first, second, files, grades = controllers + token = register(source, first) + source_phase = tmp_path / "source-phase" + source_phase.write_text(f"{int(time.time()) + 7200}\n") + unit = tmp_path / "unit" + point = transport.capture(files[0], token=token, capture_id="d" * 32, + destination=unit, phase_deadline_file=source_phase) + target = tmp_path / "phase" + with pytest.raises(ValueError, match="no work allowance"): + transport.restore(files[1], root=unit, expected_sha256=point["sha256"], run_id="branch", + phase_deadline_file=target, phase_remaining_cap_seconds=cap) + assert not target.exists() and destination.state.judge.tokens == {} and grades == [] + + +def test_cli_cap_survives_lost_activation_without_refill(controllers, tmp_path, monkeypatch): + source, destination, first, second, files, grades = controllers + token = register(source, first) + source_phase = tmp_path / "source-phase" + source_phase.write_text(f"{int(time.time()) + 7200}\n") + unit = tmp_path / "unit" + point = transport.capture(files[0], token=token, capture_id="d" * 32, + destination=unit, phase_deadline_file=source_phase) + target, result = tmp_path / "phase", tmp_path / "result" + args = ["--controller-file", str(files[1]), "restore", "--root", str(unit), + "--expected-sha256", point["sha256"], "--run-id", "branch", "--result-file", str(result), + "--phase-deadline-file", str(target), "--phase-remaining-cap-seconds", "600"] + post = transport._post + def lost(controller, route, body, timeout): + value = post(controller, route, body, timeout) + if route.endswith("/restore"): + raise TimeoutError("Lost committed activation") + return value + monkeypatch.setattr(transport, "_post", lost) + started = time.time() + with pytest.raises(TimeoutError): + transport.main(args) + binding = json.loads(target.with_name(target.name + ".restore.json").read_bytes()) + deadline = binding["deadline_epoch"] + assert started + 599 <= deadline <= time.time() + 600 + assert target.read_bytes() == b"" and result.read_bytes() == b"" + monkeypatch.setattr(transport, "time", SimpleNamespace(time=lambda: started + 120)) + transport.main(args + ["--recover"]) + assert int(target.read_text()) == deadline and json.loads(result.read_bytes())["token"] + assert len(destination.state.judge.tokens) == 1 and grades == [] + + @pytest.mark.parametrize("fault", ["missing_clock", "exhausted", "existing", "alias", "modified"]) def test_host_phase_faults_refuse_before_native_admission(controllers, tmp_path, fault): source, destination, first, second, files, grades = controllers diff --git a/benchmark/tests/test_sforge_warm_start.py b/benchmark/tests/test_sforge_warm_start.py index e000dcc043..b9e1ce4f46 100644 --- a/benchmark/tests/test_sforge_warm_start.py +++ b/benchmark/tests/test_sforge_warm_start.py @@ -14,14 +14,15 @@ from loopx.control_plane.effect_runtime import EffectRuntimeConflict, effect_runtime_result +@pytest.mark.parametrize("option", ["--restore-phase-extension-seconds", "--restore-phase-remaining-cap-seconds"]) @pytest.mark.parametrize("extension", ["-1", "64800", str(2 ** 53)]) -def test_extension_without_a_warm_entry_refuses_before_attempt_or_registration(tmp_path, extension): +def test_extension_without_a_warm_entry_refuses_before_attempt_or_registration(tmp_path, extension, option): pytest.importorskip("harbor") from benchmark.edgebench import run as entry argv = ["--task", "fixture", "--tasks-dir", str(tmp_path), "--log-dir", str(tmp_path), "--run-id", "unadmitted", "--worker", "heartbeat-explore", "--model", "fixture", "--effort", "high", "--judge-url", "http://localhost:1", - "--restore-phase-extension-seconds", extension] + option, extension] with pytest.raises(SystemExit) as rejected: entry.main(argv) assert rejected.value.code == 2 and not list(tmp_path.iterdir())