Reuse the committed quota capture in native benchmark preflight

benchmark/runtime/checkpoint.py

@@ -229,26 +229,32 @@ def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: st
                "--runtime-root", env["LOOPX_RUNTIME_ROOT"], "quota", "should-run",
                "--goal-id", env["LOOPX_GOAL_ID"], "--agent-id", env["LOOPX_AGENT_ID"],
                "--runtime-profile", "generic_cli", "--turn-instance-id", receipt["turn_id"],
-               "--available-capability", "shell", "--available-capability", "filesystem_write",
-               "--decision-output-root", str(capture_root)]
+               "--available-capability", "shell", "--available-capability", "filesystem_write"]
     native_context = env.get("LOOPX_CONTEXT_DELIVERY") == "on" and bool(env.get("LOOPX_CODEX_CONTEXT_DIR"))
     if native_context:
-        # Capture the complete Goal now, but let the actual native tool select
-        # full/delta after SessionStart. Do not consume or invalidate continuity
-        # while the Host is preparing a checkpoint outside the model session.
-        command += ["--context-delivery", "on", "--codex-context-dir", env["LOOPX_CODEX_CONTEXT_DIR"],
+        # The native reader selects full/delta after SessionStart. Ask the
+        # existing preparation route for the committed capture reference now.
+        capture_directory = capture_root / "decision-original"
+        command += ["--decision-output-dir", str(capture_directory), "--prepared-capture-reference",
+                    "--context-delivery", "on", "--codex-context-dir", env["LOOPX_CODEX_CONTEXT_DIR"],
                     "--defer-context-presentation"]
+    else:
+        command += ["--decision-output-root", str(capture_root)]
     guard_run = subprocess.run(command, cwd=env["LOOPX_PROJECT"], env=env, capture_output=True,
                                text=True, timeout=120, check=True)
     projected = json.loads(guard_run.stdout)
     captures = list(capture_root.glob("decision-*/decision.json"))
     if len(captures) != 1:
         raise ValueError("checkpoint requires exactly one captured original quota decision; do not replay the guard")
-    raw_guard = captures[0].read_bytes()
+    source_path = captures[0]
+    raw_guard = source_path.read_bytes()
+    guard_digest = hashlib.sha256(raw_guard).hexdigest() if native_context else None
     guard = json.loads(raw_guard)
     if (not isinstance(guard, dict) or guard.get("ok") is not True or
             not isinstance(projected, dict) or projected.get("ok") is not True):
         raise ValueError("checkpoint requires a successful original quota guard")
+    if native_context and projected.get("capture_sha256") != guard_digest:
+        raise ValueError("checkpoint prepared capture differs from the original guard; do not replay the guard")
     # A previous preflight on this wake may already have allowed model work,
     # even if its optional attempt marker (or final receipt) could not be saved.
     # Such reentry can verify an existing point, but cannot capture a later one.
@@ -258,17 +264,9 @@ def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: st
     if native_context:
         from .context_delivery import preflight_context_instruction
 
-        # A repeated preflight can replace wake-level paths. Bind both native
-        # read commands to the exact CLI view of this particular guard.
-        view_bytes = guard_run.stdout.encode("utf-8")
-        source_path = captures[0]
-        if view_bytes != raw_guard:
-            source_path = capture_root / "checkpoint-start-view.json"
-            with source_path.open("xb") as stream:
-                os.fchmod(stream.fileno(), 0o600)
-                stream.write(view_bytes)
-        reading = preflight_context_instruction(env, source_path,
-            hashlib.sha256(view_bytes).hexdigest(), receipt["turn_id"], guide_in_host=guide_in_host)
+        assert guard_digest is not None
+        reading = preflight_context_instruction(env, source_path, guard_digest,
+            receipt["turn_id"], guide_in_host=guide_in_host)
     else:
         view_path = guard_path.with_name("checkpoint-start-view.json")
         view_path.write_text(guard_run.stdout, encoding="utf-8")

tests/control_plane/test_benchmark_checkpoint_admission.py

@@ -78,7 +78,19 @@ def rendered(command):
     assert env.get("FIXTURE_BOUND_WORK") and guard["heartbeat_receipt"]["closeout_required"] is True
     Path("checkpoint-closeout-guard.json").write_text(json.dumps(guard))
 else:
-    assert started == 1 and "resume" in sys.argv and sys.argv[-2] == "benchmark-original-session"
+    import argparse
+    native = argparse.ArgumentParser()
+    native.add_argument("-c", action="append")
+    native.add_argument("-C")
+    native.add_argument("--skip-git-repo-check", action="store_true")
+    native.add_argument("--model")
+    native.add_argument("--json", action="store_true")
+    native.add_argument("entry", choices=["exec"])
+    native.add_argument("operation", choices=["resume"])
+    native.add_argument("session")
+    native.add_argument("input")
+    resumed = native.parse_intermixed_args(sys.argv[1:])
+    assert (started, resumed.session, resumed.input) == (1, "benchmark-original-session", "-")
     guard = json.loads(Path(env["FIXTURE_HELD_QUOTA"]).read_text())
     assert env["LOOPX_TURN"] == guard["heartbeat_receipt"]["turn_instance_id"]
     assert env["LOOPX_TURN"] != env["FIXTURE_OUTER_TURN"]
@@ -99,12 +111,31 @@ def rendered(command):
 Path("checkpoint-business.json").write_text(json.dumps({"sum":sum(json.loads(Path("retained-input.json").read_text()))}))
 evidence = "checkpoint-business:" + hashlib.sha256(Path("checkpoint-business.json").read_bytes()).hexdigest()
 vision = Path("checkpoint-vision.json").resolve()
-vision.write_text(json.dumps({"schema_version":"goal_vision_replan_contract_v0","agent_id":env["LOOPX_AGENT_ID"],
+vision_packet = {"schema_version":"goal_vision_replan_contract_v0","agent_id":env["LOOPX_AGENT_ID"],
     "state":"active", "vision_patch":{"acceptance_summary":"Validate the retained inputs and keep remaining validation work.",
         "advancement_policy":"as_needed"}, "path_delta":{"schema_version":"goal_path_delta_v0","outcome":"continue",
         "prior_assumption":"The retained inputs still require a concrete validation probe.",
-        "observed_reality":"The retained inputs sum to five; this does not validate the other open tasks.",
-        "retained":["The remaining independent validation work"],"evidence_refs":[evidence]}}))
+        "observed_reality":"The retained inputs sum to five; the other work remains open.",
+        "retained":["The remaining independent validation work"],"evidence_refs":[evidence]}}
+control = (guard.get("replan_action_packet") or {}).get("stride_control")
+if control:
+    decision = control["minimal_example"].copy()
+    decision.update(basis="uncertain_evidence", action="verify", axes={
+        "work_slice": {"plan":"Validate the remaining fixture input in one work wake.",
+                       "reason":"The retained result qualifies only the original input."},
+        "route_distance": {"plan":"Keep the deterministic sum and compare an independent result.",
+                           "reason":"The original result matched the independent sum."},
+        "validation_scope": {"plan":"Check the new input and separately verify the retained result.",
+                             "reason":"A new check must preserve the original accepted evidence."}},
+        next_probe={"scope":"One remaining input and its validation.",
+                    "prediction":"The independent and retained arithmetic agree.",
+                    "adoption_criterion":"Both sums match the fixture oracle.",
+                    "cost_bound":"One work wake and its validation.",
+                    "minimum_evidence":"Input, result bytes and independent arithmetic."})
+    if "explore_review" in decision:
+        decision["explore_review"]["reason"] = "The synthetic scope uses the retained input and exact sum evidence."
+    vision_packet.update(state="vision_patch_proposed", stride_decision=decision)
+vision.write_text(json.dumps(vision_packet))
 actions = guard["interaction_contract"]["cli_channel"]["next_cli_actions"]
 refresh = next(command for command in actions if "refresh-state" in command)
 for old,new in [("<validated_progress>","retained_inputs_validated"),
@@ -119,8 +150,11 @@ def rendered(command):
 else:
     refresh_args += ["--agent-vision-json",str(vision)]
 updated = cli(*refresh_args)
-spend = rendered(next(command for command in actions if "spend-slot" in command))
-spent, replay = cli(*spend), cli(*spend)
+if "--settle-turn" in refresh_args:
+    spent, replay = updated, cli(*refresh_args)
+else:
+    spend = rendered(updated["settlement_owed"]["command"])
+    spent, replay = cli(*spend), cli(*spend)
 Path("checkpoint-settlement.json").write_text(json.dumps({"refresh":updated,"spend":spent,"replay":replay}))
 '''
 

tests/control_plane/test_progressive_context_entrypoints.py

@@ -50,7 +50,23 @@ def test_progressive_activation_does_not_require_native_continuity(profile):
 
 @pytest.mark.parametrize("host", [(), ("--runtime-profile", "generic_cli"), ("--trae_app",), ("--antigravity_app",)])
 def test_generic_read_only_cli_defaults_progressive_without_admitting_a_turn(acceptance_goal, host):
-    workspace, document, cli, run = acceptance_goal
+    import os
+    import sys
+
+    from tests.source_cli import run_source_cli
+
+    workspace, document, cli, _ = acceptance_goal
+    registry, runtime = workspace.parent / "registry.json", workspace.parent / "runtime"
+    environment = {**os.environ, "LOOPX_PYTHON": sys.executable}
+    environment.pop("LOOPX_CONTEXT_DELIVERY", None)
+
+    def guard(*arguments):
+        # The generic acceptance fixture runs machine-facing raw CLI reads.
+        # This case exercises the omitted model-facing option instead.
+        response = run_source_cli(["--registry", str(registry), "--runtime-root", str(runtime),
+            "--format", "json", *arguments], env=environment, check=False, timeout=30)
+        return response.returncode, json.loads(response.stdout)
+
     for i in range(8):
         document.write_text(json.dumps({"outcome_id": "export", "owner_agent_id": "agent-a",
             "todo_ids": ["todo_export"], "objective": "Deliver the export.",
@@ -62,7 +78,7 @@ def test_generic_read_only_cli_defaults_progressive_without_admitting_a_turn(acc
         assert code == 0, result
     before = cli("inspect")[1]
     args = ("quota", "should-run", "--goal-id", "goal-acceptance", "--agent-id", "agent-a", *host)
-    code, packet = run(*args)
+    code, packet = guard(*args)
     assert code == 0, packet
     required = required_document(packet)
     assert "heartbeat_receipt" not in required, "plain observation must not invent a work Turn"
@@ -74,7 +90,7 @@ def test_generic_read_only_cli_defaults_progressive_without_admitting_a_turn(acc
     assert "Stop before publication." in json.dumps(required)
     assert "Prior supporting observation" not in json.dumps(required)
     assert "Prior supporting observation" in json.dumps(source)
-    code, off = run(*args, "--context-delivery", "off")
+    code, off = guard(*args, "--context-delivery", "off")
     assert code == 0 and "context_details" not in off and "reference_file" not in off
     assert "Prior supporting observation" in json.dumps(off)
     assert cli("inspect")[1] == before, "presentation must not mutate acceptance or its provider revision"

tests/control_plane/test_research_checkpoint_capture.py

@@ -31,6 +31,66 @@ def forbidden(*args, **kwargs):
     assert failed["checkpoint"]["captured"] is False
 
 
+@pytest.mark.skipif(os.name != "posix", reason="native benchmark uses the POSIX source launcher")
+@pytest.mark.parametrize("changed_reference", [False, True])
+def test_native_preflight_uses_the_committed_capture(tmp_path, monkeypatch, changed_reference):
+    import shlex
+    import subprocess
+    import sys
+
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    project, runtime, registry = cli._write_fixture(tmp_path)
+    cli._initialize_fixture_authority(project, runtime, provider="sqlite")
+    cli_path = Path(__file__).resolve().parents[2] / "scripts/loopx"
+    turn = "wake-prepared-native-context"
+    logs = tmp_path / "logs"
+    wakes = logs / "wakes"
+    wake = wakes / turn
+    wake.mkdir(parents=True)
+    context = tmp_path / "context"
+    context.mkdir(mode=0o700)
+    request = tmp_path / "request.json"
+    request.write_text(json.dumps({"trigger": "fresh_replan", "destination": str(logs / "checkpoints"),
+        "components": [{"id": "fixture", "source_path": str(project)}], "point_metadata": {}}))
+    env = {**os.environ, "LOOPX_PROJECT": str(project), "LOOPX_CLI": str(cli_path),
+        "LOOPX_PYTHON": sys.executable, "LOOPX_REGISTRY": str(registry), "LOOPX_RUNTIME_ROOT": str(runtime),
+        "LOOPX_GOAL_ID": cli.GOAL_ID, "LOOPX_AGENT_ID": cli.AGENT_ID, "LOOPX_WAKE_LOG_DIR": str(wakes),
+        "LOOPX_CHECKPOINT_REQUEST": str(request), "LOOPX_CONTEXT_DELIVERY": "on",
+        "LOOPX_CODEX_CONTEXT_DIR": str(context), "LOOPX_EXECUTION_MODE": "heartbeat"}
+    observed = []
+    actual_run = subprocess.run
+
+    def observe_guard(argv, **kwargs):
+        result = actual_run(argv, **kwargs)
+        if "should-run" in argv:
+            assert "--prepared-capture-reference" in argv and "--defer-context-presentation" in argv
+            response = json.loads(result.stdout)
+            observed.append(response)
+            if changed_reference:
+                return subprocess.CompletedProcess(argv, result.returncode,
+                    json.dumps({**response, "capture_sha256": "0" * 64}), result.stderr)
+        return result
+
+    monkeypatch.setattr("benchmark.runtime.checkpoint.subprocess.run", observe_guard)
+    receipt = {"turn_id": turn, "mode": "heartbeat", "stage": "execute"}
+    if changed_reference:
+        with pytest.raises(ValueError, match="prepared capture differs"):
+            capture_before_replan(env, receipt, "original work instructions")
+        assert not (wake / "checkpoint-start-guard.json").exists()
+    else:
+        body = capture_before_replan(env, receipt, "original work instructions")
+        command = shlex.split(next(line for line in body.splitlines() if " quota context-project " in line))
+        source = Path(command[command.index("--capture-file") + 1])
+        assert source.name == "decision.json"
+        assert command[command.index("--capture-sha256") + 1] == observed[0]["capture_sha256"]
+        assert hashlib.sha256(source.read_bytes()).hexdigest() == observed[0]["capture_sha256"]
+        assert json.loads(source.read_bytes())["heartbeat_receipt"]["turn_instance_id"] == turn
+        assert not list(wake.glob("checkpoint-start-*/checkpoint-start-view.json"))
+        assert body.endswith("original work instructions")
+    assert len(observed) == 1 and "interaction_contract" not in observed[0]
+    assert cli._heartbeat_receipt_count(runtime, turn) == 1
+
+
 def test_closed_budget_never_reads_the_request_or_replays_work(monkeypatch):
     def forbidden(*args, **kwargs):
         raise AssertionError("closed budget must not read sources")