fix(benchmark): cap restored phase at audited remaining allowance

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

benchmark/edgebench/checkpoint_controller.py

@@ -278,31 +278,39 @@ 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.
     """
     if type(phase_extension_seconds) is not int or not 0 <= phase_extension_seconds < 2 ** 53:
         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)

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,

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

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())