Pin the native first read and deliver the same-Turn selection route

benchmark/runtime/checkpoint.py

@@ -194,7 +194,8 @@ def _mark_completed_capture_attempt(destination: Path, effect_id: str, request_s
     return _write_checkpoint_attempt(marker, value)
 
 
-def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: str) -> str:
+def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: str, *,
+                          guide_in_host: bool = False) -> str:
     """Preload the original guard once, then capture before invoking the model.
 
     The typed owner selects only an admitted replan identity. The prior wake's
@@ -230,7 +231,8 @@ def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: st
                "--runtime-profile", "generic_cli", "--turn-instance-id", receipt["turn_id"],
                "--available-capability", "shell", "--available-capability", "filesystem_write",
                "--decision-output-root", str(capture_root)]
-    if env.get("LOOPX_CONTEXT_DELIVERY") == "on" and env.get("LOOPX_CODEX_CONTEXT_DIR"):
+    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.
@@ -253,16 +255,26 @@ def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: st
     preflight_seen = guard_path.exists()
     guard_path.write_bytes(raw_guard)
     guard_path.chmod(0o600)
-    view_path = guard_path.with_name("checkpoint-start-view.json")
-    view_path.write_text(guard_run.stdout, encoding="utf-8")
-    view_path.chmod(0o600)
-    reading = (f"Read the complete CLI view at {view_path} before work, then follow its interaction_contract "
-        "and the next step in the replan review when present. ")
-    if env.get("LOOPX_CONTEXT_DELIVERY") == "on" and env.get("LOOPX_CODEX_CONTEXT_DIR"):
+    if native_context:
         from .context_delivery import preflight_context_instruction
 
-        reading = preflight_context_instruction(env, view_path,
-            hashlib.sha256(view_path.read_bytes()).hexdigest(), receipt["turn_id"])
+        # 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)
+    else:
+        view_path = guard_path.with_name("checkpoint-start-view.json")
+        view_path.write_text(guard_run.stdout, encoding="utf-8")
+        view_path.chmod(0o600)
+        reading = (f"Read the complete CLI view at {view_path} before work, then follow its interaction_contract "
+            "and the next step in the replan review when present. ")
     preloaded_body = ("Host checkpoint preflight has already executed this wake's original quota guard once. "
         + reading +
         "Do not repeat the initial guard below; use its selection or recovery command when required.\n\n" + body)

benchmark/runtime/context_delivery.py

@@ -12,7 +12,8 @@
 from pathlib import Path
 
 
-def preflight_context_instruction(env: dict[str, str], path: Path, digest: str, turn_id: str) -> str:
+def preflight_context_instruction(env: dict[str, str], path: Path, digest: str, turn_id: str, *,
+                                  guide_in_host: bool = False) -> str:
     """Delay presentation of the original preflight until its native tool call.
 
     Keep the original source/hash. The product's lifecycle owner chooses the
@@ -28,7 +29,7 @@ def preflight_context_instruction(env: dict[str, str], path: Path, digest: str,
         "--turn-instance-id", turn_id, "--context-delivery", "on", "--runtime-profile", "generic_cli", "--codex-context-dir", env["LOOPX_CODEX_CONTEXT_DIR"]]
     recovery_index = [*prefix, "context-read", *source, "--list"]
     return ("Present that same immutable packet with:\n```sh\n" + shlex.join(project)
-        + "\n```\n" + CONTEXT_CAPTURE_NAVIGATION_GUIDANCE
+        + "\n```\n" + ("" if guide_in_host else CONTEXT_CAPTURE_NAVIGATION_GUIDANCE)
         + "If the projection command itself fails, recover necessary facts from the same original capture. "
         "When pointers are unknown, discover them with:\n```sh\n" + shlex.join(recovery_index)
         + "\n```\nUse the same capture's generic reader and selectors for the facts the current decision needs. "

benchmark/runtime/worker.py

@@ -639,7 +639,10 @@ def run_once(env: dict[str, str]) -> dict:
         if stage == "execute" and execution.mode == "heartbeat" and env.get("LOOPX_CHECKPOINT_REQUEST"):
             from benchmark.runtime.checkpoint import capture_before_replan
 
-            body = capture_before_replan(env, receipt, body)
+            # The prepared Codex exec session carries the complete read guide
+            # in developer instructions, including after compaction.
+            body = capture_before_replan(env, receipt, body,
+                guide_in_host=not execution.native_goal and receipt.get("context_transport", {}).get("available") is True)
         if phase_work_deadline is not None and time.time() >= phase_work_deadline:
             receipt.update(ok=True, budget_exhausted=True, host_invoked=False)
             return receipt

benchmark/tests/test_checkpoint_handoff.py

@@ -580,7 +580,8 @@ def basis(method, request, *, large_local_snapshot):
 
 
 @pytest.mark.parametrize("native_participant", [False, True])
-def test_preflight_defers_history_only_when_capture_will_verify_it(tmp_path, monkeypatch, native_participant):
+@pytest.mark.parametrize("native_context", [False, True])
+def test_preflight_defers_history_only_when_capture_will_verify_it(tmp_path, monkeypatch, native_participant, native_context):
     from types import SimpleNamespace
     from benchmark.runtime import checkpoint
     wake = tmp_path / "wakes" / "turn"
@@ -593,8 +594,11 @@ def guard(command, **kwargs):
         from pathlib import Path
         captured = Path(command[command.index("--decision-output-root") + 1]) / "decision-original"
         captured.mkdir()
-        (captured / "decision.json").write_text(json.dumps({"ok": True}))
-        return SimpleNamespace(stdout=json.dumps({"ok": True}))
+        original = json.dumps({"ok": True, "source": captured.parent.name})
+        (captured / "decision.json").write_text(original)
+        (captured / "decision.json").chmod(0o600)
+        return SimpleNamespace(stdout=original if native_participant else
+                               json.dumps({"ok": True, "view": captured.parent.name}))
     monkeypatch.setattr(checkpoint.subprocess, "run", guard)
     monkeypatch.setattr(checkpoint, "effect_runtime_result", lambda *a, **k: {"capture": False, "reason": "no_fresh_replan"})
     observed = []
@@ -606,8 +610,30 @@ def gate(*args, full_history):
         "LOOPX_CLI": "loopx", "LOOPX_PROJECT": str(tmp_path), "LOOPX_RUNTIME_ROOT": str(tmp_path),
         "LOOPX_REGISTRY": str(tmp_path / "registry"), "LOOPX_GOAL_ID": "goal", "LOOPX_AGENT_ID": "agent"}
     receipt = {"turn_id": "turn", "mode": "heartbeat", "stage": "execute"}
-    assert checkpoint.capture_before_replan(env, receipt, "current work").endswith("current work")
+    if native_context:
+        env.update(LOOPX_CONTEXT_DELIVERY="on", LOOPX_CODEX_CONTEXT_DIR=str(tmp_path / "context"))
+    body = checkpoint.capture_before_replan(env, receipt, "current work")
+    assert body.endswith("current work")
+    view = wake / "checkpoint-start-view.json"
     assert observed == [not native_participant]
+    if native_context:
+        pattern = ("checkpoint-start-*/decision-original/decision.json" if native_participant else
+                   "checkpoint-start-*/checkpoint-start-view.json")
+        original, = wake.glob(pattern)
+        assert str(original) in body and hashlib.sha256(original.read_bytes()).hexdigest() in body
+        first_bytes = original.read_bytes()
+        if native_participant:
+            assert not list(wake.glob("checkpoint-start-*/checkpoint-start-view.json"))
+        else:
+            assert json.loads(first_bytes) == {"ok": True, "view": original.parent.name}
+        assert original.stat().st_mode & 0o077 == 0
+        resumed = checkpoint.capture_before_replan(env, receipt, "resume")
+        second, = set(wake.glob(pattern)) - {original}
+        assert str(second) in resumed and hashlib.sha256(second.read_bytes()).hexdigest() in resumed
+        assert original.read_bytes() == first_bytes != second.read_bytes()
+        assert not view.exists()
+    else:
+        assert str(view) in body and json.loads(view.read_text())["ok"]
     assert receipt["checkpoint"]["reason"] == "no_fresh_replan"
 
 

benchmark/tests/test_context_delivery.py

@@ -8,10 +8,14 @@
 import subprocess
 import sys
 
+import pytest
+
+from loopx.control_plane.turn_driver.decision_capture import CONTEXT_CAPTURE_NAVIGATION_GUIDANCE
 from benchmark.runtime.context_delivery import preflight_context_instruction
 
 
-def test_projection_failure_keeps_original_capture_selectively_readable(tmp_path, monkeypatch):
+@pytest.mark.parametrize("guide_in_host", [False, True], ids=["standalone", "prepared-host"])
+def test_projection_failure_keeps_original_capture_selectively_readable(tmp_path, monkeypatch, guide_in_host):
     capture = tmp_path / "captured decision.json"
     source = {
         "ok": True,
@@ -56,8 +60,9 @@ def test_projection_failure_keeps_original_capture_selectively_readable(tmp_path
         },
         capture,
         digest,
-        "synthetic-turn",
+        "synthetic-turn", guide_in_host=guide_in_host,
     )
+    assert (CONTEXT_CAPTURE_NAVIGATION_GUIDANCE in instruction) == (not guide_in_host)
     commands = [
         shlex.split(command)
         for command in re.findall(r"```sh\n(.*?)\n```", instruction, re.S)
@@ -78,7 +83,7 @@ def invoke(argv):
     assert refused_payload["ok"] is False
     assert refused_payload["error_code"] == "quota_decision_transport_failed"
     assert "--list" in shlex.split(refused_payload["recovery"]["command"])
-    assert "Choose the sections needed for the current decision" in refused_payload["instruction"]
+    assert CONTEXT_CAPTURE_NAVIGATION_GUIDANCE.strip() in refused_payload["instruction"]
     assert "do not drain unrelated optional material" in refused_payload["instruction"]
     assert "Follow next_command through eof" not in refused_payload["instruction"]