From d0c7145987a097895880ec3800722477f3421126 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Thu, 5 Nov 2026 15:56:46 +0800
Subject: [PATCH 01/13] Reuse capture-bound navigation for the first
 progressive read

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 loopx/cli_commands/quota_capture.py           |  68 +++++++++--
 .../quota/progressive_context.ts              |  21 +++-
 loopx/control_plane/quota/turn_envelope.py    |   3 +
 loopx/control_plane/quota/turn_envelope.ts    |  13 +-
 .../turn_driver/codex_context.py              |  39 +++++-
 .../test_context_delivery_cli.py              |   6 +-
 .../control_plane/test_context_first_read.py  | 113 +++++++++++++++++-
 .../test_quota_decision_capture.py            |   2 +-
 .../test_saved_context_project.py             |  54 ++++++---
 9 files changed, 270 insertions(+), 49 deletions(-)

diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py
index 2641c6bf77..75d9487944 100644
--- a/loopx/cli_commands/quota_capture.py
+++ b/loopx/cli_commands/quota_capture.py
@@ -19,6 +19,7 @@
 from ..control_plane.quota.error_codes import QuotaCommandValidationError, QuotaDecisionTransportError
 from ..control_plane.turn_driver.decision_capture import (
     context_index_fields as _context_index_fields,
+    context_selection_bytes as _context_selection_bytes,
     context_selection_view as _context_selection_view,
     validate_handoff_socket, CONTEXT_CAPTURE_READ_GUIDANCE, CONTEXT_CAPTURE_PAGE_GUIDANCE, context_capture_read_command,
     CONTEXT_CAPTURE_NAVIGATION_GUIDANCE,
@@ -331,9 +332,16 @@ def reference(value: dict[str, Any], *, navigation: bool = False) -> dict[str, s
         result["continuity_basis"] = view["continuity_basis"]
     if "first_read_command" in view:
         recovery = view["recovery"]
-        result["first_read_command"] = context_capture_read_command(Path(recovery["path"]), recovery["sha256"],
-            page_bytes=page_bytes, max_page_output_bytes=output_bytes, initial_read=True,
-            reference_file=Path(result["recovery"]["reference_file"]), reader_argv=reader_argv)
+        if view.get("first_read_reference"):
+            # The producer already pinned this same initial selection. Keep
+            # that existing cursor instead of asking the reader to replan it.
+            result["first_read_command"] = context_capture_read_command(Path(recovery["path"]), recovery["sha256"],
+                page_bytes=page_bytes, max_page_output_bytes=output_bytes,
+                reference_file=Path(view["first_read_reference"]), reader_argv=reader_argv)
+        else:
+            result["first_read_command"] = context_capture_read_command(Path(recovery["path"]), recovery["sha256"],
+                page_bytes=page_bytes, max_page_output_bytes=output_bytes, initial_read=True,
+                reference_file=Path(result["recovery"]["reference_file"]), reader_argv=reader_argv)
     result["instruction"] += (" Append --select '<pointer>' or --list to recovery.command for current fields. "
         "Reader references keep complete digests locally; the CLI checks them on every page. "
         "A reference or digest grants no authority or freshness.")
@@ -347,6 +355,7 @@ def _native_context_reference(projected: dict[str, Any], reference: dict[str, An
                               total_bytes: int, metadata: dict[str, Any] | None,
                               progressive_enabled: bool, field_index: list[dict[str, Any]] | None,
                               initial_read_pointers: list[str] | None,
+                              initial_read_reference: Path | None,
                               page_bytes: int, output_bytes: int,
                               reader_argv: list[str] | None = None) -> dict[str, Any]:
     """Build the bounded native reference without publishing or registering it."""
@@ -398,6 +407,10 @@ def fits() -> bool:
             result.pop("first_read_command")  # A complete optional selection or the existing generic recovery.
             result["overview"] = overview
             result["instruction"] = instruction
+        elif initial_read_reference is not None:
+            result["first_read_reference"] = str(initial_read_reference)
+            if not fits():
+                result.pop("first_read_reference")  # Optional cursor must not displace the original recipe.
     # Prefer the concrete initial read over a broad scalar preview/root index.
     # Otherwise preserve the complete root index, bound to current recovery,
     # never a delta's patch keys. All deeper discovery remains available.
@@ -461,7 +474,8 @@ def native_context_capture_view(projected: dict[str, Any], directory: Path, *,
                                 captured_recovery: dict[str, Any] | None = None,
                                 metadata: dict[str, Any] | None = None,
                                 max_page_output_bytes: int | None = None,
-                                reader_argv: list[str] | None = None) -> dict[str, Any]:
+                                reader_argv: list[str] | None = None,
+                                initial_read_plan: dict[str, Any] | None = None) -> dict[str, Any]:
     """Publish the cheaper readable native transport for an already-decided view.
 
     TS owns valid deltas and their bindings. This IO adapter can only fall back
@@ -486,6 +500,11 @@ def native_context_capture_view(projected: dict[str, Any], directory: Path, *,
         if publish_wire else captured_recovery or projected["recovery"])
     current = captured_recovery or (projected.get("recovery") if typed else None) or reference
     current_raw = Path(current["path"]).read_bytes() if typed or captured_recovery is not None else raw
+    if hashlib.sha256(current_raw).hexdigest() != current["sha256"]:
+        raise QuotaDecisionTransportError("Current context capture changed; restore its original bytes without rerunning admission.")
+    if initial_read_plan is not None and initial_read_plan.get("capture") != {
+            key: current[key] for key in ("path", "sha256")}:
+        raise QuotaDecisionTransportError("Native read plan does not match the current immutable capture.")
     current_document = json.loads(current_raw)
     if not publish_wire:
         raw = current_raw
@@ -496,12 +515,32 @@ def native_context_capture_view(projected: dict[str, Any], directory: Path, *,
     # once for current facts, not once per full/delta transport candidate.
     field_index = None
     initial_read_pointers = None
-    if progressive_enabled:
+    initial_read_reference = None
+    if initial_read_plan is not None:
         try:
-            plan = _turn_envelope_result("context_select", context=current_document, select=[""], list_fields=True)
+            initial, plan = initial_read_plan["initial"], initial_read_plan["plan"]
+            if initial:
+                selections = plan["selections"]
+                initial_read_pointers = [item["pointer"] for item in selections]
+                selected = _context_selection_bytes(current_document, selections, list_fields=False, list_depth=None)
+                try:
+                    initial_read_reference = write_context_read_reference(Path(current["path"]), current["sha256"], selection={
+                        "select": initial_read_pointers, "initial_read": False, "list_fields": False, "list_depth": None,
+                        "sha256": hashlib.sha256(selected).hexdigest()})
+                except OSError:
+                    # Read-only capture roots keep the existing initial recipe;
+                    # metadata is optional and never a read/authority receipt.
+                    pass
+            else:
+                field_index = _context_index_fields(current_document, plan["selections"][0]["children"])
         except (EffectRuntimeRemoteError, EffectRuntimeStartupError):
             # Optional discovery cannot invalidate the immutable file reference.
             pass
+    elif progressive_enabled:
+        try:
+            plan = _turn_envelope_result("context_select", context=current_document, select=[""], list_fields=True)
+        except (EffectRuntimeRemoteError, EffectRuntimeStartupError):
+            pass
         else:
             field_index = _context_index_fields(current_document, plan["selections"][0]["children"])
             initial_read_pointers = plan.get("initial_read_pointers")
@@ -511,6 +550,7 @@ def view(kind: str, ref: dict[str, Any], body: bytes) -> dict[str, Any]:
             current_document=current_document, total_bytes=len(body), metadata=metadata,
             field_index=field_index, initial_read_pointers=initial_read_pointers,
             progressive_enabled=progressive_enabled,
+            initial_read_reference=initial_read_reference,
             page_bytes=page_bytes, output_bytes=output_bytes, reader_argv=reader_argv)
 
     if typed and mode == "delta":
@@ -557,20 +597,21 @@ def project_native_quota_context(payload: dict[str, Any], args: argparse.Namespa
                                  progressive_enabled: bool, reader_argv: list[str] | None) -> dict[str, Any]:
     """Prepare one already-decided native view; callers retain receipt failures."""
     from ..control_plane.turn_driver.codex_context import (
-        NativeContextSessionUnavailable, codex_context_directory, project_codex_context,
+        NativeContextSessionUnavailable, codex_context_directory, prepare_codex_context,
     )
     try:
         directory = codex_context_directory(directory=args.codex_context_dir,
             root=getattr(args, "codex_context_root", None), create=True)
     except NativeContextSessionUnavailable:
         directory = None  # Current full input, no fabricated native journal.
-    payload = project_codex_context(payload, directory=directory,
+    prepared = prepare_codex_context(payload, directory=directory,
         capture_directory=capture_directory, goal_id=decision_identity[0], agent_id=decision_identity[1],
         surface="turn_envelope" if getattr(args, "turn_envelope", False) else "quota_packet",
         capture_id=capture_id, continuity_generation=getattr(args, "_context_delivery_generation", None),
         progressive_enabled=progressive_enabled)
-    recovery = payload.get("recovery")
-    view = native_context_capture_view(payload, capture_directory, context_directory=directory, progressive_enabled=progressive_enabled,
+    recovery = prepared.delivery.get("recovery")
+    view = native_context_capture_view(prepared.delivery, capture_directory, context_directory=directory,
+        progressive_enabled=progressive_enabled, initial_read_plan=prepared.initial_read_plan,
         metadata={"capture_sha256": capture_id, **({"recovery": recovery} if recovery is not None else {})},
         max_page_output_bytes=getattr(args, "max_page_output_bytes", None), reader_argv=reader_argv)
     return model_context_reference(view, reader_argv=reader_argv)
@@ -663,7 +704,7 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
         except (QuotaDecisionTransportError, OSError, ValueError, EffectRuntimeRemoteError, EffectRuntimeStartupError):
             pass  # The complete original reference remains the fallback below.
     try:
-        from ..control_plane.turn_driver.codex_context import codex_context_directory, project_codex_context
+        from ..control_plane.turn_driver.codex_context import codex_context_directory, prepare_codex_context
 
         if not activation["enabled"]:
             raise ValueError("context delivery is off")
@@ -698,11 +739,12 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
         # into the immutable source or a second admission. Native lifecycle
         # still owns when the following full/delta view can be published.
         bind_capture_selection_transport(payload, args, capture_root=root)
-        projected = project_codex_context(payload, directory=directory, capture_directory=cache,
+        prepared = prepare_codex_context(payload, directory=directory, capture_directory=cache,
             goal_id=args.goal_id, agent_id=args.agent_id, surface="quota_packet",
             capture_id=hashlib.sha256(raw).hexdigest(),
             continuity_generation=getattr(args, "_context_delivery_generation", None),
             progressive_enabled=activation["progressive_enabled"], hook_session=hook_session)
+        projected = prepared.delivery
         native_prepared = True
         if projected.get("schema_version") == "context_delivery_v1":
             # Native publication owns the complete required view. Keep its
@@ -712,7 +754,7 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
                 page_bytes=page_bytes, output_bytes=output_bytes, reader_argv=reader_argv)
             reference = dict(full_reference)
         native_view = native_context_capture_view(projected, cache, context_directory=directory, hook_session=hook_session,
-            progressive_enabled=activation["progressive_enabled"],
+            progressive_enabled=activation["progressive_enabled"], initial_read_plan=prepared.initial_read_plan,
             metadata={"recovery": full_reference, "capture_sha256": args.capture_sha256,
                 "context_delivery": activation},
             max_page_output_bytes=getattr(args, "max_page_output_bytes", None), reader_argv=reader_argv)
diff --git a/loopx/control_plane/quota/progressive_context.ts b/loopx/control_plane/quota/progressive_context.ts
index 22bb9899d3..6c5f733650 100644
--- a/loopx/control_plane/quota/progressive_context.ts
+++ b/loopx/control_plane/quota/progressive_context.ts
@@ -3,7 +3,7 @@
 import {isDeepStrictEqual} from "node:util";
 import {visionAuthoringContract} from "../goals/vision_checkpoint.ts";
 import type {JsonObject} from "../effect_program.ts";
-import {jsonObject, requireJsonObject} from "../runtime_decode.ts";
+import {jsonObject, requireBoolean, requireJsonObject} from "../runtime_decode.ts";
 import {GOAL_OUTCOMES_SCHEMA, OUTCOME_CONTEXT_NAVIGATION_SCHEMA} from "../goals/outcome_requirements.ts";
 import {FULL_OUTCOME_READ_GUIDANCE, PROGRESSIVE_OUTCOME_READ_GUIDANCE} from "../work_items/interaction_contract.ts";
 import {quotaSummaryReferences, contextSourceIdentity, contextProjectionChanges,
@@ -11,6 +11,8 @@ import {quotaSummaryReferences, contextSourceIdentity, contextProjectionChanges,
 import {encodeContextDelivery, fileNavigationPaths} from "./context_delivery.ts";
 import {createHash} from "node:crypto";
 
+import {capturedContextSelection} from "./context_selection.ts";
+
 type Path = (string | number)[];
 const pointer = (path: Path): string => path.length ? "/" + path.map(key =>
   String(key).replaceAll("~", "~0").replaceAll("/", "~1")).join("/") : "";
@@ -239,7 +241,10 @@ function nativeContextContinuity(codec: JsonObject): JsonObject {
 
 /** Plan and encode one native capture without sending its intermediate view
  * back through Python. Capture publication and native locking remain Host IO. */
-export function prepareProgressiveContextDelivery(request: JsonObject, legacyFields: readonly string[]): JsonObject {
+export function prepareProgressiveContextDelivery(request: JsonObject, legacyFields: readonly string[],
+    decisionFields: readonly string[]): JsonObject {
+  if (request.include_initial_read_plan !== undefined)
+    requireBoolean(request.include_initial_read_plan, "include_initial_read_plan");
   const sourceValues: ProjectionSourceValues = new WeakMap();
   const required = requiredNativeContext(request, sourceValues);
   const codec = nativeContextContinuity({...request, context: required.context,
@@ -247,7 +252,9 @@ export function prepareProgressiveContextDelivery(request: JsonObject, legacyFie
   const delivery = encodeContextDelivery(codec, legacyFields, sourceValues, fileNavigationPaths(required.context));
   const {operation: _operation, source_edits: _edits, source_number_tokens: _numbers, ...source} = request;
   return contextProjectionChanges(source, {context: required.context, delivery, projected: required.projected,
-    read_sections: inlineReadSections(request, required.context)}, sourceValues);
+    read_sections: inlineReadSections(request, required.context),
+    ...(request.include_initial_read_plan === true
+      ? {initial_read_plan: nativeContextReadPlan(required.context, decisionFields)} : {})}, sourceValues);
 }
 
 /** Stateless presentation uses the same plan without inventing a native session,
@@ -259,6 +266,14 @@ export function prepareProgressiveContext(request: JsonObject): JsonObject {
   return contextProjectionChanges(source, {context: required.context, projected: required.projected}, values);
 }
 
+/** Invocation-local navigation, not captured input or a read/retention receipt.
+ * Decision vocabulary comes from the same owner as standalone initial reads. */
+function nativeContextReadPlan(context: JsonObject, decisionFields: readonly string[]): JsonObject {
+  const pointers = initialContextReadPointers(context, decisionFields), initial = pointers.length > 0;
+  return {initial, plan: capturedContextSelection(initial ? {context, initial_read: true}
+    : {context, select: [""], list_fields: true}, pointers)};
+}
+
 export function initialContextReadPointers(context: JsonObject, decisionFields: readonly string[]): string[] {
   const object = (value: unknown): JsonObject | undefined =>
     value !== null && typeof value === "object" && !Array.isArray(value) ? value as JsonObject : undefined;
diff --git a/loopx/control_plane/quota/turn_envelope.py b/loopx/control_plane/quota/turn_envelope.py
index 85e58fc392..dd53b996aa 100644
--- a/loopx/control_plane/quota/turn_envelope.py
+++ b/loopx/control_plane/quota/turn_envelope.py
@@ -185,6 +185,7 @@ def prepare_context_delivery(
     continuity: Mapping[str, Any] | None = None,
     surface: str = "work_context", native_projection: Mapping[str, Any] | None = None,
     consumed_pointers: Mapping[str, str] | None = None,
+    include_initial_read_plan: bool = False,
 ) -> dict[str, Any]:
     """Prepare under the Host's explicit continuity basis, not a model assertion.
 
@@ -201,6 +202,8 @@ def prepare_context_delivery(
         continuity=dict(continuity) if continuity is not None else None)
     if native_projection is not None:
         source["native_projection"] = dict(native_projection)
+        if include_initial_read_plan:
+            source["include_initial_read_plan"] = True
     if consumed_pointers is not None:
         source["consumed_pointers"] = dict(consumed_pointers)
     result = _turn_envelope_result("context_delivery", **source, source_edits=True)
diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts
index b25d761da0..9daa57c96a 100644
--- a/loopx/control_plane/quota/turn_envelope.ts
+++ b/loopx/control_plane/quota/turn_envelope.ts
@@ -1517,13 +1517,16 @@ export function turnEnvelopeRoute(envelope: JsonObject, historicalText?: JsonObj
 /** Navigation over captured facts, never a summary or a source-read receipt. */
 // Quota field vocabulary stays with its decision owner; the RFC 6901 reader
 // receives addresses and remains independent of Goal/permission policy.
+const INITIAL_CONTEXT_DECISION_FIELDS = ["ok", "schema_version", "goal_id", "agent_id", "should_run", "effective_action",
+  "normal_delivery_allowed", "recovery_delivery_allowed", "self_repair_allowed", "capability_repair_allowed", "workspace_repair_allowed",
+  "safe_bypass_allowed", "safe_bypass_kind", "blocked_action_scope", "reason",
+  "execution_obligation", "goal_boundary", "heartbeat_receipt", "work_lane_contract", "selected_todo"];
+
+/** Navigation over captured facts, never a summary or a source-read receipt. */
 function capturedDecisionSelection(request: JsonObject): JsonObject {
   const initial = request.initial_read === true || request.list_fields === true
     && Array.isArray(request.select) && request.select.length === 1 && request.select[0] === "";
-  return capturedContextSelection(request, initial ? initialContextReadPointers(requireJsonObject(request.context, "captured context"), ["ok", "schema_version", "goal_id", "agent_id", "should_run", "effective_action",
-    "normal_delivery_allowed", "recovery_delivery_allowed", "self_repair_allowed", "capability_repair_allowed", "workspace_repair_allowed",
-    "safe_bypass_allowed", "safe_bypass_kind", "blocked_action_scope", "reason",
-    "execution_obligation", "goal_boundary", "heartbeat_receipt", "work_lane_contract", "selected_todo"]) : []);
+  return capturedContextSelection(request, initial ? initialContextReadPointers(requireJsonObject(request.context, "captured context"), INITIAL_CONTEXT_DECISION_FIELDS) : []);
 }
 
 export function evaluateTurnEnvelope(value: unknown): JsonObject {
@@ -1606,7 +1609,7 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject {
   if (operation === "progressive_context_delivery") return prepareProgressiveContext(request);
   if (operation === "context_delivery") return request.native_projection === undefined
     ? prepareContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS)
-    : prepareProgressiveContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS);
+    : prepareProgressiveContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS, INITIAL_CONTEXT_DECISION_FIELDS);
   if (operation === "native_context_lifecycle") return updateNativeContextLifecycle(request);
   if (operation === "context_recovery") return restoreContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS);
   if (operation === "bind_capture") return bindTurnDecisionCapture(request);
diff --git a/loopx/control_plane/turn_driver/codex_context.py b/loopx/control_plane/turn_driver/codex_context.py
index a9803d94ea..6235b9101c 100644
--- a/loopx/control_plane/turn_driver/codex_context.py
+++ b/loopx/control_plane/turn_driver/codex_context.py
@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 from contextlib import contextmanager
+from dataclasses import dataclass
 from collections.abc import Iterator, Mapping
 from typing import Any
 import json
@@ -603,10 +604,31 @@ def register_native_context_view(directory: str | None, view: dict[str, Any], *,
         _save(root, state)
 
 
+@dataclass(frozen=True)
+class NativeContextPreparation:
+    """One invocation's wire and optional read plan; never native journal state."""
+
+    delivery: dict[str, Any]
+    initial_read_plan: dict[str, Any] | None = None
+
+
 def project_codex_context(payload: dict[str, Any], *, directory: str | None, capture_directory: Path,
                           goal_id: str, agent_id: str, surface: str, capture_id: str,
                           continuity_generation: str | None = None,
                           progressive_enabled: bool | None = None, hook_session: str | None = None) -> dict[str, Any]:
+    """Keep the historical projection API without requiring native navigation."""
+    return prepare_codex_context(payload, directory=directory, capture_directory=capture_directory,
+        goal_id=goal_id, agent_id=agent_id, surface=surface, capture_id=capture_id,
+        continuity_generation=continuity_generation, include_initial_read_plan=False,
+        progressive_enabled=progressive_enabled, hook_session=hook_session).delivery
+
+
+def prepare_codex_context(payload: dict[str, Any], *, directory: str | None, capture_directory: Path,
+                          goal_id: str, agent_id: str, surface: str, capture_id: str,
+                          continuity_generation: str | None = None,
+                          include_initial_read_plan: bool = True,
+                          progressive_enabled: bool | None = None,
+                          hook_session: str | None = None) -> NativeContextPreparation:
     """Commit current recovery before optional projection or candidate updates.
 
     Serialization prevents invalidation from moving the epoch during preparation.
@@ -619,10 +641,11 @@ def project_codex_context(payload: dict[str, Any], *, directory: str | None, cap
 
     if directory is None:
         if not progressive_enabled:
-            return payload
+            return NativeContextPreparation(payload)
         from .decision_capture import capture_progressive_context
 
-        return capture_progressive_context(payload, directory=capture_directory, surface=surface)[0]
+        return NativeContextPreparation(capture_progressive_context(
+            payload, directory=capture_directory, surface=surface)[0])
     try:
         with _locked(directory) as root:
             recovery_committed = False
@@ -632,7 +655,7 @@ def project_codex_context(payload: dict[str, Any], *, directory: str | None, cap
                 native_session = hook_session or os.environ.get("CODEX_THREAD_ID")
                 if (not isinstance(session, str) or not session
                         or (native_session and session != native_session)):
-                    return payload  # Do not replace a different native session's recovery.
+                    return NativeContextPreparation(payload)  # Do not replace a different native session's recovery.
                 # Native lifecycle events own workspace identity and epoch.
                 # A shell tool may run in another checkout without changing
                 # the host session or losing its previously delivered context.
@@ -676,6 +699,7 @@ def publish_recovery(reference: dict[str, Any]) -> None:
                     transaction = prepare_context_delivery(payload, scope=scope, capture_id=capture_id,
                         surface=surface, baseline=request.get("baseline"),
                         retained=request.get("retained"), continuity=request.get("continuity"),
+                        include_initial_read_plan=include_initial_read_plan if progressive else False,
                         native_projection={
                             "path": recovery["path"], "sha256": recovery["sha256"],
                             "source_path": recovery["path"],
@@ -684,7 +708,7 @@ def publish_recovery(reference: dict[str, Any]) -> None:
                                 + " --select '<supporting_material_pointer>/sources'",
                         } if progressive else None, consumed_pointers=state.get("consumed_pointers"))
                 except (EffectRuntimeRemoteError, EffectRuntimeStartupError):
-                    return payload
+                    return NativeContextPreparation(payload)
                 if progressive and transaction["projected"]:
                     recovery = capture_context_projection(transaction["context"], directory=capture_directory,
                         filename="required-context.json", read_guidance=CONTEXT_CAPTURE_PAGE_GUIDANCE)
@@ -705,11 +729,14 @@ def publish_recovery(reference: dict[str, Any]) -> None:
                         full_read_sections=read_sections,
                         read_sections=read_sections if result["mode"] == "full" else {})
                     _save(root, state)
-                return result
+                read_plan = transaction.get("initial_read_plan")
+                if read_plan is not None:
+                    read_plan = {**read_plan, "capture": {key: recovery[key] for key in ("path", "sha256")}}
+                return NativeContextPreparation(result, read_plan)
             except (OSError, ValueError, KeyError, TypeError, RecursionError):
                 if not recovery_committed:
                     _discard_recovery(root)
-                return payload
+                return NativeContextPreparation(payload)
     except (OSError, ValueError, KeyError, TypeError, RecursionError) as exc:
         # No safe fallback if the old selector cannot be invalidated under lock.
         raise QuotaDecisionTransportError(
diff --git a/tests/control_plane/test_context_delivery_cli.py b/tests/control_plane/test_context_delivery_cli.py
index 40c2242e40..77cd115a80 100644
--- a/tests/control_plane/test_context_delivery_cli.py
+++ b/tests/control_plane/test_context_delivery_cli.py
@@ -662,7 +662,7 @@ def native(payload, **kwargs):
             pytest.fail("failed native directory preparation must not reach projection")
         if native_failure == "projection":
             raise OSError("native observation unavailable")
-        return {"native_projection": payload}
+        return codex_context.NativeContextPreparation({"native_projection": payload})
 
     def unavailable_directory(**kwargs):
         raise OSError("native directory unavailable")
@@ -671,7 +671,7 @@ def unavailable_reference(*args, **kwargs):
         raise OSError("synthetic reader metadata unavailable")
 
     monkeypatch.setattr(quota, "project_captured_context", competing)
-    monkeypatch.setattr(codex_context, "project_codex_context", native)
+    monkeypatch.setattr(codex_context, "prepare_codex_context", native)
     if native_failure == "directory":
         monkeypatch.setattr(codex_context, "codex_context_directory", unavailable_directory)
     elif native_failure == "reader_reference":
@@ -699,7 +699,7 @@ def test_plain_quota_context_default_does_not_touch_transport(tmp_path, monkeypa
 
     def unexpected(*args, **kwargs):
         pytest.fail("off must not observe native context or create transport state")
-    monkeypatch.setattr(codex_context, "project_codex_context", unexpected)
+    monkeypatch.setattr(codex_context, "prepare_codex_context", unexpected)
     project, runtime, registry = _write_fixture(tmp_path)
     state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md"
     initialize_canonical_authority(runtime, GOAL_ID,
diff --git a/tests/control_plane/test_context_first_read.py b/tests/control_plane/test_context_first_read.py
index 3e32980ec0..aa2e4ff43f 100644
--- a/tests/control_plane/test_context_first_read.py
+++ b/tests/control_plane/test_context_first_read.py
@@ -64,9 +64,11 @@ def decision():
 
 def read_selection(command):
     words = shlex.split(command)
+    def option(name):
+        return words[words.index(name) + 1] if name in words else None
     selectors = [words[index + 1] for index, word in enumerate(words) if word == "--select"]
-    args = Namespace(capture_file=words[words.index("--capture-file") + 1],
-        capture_sha256=words[words.index("--capture-sha256") + 1], format="json", select=selectors or None,
+    args = Namespace(capture_file=option("--capture-file"), capture_sha256=option("--capture-sha256"),
+        reference_file=option("--reference-file"), format="json", select=selectors or None,
         initial_read="--initial-read" in words,
         offset=0, page_bytes=64000, max_page_output_bytes=64000)
     page = read_context_capture(args)
@@ -116,6 +118,113 @@ def test_initial_read_keeps_supporting_index_obligations_and_recovers_its_full_l
     assert json.loads(raw) == decision
 
 
+@pytest.mark.parametrize("damage", [None, "foreign_plan", "changed_file"])
+def test_native_preparation_reuses_only_its_current_digest_bound_navigation(
+    tmp_path, monkeypatch, decision, damage,
+):
+    from loopx.control_plane.quota import turn_envelope
+    from loopx.control_plane.quota.error_codes import QuotaDecisionTransportError
+    from tests.control_plane.test_codex_context_lifecycle import setup
+
+    setup(tmp_path, monkeypatch)
+    decision["selected_todo"]["future_number"] = 10**45 + 17
+    capture = tmp_path / "current"
+    capture.mkdir(mode=0o700)
+    prepared = codex_context.prepare_codex_context(decision, directory=str(tmp_path), capture_directory=capture,
+        goal_id="fixture", agent_id="agent", surface="quota_packet", capture_id="current")
+    assert prepared.initial_read_plan is not None
+    source = Path(prepared.delivery["recovery"]["path"])
+    raw = source.read_bytes()
+    journal = (tmp_path / "session.json").read_bytes()
+    if damage == "foreign_plan":
+        prepared.initial_read_plan["capture"] = {"path": str(source), "sha256": "0" * 64}
+    elif damage == "changed_file":
+        source.write_bytes(raw + b" ")  # Decodes identically; original byte identity still changed.
+
+    request = turn_envelope._turn_envelope_result
+
+    def no_second_selection(operation, **kwargs):
+        assert operation != "context_select", "The producer already planned this same capture"
+        return request(operation, **kwargs)
+
+    with monkeypatch.context() as publication:
+        publication.setattr(turn_envelope, "_turn_envelope_result", no_second_selection)
+        if damage is not None:
+            with pytest.raises(QuotaDecisionTransportError):
+                native_context_capture_view(prepared.delivery, capture, context_directory=str(tmp_path),
+                    initial_read_plan=prepared.initial_read_plan)
+            assert (tmp_path / "session.json").read_bytes() == journal
+            return
+        view = native_context_capture_view(prepared.delivery, capture, context_directory=str(tmp_path),
+            initial_read_plan=prepared.initial_read_plan)
+    _, values = read_selection(model_context_reference(view)["first_read_command"])
+    assert values["/selected_todo"]["future_number"] == 10**45 + 17
+    assert values["/goal_boundary"] == decision["goal_boundary"]
+    assert source.read_bytes() == raw
+    assert "initial_read_plan" not in json.loads(raw)
+
+
+@pytest.mark.parametrize("opaque", [False, True])
+def test_prepared_initial_read_keeps_exact_body_without_retransmitting_unused_prose(
+    tmp_path, monkeypatch, decision, opaque,
+):
+    from loopx.control_plane.quota import turn_envelope
+    from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime
+
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    tmp_path.chmod(0o700)
+    decision["unselected_evidence"] = "Historical evidence remains available. " * 10000
+    decision["unselected_structure"] = {f"unread_branch_{index}": {"detail": index} for index in range(500)}
+    decision["selected_todo"]["future_number"] = 10**45 + 17
+    view = native_context_capture_view(decision, tmp_path)
+    raw = Path(view["recovery"]["path"]).read_bytes()
+    original = read_context_capture(Namespace(capture_file=view["recovery"]["path"],
+        capture_sha256=view["recovery"]["sha256"], format="json", initial_read=True,
+        offset=0, page_bytes=64000, max_page_output_bytes=64000))
+    assert original["ok"] and original["eof"]
+    observed = []
+    project = turn_envelope._turn_envelope_result
+
+    def observe(operation, **params):
+        if operation == "context_select":
+            if opaque:
+                assert not params.get("initial_read"), "The prepared cursor must retain its resolved addresses"
+            observed.append(json.dumps(params, ensure_ascii=False).encode())
+        return project(operation, **params)
+
+    monkeypatch.setattr(turn_envelope, "_turn_envelope_result", observe)
+    navigation = model_context_reference(view) if opaque else view
+    page, values = read_selection(navigation["first_read_command"])
+    assert page["content"] == original["content"]
+    assert values["/selected_todo"]["future_number"] == 10**45 + 17
+    assert len(observed) == 1
+    if opaque:
+        assert b"Historical evidence remains available." not in observed[0]
+        assert len(observed[0]) < len(raw) // 2
+    else:
+        assert b"Historical evidence remains available." in observed[0]
+    assert Path(view["recovery"]["path"]).read_bytes() == raw
+
+
+def test_initial_navigation_keeps_the_original_recipe_when_cursor_metadata_is_unwritable(
+    tmp_path, monkeypatch, decision,
+):
+    import errno
+    from loopx.cli_commands import quota_capture
+
+    def unavailable(*args, **kwargs):
+        raise OSError(errno.ENOSPC, "synthetic cursor metadata unavailable")
+
+    monkeypatch.setattr(quota_capture, "write_context_read_reference", unavailable)
+    view = native_context_capture_view(decision, tmp_path)
+    assert "first_read_reference" not in view
+    assert "--initial-read" in view["first_read_command"]
+    page, values = read_selection(view["first_read_command"])
+    assert page["ok"] and page["eof"]
+    assert values["/goal_boundary"] == decision["goal_boundary"]
+    assert values["/selected_todo"] == decision["selected_todo"]
+
+
 @pytest.mark.parametrize("output_limit", [8192, 64000])
 def test_cli_initial_pages_prioritize_required_facts_and_keep_the_complete_index(tmp_path, decision, output_limit):
     goal = decision["interaction_contract"]["agent_channel"]["work_context"]["sources"][0]["content"]
diff --git a/tests/control_plane/test_quota_decision_capture.py b/tests/control_plane/test_quota_decision_capture.py
index 0cbf4b7cc4..cc5a791aa6 100644
--- a/tests/control_plane/test_quota_decision_capture.py
+++ b/tests/control_plane/test_quota_decision_capture.py
@@ -736,7 +736,7 @@ def refuse_recovery(payload, **kwargs):
             'Read the saved original decision/receipt; do not rerun admission.'
         )
 
-    monkeypatch.setattr(codex_context, 'project_codex_context', refuse_recovery)
+    monkeypatch.setattr(codex_context, 'prepare_codex_context', refuse_recovery)
     args = ['--registry', str(registry), '--runtime-root', str(runtime), '--format', 'json',
         'quota', 'should-run', '--goal-id', GOAL_ID, '--agent-id', AGENT_ID,
         '--todo-id', TODO_ID, '--turn-instance-id', TURN_ID, '--codex-app',
diff --git a/tests/control_plane/test_saved_context_project.py b/tests/control_plane/test_saved_context_project.py
index d00c414eef..f24f3994dc 100644
--- a/tests/control_plane/test_saved_context_project.py
+++ b/tests/control_plane/test_saved_context_project.py
@@ -13,7 +13,8 @@
 
 
 @pytest.mark.parametrize("failure", [None, "digest", "goal", "agent", "turn", "symlink", "object"])
-def test_saved_projection_checks_original_before_any_publication(tmp_path, failure):
+@pytest.mark.parametrize("setting", ["auto", "off"])
+def test_saved_projection_checks_original_before_any_publication(tmp_path, failure, setting):
     original = {"ok": True, "goal_id": "saved-goal", "agent_identity": {"agent_id": "saved-agent"},
         "heartbeat_receipt": {"turn_instance_id": "saved-turn"}, "requirement": "保留完整要求🙂" * 100}
     path = tmp_path / "original.json"
@@ -27,23 +28,37 @@ def test_saved_projection_checks_original_before_any_publication(tmp_path, failu
         path.symlink_to(source)
     args = ["quota", "context-project", "--capture-file", str(path), "--capture-sha256", digest,
         "--goal-id", "saved-goal", "--agent-id", "saved-agent", "--turn-instance-id", "saved-turn",
-        "--context-delivery", "off"]
+        "--context-delivery", setting]
     for name in ("digest", "goal", "agent", "turn"):
         if failure == name:
             option = {"digest": "--capture-sha256", "goal": "--goal-id", "agent": "--agent-id", "turn": "--turn-instance-id"}[name]
             args[args.index(option) + 1] = "0" * 64 if name == "digest" else "foreign"
     registry, runtime = tmp_path / "absent-registry.json", tmp_path / "absent-runtime"
     code, result = _run_cli(registry, runtime, *args, cwd=tmp_path)
-    assert not registry.exists() and not runtime.exists()
+    assert not registry.exists()
+    if failure or setting == "off":
+        assert not runtime.exists()
+    else:
+        # Main's default Progressive presentation may create private transport;
+        # it may not create authority, evaluate a guard or debit the saved Turn.
+        if runtime.exists():
+            assert {item.name for item in runtime.iterdir()} <= {"context-delivery"}
     if failure:
         assert code != 0 and not result["ok"] and "command" not in result
     else:
-        assert code == 0 and result["ok"] and result["mode"] == "full"
-        assert result["path"] == str(path) and result["sha256"] == digest
-        assert result["recovery"]["path"] == str(path)
-        command = shlex.split(result["command"])
-        code, page = _run_cli(registry, runtime, *command[3:], cwd=tmp_path)
-        assert code == 0 and page["content"].encode() == path.read_bytes()[:page["next_offset"]]
+        assert code == 0 and result["ok"]
+        if setting == "off":
+            assert result["mode"] == "full"
+            assert result["path"] == str(path) and result["sha256"] == digest
+            assert result["recovery"]["path"] == str(path)
+            command = shlex.split(result["command"])
+            code, page = _run_cli(registry, runtime, *command[3:], cwd=tmp_path)
+            assert code == 0 and page["content"].encode() == path.read_bytes()[:page["next_offset"]]
+        else:
+            from tests.control_plane.codex_checkpoint_fixture import read_cli_policy_payload
+
+            assert read_cli_policy_payload(json.dumps(result), returncode=code) == original
+        assert hashlib.sha256(path.read_bytes()).hexdigest() == digest
 
 
 @pytest.mark.parametrize("root_kind", ["unsafe", "symlink", "missing_parent"])
@@ -349,7 +364,7 @@ def refuse_recovery(payload, **kwargs):
             'Read the saved original decision/receipt; do not rerun admission.'
         )
 
-    monkeypatch.setattr(codex_context, "project_codex_context", refuse_recovery)
+    monkeypatch.setattr(codex_context, "prepare_codex_context", refuse_recovery)
     registry, runtime = tmp_path / "absent-registry.json", tmp_path / "absent-runtime"
     code = main(["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json",
         "quota", "context-project", "--capture-file", str(path), "--capture-sha256", digest,
@@ -942,7 +957,8 @@ def request(method, params):
     assert not list(home.rglob("*.pack")), "discovery must not clone a plugin catalog"
 
 
-def test_saved_projection_explicit_off_ignores_available_native_cache(tmp_path, monkeypatch):
+@pytest.mark.parametrize("setting", [None, "off"])
+def test_saved_projection_default_and_explicit_off_ignore_available_native_cache(tmp_path, monkeypatch, setting):
     from loopx.cli import build_parser
     from loopx.cli_commands.quota_capture import project_saved_context_capture
     from loopx.control_plane.turn_driver import codex_context
@@ -952,14 +968,20 @@ def test_saved_projection_explicit_off_ignores_available_native_cache(tmp_path,
         "heartbeat_receipt": {"turn_instance_id": "t"}}).encode()
     path.write_bytes(raw)
     def unexpected(*args, **kwargs):
-        pytest.fail("explicit-off saved presentation must not inspect/create native state")
+        pytest.fail(f"{setting or 'default'} saved presentation must not inspect/create native state")
     monkeypatch.setattr(codex_context, "codex_context_directory", unexpected)
-    args = build_parser().parse_args(["quota", "context-project", "--capture-file", str(path),
+    argv = ["quota", "context-project", "--capture-file", str(path),
         "--capture-sha256", hashlib.sha256(raw).hexdigest(), "--goal-id", "g", "--agent-id", "a",
-        "--turn-instance-id", "t", "--context-delivery", "off",
-        "--codex-context-dir", str(root), "--decision-output-root", str(root)])
+        "--turn-instance-id", "t", "--codex-context-dir", str(root), "--decision-output-root", str(root)]
+    if setting is not None:
+        argv.extend(("--context-delivery", setting))
+    args = build_parser().parse_args(argv)
     result = project_saved_context_capture(args)
-    assert result["ok"] and result["mode"] == "full" and result["path"] == str(path)
+    assert result["ok"]
+    if setting == "off":
+        assert result["mode"] == "full" and result["path"] == str(path)
+    else:
+        assert result == json.loads(raw)
     assert not root.exists() and path.read_bytes() == raw
 
 

From 25e4dc15532116d51bdc469c07e94aa043d10a0e Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Thu, 5 Nov 2026 16:10:46 +0800
Subject: [PATCH 02/13] Distinguish the pinned reader from an explicit
 navigation replan

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 tests/control_plane/test_context_read_reference.py | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/tests/control_plane/test_context_read_reference.py b/tests/control_plane/test_context_read_reference.py
index 9086d8327f..46658b99c3 100644
--- a/tests/control_plane/test_context_read_reference.py
+++ b/tests/control_plane/test_context_read_reference.py
@@ -229,9 +229,11 @@ def changed_navigation(operation, **kwargs):
         return original(operation, **kwargs)
 
     monkeypatch.setattr(turn_envelope, "_turn_envelope_result", changed_navigation)
-    # A fresh initial read now has different navigation over the same source.
-    current = read(model["first_read_command"])
+    # The explicit recipe replans; the prepared model cursor keeps the
+    # addresses selected when this immutable capture was presented.
+    current = read(machine["first_read_command"])
     assert json.loads(current["content"])["selections"] == [{"pointer": "/criterion", "value": payload["criterion"]}]
+    assert read(model["first_read_command"])["content"] == expected["content"]
     following = read(first["next_command"], reference_file=str(reference))
     if legacy:
         assert following["ok"] is False and "digest changed" in following["reason"]

From 2f416239e3ba29fc9a32916c404375dab7aa8f85 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Thu, 5 Nov 2026 16:23:25 +0800
Subject: [PATCH 03/13] Pin the benchmark preflight view and deliver its
 same-Turn selection route

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
(cherry picked from commit 47db78b33792e584036b2f099bed4aa0904f7e2b)
Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 benchmark/runtime/checkpoint.py               | 29 +++++++++++-----
 benchmark/tests/test_checkpoint_handoff.py    | 34 ++++++++++++++++---
 .../quota/progressive_context.ts              |  8 +++++
 .../control_plane/test_context_first_read.py  | 27 +++++++++++++++
 .../control_plane/test_replan_staged_read.py  |  7 +++-
 .../test_saved_context_project.py             | 11 ++++--
 6 files changed, 99 insertions(+), 17 deletions(-)

diff --git a/benchmark/runtime/checkpoint.py b/benchmark/runtime/checkpoint.py
index 164dd0f010..4d63ea17d7 100644
--- a/benchmark/runtime/checkpoint.py
+++ b/benchmark/runtime/checkpoint.py
@@ -230,7 +230,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 +254,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"])
+    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)
diff --git a/benchmark/tests/test_checkpoint_handoff.py b/benchmark/tests/test_checkpoint_handoff.py
index c86180f2c6..50eec72056 100644
--- a/benchmark/tests/test_checkpoint_handoff.py
+++ b/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"
 
 
diff --git a/loopx/control_plane/quota/progressive_context.ts b/loopx/control_plane/quota/progressive_context.ts
index 6c5f733650..009df7defb 100644
--- a/loopx/control_plane/quota/progressive_context.ts
+++ b/loopx/control_plane/quota/progressive_context.ts
@@ -371,6 +371,14 @@ export function initialContextReadPointers(context: JsonObject, decisionFields:
   for (const field of ["execution_preparation", "next_cli_actions", "spend_allowed_now", "spend_after_validation", "spend_policy",
     "quota_spend_source", "delivery_workspace_causality", "vision_continuation_audit"])
     include("interaction_contract", "cli_channel", field);
+  // Explicit selection clears next_cli_actions. Deliver the owner's discovery
+  // and same-Turn binding recipe alongside its non-exhaustive suggestions.
+  if (object(interaction.cli_channel)?.selection_required === true) {
+    for (const field of ["selection_required", "selection_policy_ref", "selection_command"])
+      include("interaction_contract", "cli_channel", field);
+    include("action_portfolio", "selection_policy");
+    include("action_portfolio", "suggested_actions");
+  }
   const planPath = ["interaction_contract", "cli_channel", "settlement_plan"];
   const plan = object(object(interaction.cli_channel)?.settlement_plan);
   if (plan?.schema_version === "quota_settlement_plan_v1" && Array.isArray(plan.ordered_steps)) {
diff --git a/tests/control_plane/test_context_first_read.py b/tests/control_plane/test_context_first_read.py
index aa2e4ff43f..88272836e1 100644
--- a/tests/control_plane/test_context_first_read.py
+++ b/tests/control_plane/test_context_first_read.py
@@ -76,6 +76,33 @@ def option(name):
     return page, selected_context_values(page["content"])
 
 
+@pytest.mark.parametrize("required", [True, False])
+def test_initial_read_delivers_explicit_selection_only_at_its_gate(tmp_path, decision, required):
+    cli = decision["interaction_contract"]["cli_channel"]
+    route = {"schema_version": "action_selection_cli_command_v1", "route_prefix": "loopx --format json",
+             "command_args_template": "quota should-run --todo-id <todo-id> --turn-instance-id fixture-turn",
+             "candidate_discovery_args": "todo list --goal-id fixture --status open"}
+    policy = {"decision_owner": "agent", "suggestions_exhaustive": False}
+    candidates = [{"todo_id": "another", "text": "Verify a candidate."}]
+    cli.update(selection_required=required, next_cli_actions=[], selection_command=route,
+               selection_policy_ref="$.action_portfolio.selection_policy")
+    decision["action_portfolio"] = {"selection_policy": policy, "suggested_actions": candidates}
+    source = tmp_path / "decision.json"
+    source.write_bytes(json.dumps(decision).encode())
+    command = shlex.join(["loopx", "quota", "context-read", "--capture-file", str(source),
+                          "--capture-sha256", hashlib.sha256(source.read_bytes()).hexdigest(), "--initial-read"])
+    _, values = read_selection(command)
+    expected = {"/interaction_contract/cli_channel/selection_required": True,
+                "/interaction_contract/cli_channel/selection_command": route,
+                "/interaction_contract/cli_channel/selection_policy_ref": "$.action_portfolio.selection_policy",
+                "/action_portfolio/selection_policy": policy, "/action_portfolio/suggested_actions": candidates}
+    for pointer, value in expected.items():
+        if required:
+            assert values[pointer] == value
+        else:
+            assert pointer not in values
+
+
 def test_initial_read_keeps_supporting_index_obligations_and_recovers_its_full_list(tmp_path, decision):
     # A complete address inventory is navigation, not the Goal/Task/stop text.
     # Exercise the native reader and exact cold path over the same capture.
diff --git a/tests/control_plane/test_replan_staged_read.py b/tests/control_plane/test_replan_staged_read.py
index 869edebeda..2a65915f17 100644
--- a/tests/control_plane/test_replan_staged_read.py
+++ b/tests/control_plane/test_replan_staged_read.py
@@ -17,6 +17,7 @@
     _turn_envelope_result, build_turn_envelope, quota_action_signature_document,
     turn_envelope_action_signature_document, prepare_context_delivery, restore_context_delivery,
 )
+from tests.control_plane.codex_checkpoint_fixture import reference_metadata_oracle
 from test_replan_host_context_projection import _quota_payload
 from test_replan_progress_native_cli import (
     AGENT, GOAL, TODO, TURN, call_progress_cli, make_progress_cli_fixture,
@@ -229,6 +230,10 @@ def observe(command, **kwargs):
         assert len(commands) == 1 and "--turn-envelope" not in commands[0]
         assert "--decision-output-root" in commands[0]
         source_path, view_path = wake / "checkpoint-start-guard.json", wake / "checkpoint-start-view.json"
+        if context_delivery:
+            argv = shlex.split(body.split("```sh\n", 1)[1].split("\n```", 1)[0])
+            view_path = Path(argv[argv.index("--capture-file") + 1])
+            assert view_path.parent != wake and view_path.is_relative_to(wake)
         source, view = json.loads(source_path.read_bytes()), json.loads(view_path.read_bytes())
         assert str(view_path) in body and str(source_path) not in body
         assert body.endswith("original heartbeat instructions")
@@ -256,7 +261,7 @@ def observe(command, **kwargs):
             assert projected.returncode == 0, projected.stdout
             presentation = json.loads(projected.stdout)
             assert presentation["ok"] and presentation["mode"] == "full"
-            assert presentation["recovery"]["path"] == str(view_path)
+            assert reference_metadata_oracle(presentation)["recovery"]["path"] == str(view_path)
         assert source["heartbeat_receipt"]["turn_instance_id"] == TURN
         assert source["heartbeat_receipt"] == view["heartbeat_receipt"]
         assert receipt["checkpoint"]["reason"] == "synthetic_no_native_capture"
diff --git a/tests/control_plane/test_saved_context_project.py b/tests/control_plane/test_saved_context_project.py
index f24f3994dc..3731f9db36 100644
--- a/tests/control_plane/test_saved_context_project.py
+++ b/tests/control_plane/test_saved_context_project.py
@@ -453,9 +453,15 @@ def test_checkpoint_preflight_selection_inherits_native_transport(tmp_path, monk
     env.pop("CODEX_THREAD_ID", None)
     receipt = {"turn_id": turn, "mode": "heartbeat", "stage": "execute"}
     body = capture_before_replan(env, receipt, "Current work.")
-    source = wake / "checkpoint-start-view.json"
-    raw = source.read_bytes()
     original_guard = (wake / "checkpoint-start-guard.json").read_bytes()
+    if enabled:
+        command = shlex.split(body.split("```sh\n", 1)[1].split("\n```", 1)[0])
+        source = Path(command[command.index("--capture-file") + 1])
+        assert source.parent != wake and source.is_relative_to(wake)
+        assert command[command.index("--capture-sha256") + 1] == hashlib.sha256(source.read_bytes()).hexdigest()
+    else:
+        source = wake / "checkpoint-start-view.json"
+    raw = source.read_bytes()
     assert _heartbeat_receipt_count(runtime, turn) == 1 and _spend_run_count(runtime) == 0
     assert not list(native.iterdir())  # Host preflight must not publish before native SessionStart.
 
@@ -482,7 +488,6 @@ def consume(wire):
             assert rc == 0, page
 
     if enabled:
-        command = shlex.split(body.split("```sh\n", 1)[1].split("\n```", 1)[0])
         command.extend(("--max-page-output-bytes", "32000"))
         rc, cold = _run_cli(registry, runtime, *command[command.index("quota"):], cwd=project)
         assert rc == 0 and cold["mode"] == "full", cold

From 2c3ff2c0bf161dc197146c03541498b40f680298 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Thu, 5 Nov 2026 16:45:50 +0800
Subject: [PATCH 04/13] Reuse prepared Codex read guidance in benchmark
 checkpoint prompts

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 benchmark/runtime/checkpoint.py                   |  5 +++--
 benchmark/runtime/context_delivery.py             |  5 +++--
 benchmark/runtime/worker.py                       |  5 ++++-
 benchmark/tests/test_context_delivery.py          | 11 ++++++++---
 tests/control_plane/test_saved_context_project.py | 10 +++++++++-
 5 files changed, 27 insertions(+), 9 deletions(-)

diff --git a/benchmark/runtime/checkpoint.py b/benchmark/runtime/checkpoint.py
index 4d63ea17d7..407d486e45 100644
--- a/benchmark/runtime/checkpoint.py
+++ b/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
@@ -267,7 +268,7 @@ def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: st
                 os.fchmod(stream.fileno(), 0o600)
                 stream.write(view_bytes)
         reading = preflight_context_instruction(env, source_path,
-            hashlib.sha256(view_bytes).hexdigest(), receipt["turn_id"])
+            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")
diff --git a/benchmark/runtime/context_delivery.py b/benchmark/runtime/context_delivery.py
index fc21c92f09..277ec2631e 100644
--- a/benchmark/runtime/context_delivery.py
+++ b/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. "
diff --git a/benchmark/runtime/worker.py b/benchmark/runtime/worker.py
index 105d77bd0c..17b7dab84b 100644
--- a/benchmark/runtime/worker.py
+++ b/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
diff --git a/benchmark/tests/test_context_delivery.py b/benchmark/tests/test_context_delivery.py
index 0fd4a3e498..11b42fb975 100644
--- a/benchmark/tests/test_context_delivery.py
+++ b/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"]
 
diff --git a/tests/control_plane/test_saved_context_project.py b/tests/control_plane/test_saved_context_project.py
index 3731f9db36..5dca3e2ab6 100644
--- a/tests/control_plane/test_saved_context_project.py
+++ b/tests/control_plane/test_saved_context_project.py
@@ -653,7 +653,15 @@ def response(request, number, message):
             elif phase in ("cold", "warm"):
                 if benchmark:
                     from loopx.control_plane.turn_driver.decision_capture import CONTEXT_CAPTURE_NAVIGATION_GUIDANCE
-                    assert CONTEXT_CAPTURE_NAVIGATION_GUIDANCE in _input_text(request)
+                    guide = CONTEXT_CAPTURE_NAVIGATION_GUIDANCE.strip()
+                    developer_text = [request.get("instructions") or ""]
+                    developer_text.extend(part.get("text", "") for item in request["input"]
+                        if item.get("role") == "developer" for part in item.get("content", []))
+                    assert any(guide in text for text in developer_text)
+                    wake_prompt = next(part["text"] for item in reversed(request["input"])
+                        if item.get("role") == "user" for part in item.get("content", [])
+                        if " quota context-project " in part.get("text", ""))
+                    assert guide not in wake_prompt
                     command = [line for line in _input_text(request).splitlines() if " quota context-project " in line][-1]
                     argv = shlex.split(command)
                     actual_turn = argv[argv.index("--turn-instance-id") + 1]

From ac31f36f428a9bebad73dc99ab1fcda82ae8057f Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Thu, 5 Nov 2026 17:02:21 +0800
Subject: [PATCH 05/13] Prove progressive benchmark choices from
 receiver-visible pages

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../test_saved_context_project.py             | 36 +++++++++++++++----
 1 file changed, 30 insertions(+), 6 deletions(-)

diff --git a/tests/control_plane/test_saved_context_project.py b/tests/control_plane/test_saved_context_project.py
index 5dca3e2ab6..ba3f003a9e 100644
--- a/tests/control_plane/test_saved_context_project.py
+++ b/tests/control_plane/test_saved_context_project.py
@@ -609,6 +609,7 @@ def test_native_saved_preflight_uses_lifecycle_without_guard_replay(tmp_path, mo
         (home / "hooks.json").write_text(json.dumps({"hooks": configuration["hooks"]}))
     phase, queue, observed, errors = "cold", [], {}, []
     read_offsets = {}
+    visible_choices = {}
     actual_turn = None
     rejected_read = False
     reading_initial = False
@@ -714,9 +715,17 @@ def response(request, number, message):
                     # Exercise the actual small reference and selective public
                     # reader in native model input; never consume either file in full.
                     assert "content" not in descriptor
-                    queue = [navigation["recovery"]["command"] + " --list", navigation["recovery"]["command"]
-                        + " --select /goal_id --select /heartbeat_receipt"
-                        + (" --select /retained_material/requirement" if large_capture else "")]
+                    reader = navigation["recovery"]["command"]
+                    select = " --select /goal_id --select /heartbeat_receipt"
+                    if benchmark and name in ("cold", "warm"):
+                        select += (" --select /interaction_contract/cli_channel/selection_command"
+                            " --select /action_portfolio/suggested_actions/0/todo_id"
+                            " --select /action_portfolio/suggested_actions/1/todo_id")
+                    if large_capture:
+                        select += " --select /retained_material/requirement"
+                    # The benchmark receiver knows these paths. Other native
+                    # scenarios exercise index discovery and digest recovery.
+                    queue = ([] if benchmark else [reader + " --list"]) + [reader + select]
                     if large_capture:
                         # Reject a different digest through the real native CLI,
                         # then continue with the unchanged original reference.
@@ -749,6 +758,13 @@ def response(request, number, message):
                         assert len(page["content"].encode()) == page["total_bytes"]
                         assert values["/retained_material/requirement"] == unicode_requirement
                         assert "资料🙂" not in _input_text(request)
+                    if benchmark:
+                        if name.endswith("_bound"):
+                            bound_id = values["/heartbeat_receipt"]["settlement_identity"]["todo_id"]
+                            assert bound_id == visible_choices[name.removesuffix("_bound")]
+                        else:
+                            slot = 0 if name == "cold" else 1
+                            visible_choices[name] = values[f"/action_portfolio/suggested_actions/{slot}/todo_id"]
                     native_context = home / "loopx-context" if benchmark else context
                     assert "candidate" not in json.loads((native_context / "session.json").read_bytes())
                 else:
@@ -764,9 +780,15 @@ def response(request, number, message):
                 if benchmark and name in ("cold", "warm"):
                     # Consume the command actually delivered to the native
                     # worker, rather than manually supplying missing flags.
-                    presented = json.loads(Path(descriptor["recovery"]["path"]).read_text())
-                    selection = presented["interaction_contract"]["cli_channel"]["selection_command"]
-                    command = selection["route_prefix"] + " " + selection["command_args_template"].replace("{todo_id}", TODO_ID)
+                    if progressive:
+                        assert page["offset"] == 0 and page["eof"]
+                        selection = values["/interaction_contract/cli_channel/selection_command"]
+                        todo_id = visible_choices[name]
+                    else:
+                        presented = json.loads(Path(descriptor["recovery"]["path"]).read_text())
+                        selection = presented["interaction_contract"]["cli_channel"]["selection_command"]
+                        todo_id = TODO_ID
+                    command = selection["route_prefix"] + " " + selection["command_args_template"].format(todo_id=todo_id)
                     argv = shlex.split(command)
                     assert argv[argv.index("--context-delivery") + 1] == "on"
                     assert argv[argv.index("--codex-context-dir") + 1] == str(home / "loopx-context")
@@ -862,6 +884,8 @@ def response(request, number, message):
         assert set(initial_content) == {"cold"} and not wrong_runtime.exists()
     if benchmark:
         turns = [json.loads(raw)["heartbeat_receipt"]["turn_instance_id"] for _, raw in sources]
+        if progressive:
+            assert visible_choices["cold"] == TODO_ID and visible_choices["warm"] != TODO_ID
         assert len(set(turns)) == 2
         assert all(_heartbeat_receipt_count(runtime, turn) == 2 for turn in turns)  # original guard + selection
         assert _spend_run_count(runtime) == 1  # only the explicit inter-wake settlement above

From 4798ab2dbdddeb3ad22e89317704f13540372a54 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Thu, 5 Nov 2026 17:33:51 +0800
Subject: [PATCH 06/13] Deliver the captured review command on the first replan
 read

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../quota/progressive_context.ts              |  9 +++
 .../test_quota_acceptance_references.py       | 78 ++++++++++++-------
 2 files changed, 58 insertions(+), 29 deletions(-)

diff --git a/loopx/control_plane/quota/progressive_context.ts b/loopx/control_plane/quota/progressive_context.ts
index 009df7defb..97315e2111 100644
--- a/loopx/control_plane/quota/progressive_context.ts
+++ b/loopx/control_plane/quota/progressive_context.ts
@@ -290,6 +290,15 @@ export function initialContextReadPointers(context: JsonObject, decisionFields:
     pointers.push("/" + path.map(key => String(key).replaceAll("~", "~0").replaceAll("/", "~1")).join("/"));
   };
   for (const field of decisionFields) include(field);
+  // A staged replan has already replaced inline acceptance with a pinned
+  // review reader. Deliver its first action and ordering without writeback.
+  const replan = object(context.replan_action_packet);
+  if (context.effective_action === "autonomous_replan_required" &&
+      replan?.schema_version === "replan_action_packet_v0" && replan.decision === "replan_required" &&
+      replan.settlement_only !== true && typeof replan.review === "string" && replan.review) {
+    include("replan_action_packet", "review");
+    include("replan_action_packet", "guidance");
+  }
   include("agent_lane_next_action", "next_step");
   include("agent_lane_next_action", "next_action_basis");
   include("interaction_contract", "user_channel");
diff --git a/tests/control_plane/test_quota_acceptance_references.py b/tests/control_plane/test_quota_acceptance_references.py
index cc5f7b8e7c..593de862d6 100644
--- a/tests/control_plane/test_quota_acceptance_references.py
+++ b/tests/control_plane/test_quota_acceptance_references.py
@@ -192,10 +192,29 @@ def test_shared_acceptance_survives_delta_and_lost_baseline_recovery():
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
-def test_real_cli_reuses_complete_acceptance_for_selection_and_replan(tmp_path, monkeypatch, provider):
-    from test_replan_successor_guard_reentry import _fixture, AGENT, GOAL
+def test_progressive_cli_delivers_complete_acceptance_for_selection_and_replan(tmp_path, monkeypatch, provider):
+    from test_replan_successor_guard_reentry import _fixture, AGENT, GOAL, ROOT
+    from tests.control_plane.codex_checkpoint_fixture import selected_context_values
+    from loopx.control_plane.testing.model_tool_behavior import execute_loopx_cli
     monkeypatch.setenv("LOOPX_USAGE_PING", "0")
     call, runtime, index = _fixture(tmp_path, monkeypatch, provider, future_monitor=False)
+    def run_reader(command):
+        return json.loads(execute_loopx_cli(command, source_root=ROOT, project_root=tmp_path / "project"))
+
+    def first_read(*args):
+        view = call("quota", "should-run", "--codex-app", "--context-delivery", "auto",
+                    "--goal-id", GOAL, "--agent-id", AGENT, *args)
+        assert view["schema_version"] == "context_capture_navigation_v1"
+        page = run_reader(view["first_read_command"])
+        chunks = bytearray()
+        while True:
+            assert page["ok"] and page["offset"] == len(chunks)
+            chunks.extend(page["content"].encode())
+            if page["eof"]:
+                assert len(chunks) == page["total_bytes"]
+                return selected_context_values(chunks)
+            page = run_reader(page["next_command"])
+
     registry_path = tmp_path / "registry.json"
     registry = json.loads(registry_path.read_text())
     registry["goals"][0]["execution_profile"] = {"planning_horizon_goal": True}
@@ -211,17 +230,20 @@ def test_real_cli_reuses_complete_acceptance_for_selection_and_replan(tmp_path,
          "--document", str(document), "--expected-provider-revision", before["provider_revision"],
          "--operation-id", "source-requirements", "--execute")
     expected = call("goal-acceptance", "inspect", "--goal-id", GOAL)["goal_acceptance_contract"]
-    guard = call("quota", "should-run", "--codex-app", "--context-delivery", "off", "--goal-id", GOAL, "--agent-id", AGENT,
-                 "--turn-instance-id", "turn-acceptance-sharing")
-    assert guard["effective_action"] == "autonomous_replan_required"
-    ref = "$.interaction_contract.agent_channel.work_context.sources[0].content.goal_acceptance_contract"
-    assert guard["autonomous_replan_obligation"]["replan_context"]["core_goal"]["acceptance_contract_ref"] == ref
-    assert guard["agent_todo_summary"]["goal_acceptance_contract_ref"] == ref
-    assert guard["interaction_contract"]["agent_channel"]["work_context"]["sources"][0]["content"]["goal_acceptance_contract"] == expected
-    assert build_turn_envelope(guard)["action_signature"]["matches"] is True
-
-    # Normal selected delivery must use the exact inline acceptance read and
-    # relocate its reference when a host consumes the CLI as a TurnEnvelope.
+    guard = first_read("--turn-instance-id", "turn-acceptance-sharing")
+    assert guard["/effective_action"] == "autonomous_replan_required"
+    assert guard["/normal_delivery_allowed"] is False
+    assert "/replan_action_packet/writeback" not in guard
+    review = run_reader(guard["/replan_action_packet/review"])
+    assert review["obligation_id"] == guard["/heartbeat_receipt/semantic_replan_obligation_id"]
+    acceptance = review["acceptance_index"]
+    row, = acceptance["outcomes"]
+    assert row["id"] == "export"
+    command = acceptance["read_action_template"].replace("{outcome_ref}", row["outcome_ref"])
+    original = expected["outcome_requirements"]["outcomes"][0]
+    assert run_reader(command)["outcome"] == original
+
+    # A later linked revision must reach the selected caller intact.
     index.write_text("")
     added = call("todo", "add", "--goal-id", GOAL, "--role", "agent", "--claimed-by", AGENT,
                  "--text", "Validate the complete export", "--task-class", "advancement_task",
@@ -237,19 +259,17 @@ def test_real_cli_reuses_complete_acceptance_for_selection_and_replan(tmp_path,
          "--document", str(document), "--expected-provider-revision", revision,
          "--operation-id", "link-export", "--execute")
     expected = call("goal-acceptance", "inspect", "--goal-id", GOAL)["goal_acceptance_contract"]
-    planning = call("quota", "should-run", "--codex-app", "--context-delivery", "off", "--goal-id", GOAL, "--agent-id", AGENT,
-                    "--turn-instance-id", "turn-export-planning")
-    assert planning["planning_horizon"]["core_goal"]["acceptance_contract"] == expected
-    assert planning["agent_todo_summary"]["goal_acceptance_contract_ref"] == "$.planning_horizon.core_goal.acceptance_contract"
-    selected = call("quota", "should-run", "--codex-app", "--context-delivery", "off", "--goal-id", GOAL, "--agent-id", AGENT,
-                    "--todo-id", added["todo_id"], "--turn-instance-id", "turn-export-delivery")
-    assert selected["normal_delivery_allowed"] is True
-    sources = selected["interaction_contract"]["agent_channel"]["work_context"]["sources"]
-    source_index = next(i for i, row in enumerate(sources) if row["source"] == "goal_acceptance")
-    retained = sources[source_index]["content"]["goal_acceptance_contract"]
-    assert retained == expected
-    assert selected["agent_todo_summary"]["goal_acceptance_contract_ref"] == (
-        f"$.interaction_contract.agent_channel.work_context.sources[{source_index}].content.goal_acceptance_contract")
-    envelope = build_turn_envelope(selected)
-    assert envelope["work_context"]["sources"][source_index]["content"]["goal_acceptance_contract"] == expected
-    assert envelope["action_signature"]["matches"] is True
+    planning = first_read("--turn-instance-id", "turn-export-planning")
+    assert planning["/interaction_contract/agent_channel/delivery_allowed"] is False
+    assert "/replan_action_packet/review" not in planning
+    selected = first_read("--todo-id", added["todo_id"], "--turn-instance-id", "turn-export-delivery")
+    assert selected["/normal_delivery_allowed"] is True
+    assert selected["/selected_todo/todo_id"] == added["todo_id"]
+    assert "/replan_action_packet/review" not in selected
+    sources = "/interaction_contract/agent_channel/work_context/sources/"
+    owner, = (path.removesuffix("/source") for path, value in selected.items()
+              if path.startswith(sources) and path.endswith("/source") and value == "goal_acceptance")
+    outcome = selected[owner + "/content/goal_acceptance_contract/outcome_requirements/outcomes/0"]
+    assert outcome == expected["outcome_requirements"]["outcomes"][0]
+    assert original["revision"] < outcome["revision"]
+    assert run_reader(command)["outcome"] == original

From aed73a8d2c54158281cf1fcc06a5f59bdf9e2b4a Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Thu, 5 Nov 2026 17:40:01 +0800
Subject: [PATCH 07/13] Validate progressive replan against its published
 reader contract

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 tests/control_plane/test_replan_staged_read.py | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/tests/control_plane/test_replan_staged_read.py b/tests/control_plane/test_replan_staged_read.py
index 2a65915f17..bb86b09ca3 100644
--- a/tests/control_plane/test_replan_staged_read.py
+++ b/tests/control_plane/test_replan_staged_read.py
@@ -11,6 +11,8 @@
 
 import pytest
 
+from tests.control_plane.codex_checkpoint_fixture import read_cli_policy_payload
+
 from canonical_authority_fixture import isolate_sqlite_runtime
 from loopx.control_plane.effect_runtime import EffectRuntimeRejected, restart_effect_runtime
 from loopx.control_plane.quota.turn_envelope import (
@@ -107,6 +109,7 @@ def test_pure_replan_does_not_advertise_unavailable_explore_attachment(tmp_path,
     try:
         entry = call("quota", "should-run", "--codex-app", "--goal-id", goal,
             "--agent-id", agent, "--turn-instance-id", turn, "--decision-output-dir", str(capture))
+        entry = read_cli_policy_payload(json.dumps(entry), returncode=0)
         source = json.loads((capture / "decision.json").read_bytes())
         original = explore_source(source)["content"]["graph"]
         graph = explore_source(entry)["content"]["graph"]
@@ -390,6 +393,7 @@ def test_real_guard_capture_read_and_writeback_without_another_guard(tmp_path, m
             "--goal-id", GOAL, "--agent-id", AGENT, "--turn-instance-id", TURN,
             "--todo-id", TODO, "--decision-output-dir", str(capture),
             *(["--turn-envelope"] if envelope else []))
+        guard = read_cli_policy_payload(json.dumps(guard), returncode=0)
         packet = guard["replan_action_packet"]
         if not envelope:
             assert "settlement_plan" not in guard["interaction_contract"]["cli_channel"]

From 5582b748ab51b812c324836595aee97191f75bae Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Thu, 5 Nov 2026 17:51:22 +0800
Subject: [PATCH 08/13] Isolate the user skill directory for native context
 comparisons

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 tests/control_plane/test_codex_context_boundary.py | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/tests/control_plane/test_codex_context_boundary.py b/tests/control_plane/test_codex_context_boundary.py
index 5c82cedcb7..3295a28a6c 100644
--- a/tests/control_plane/test_codex_context_boundary.py
+++ b/tests/control_plane/test_codex_context_boundary.py
@@ -127,13 +127,18 @@ def _native_fixture(tmp_path):
     home.mkdir()
     workspace.mkdir()
     env = {k: v for k, v in os.environ.items() if k in {
-        "PATH", "HOME", "USER", "LOGNAME", "SHELL", "TMPDIR", "LANG", "LC_ALL", "SYSTEMROOT"}}
+        "PATH", "USER", "LOGNAME", "SHELL", "TMPDIR", "LANG", "LC_ALL", "SYSTEMROOT"}}
     # Authority fixtures isolate temp files below tmp_path. Native release
     # helpers refuse a CODEX_HOME inside their temp root, which otherwise
     # breaks Linux sandbox dispatch. Keep both private, as sibling directories.
     temporary = tmp_path / "native-tmp"
     temporary.mkdir(mode=0o700)
     env.update({key: str(temporary) for key in ("TMPDIR", "TEMP", "TMP")})
+    # Codex can also discover skills in the OS user's home. Isolate both
+    # roots so a developer's personal skills cannot enter the measured input.
+    user_home = tmp_path / "user-home"
+    user_home.mkdir(mode=0o700)
+    env["HOME"] = env["USERPROFILE"] = str(user_home)
     env["CODEX_HOME"] = str(home)
     return binary, env, workspace, home
 

From b50f6486bf31d5af36f75cbdf06db992bee9935b Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Thu, 5 Nov 2026 18:01:16 +0800
Subject: [PATCH 09/13] Verify explicit guard opt-out on the real CLI

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../test_context_delivery_cli.py              | 29 ++++++++-----------
 1 file changed, 12 insertions(+), 17 deletions(-)

diff --git a/tests/control_plane/test_context_delivery_cli.py b/tests/control_plane/test_context_delivery_cli.py
index 77cd115a80..de5c27dfce 100644
--- a/tests/control_plane/test_context_delivery_cli.py
+++ b/tests/control_plane/test_context_delivery_cli.py
@@ -692,28 +692,23 @@ def unavailable_reference(*args, **kwargs):
     assert json.loads((tmp_path / "decision.json").read_text()) == original
 
 
-@pytest.mark.parametrize("scenario", [None, "off"])
-def test_plain_quota_context_default_does_not_touch_transport(tmp_path, monkeypatch, scenario):
-    from loopx.control_plane.turn_driver import codex_context
-    from loopx.cli import build_parser
-
-    def unexpected(*args, **kwargs):
-        pytest.fail("off must not observe native context or create transport state")
-    monkeypatch.setattr(codex_context, "prepare_codex_context", unexpected)
+def test_explicit_off_returns_inline_guard_without_native_transport(tmp_path, monkeypatch):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
     project, runtime, registry = _write_fixture(tmp_path)
     state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md"
     initialize_canonical_authority(runtime, GOAL_ID,
         build_todo_runtime_shadow_projection(goal_id=GOAL_ID, todos=read_fixture_todos(state), handoff_mode="soft_claim"),
         state_path=state, provider="file")
-    args = ["quota", "should-run", "--codex-app", "--goal-id", GOAL_ID,
-        "--agent-id", AGENT_ID, "--todo-id", TODO_ID, "--turn-instance-id", "off-scenario"]
-    if scenario:
-        args += ["--context-delivery", scenario]
-    code, result = _run_cli(registry, runtime, *args, cwd=project)
-    assert code == 0 and result["ok"] and result.get("schema_version") != "context_delivery_v1", result
-    # Absence is distinct from an explicit off so a configured machine default
-    # can be inherited. The ordinary output above still proves default-off.
-    assert build_parser().parse_args(["quota", "should-run"]).context_delivery is None
+    native = tmp_path / "unused-native-root"
+    turn = "explicit-off"
+    code, result = _run_cli(registry, runtime, "quota", "should-run", "--codex-app",
+        "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--todo-id", TODO_ID, "--turn-instance-id", turn,
+        "--codex-context-root", str(native), "--context-delivery", "off", cwd=project, raw_guard=False)
+    assert code == 0 and result["ok"], result
+    assert result["goal_id"] == GOAL_ID and result["heartbeat_receipt"]["turn_instance_id"] == turn
+    assert result["interaction_contract"]["cli_channel"]["settlement_plan"]
+    assert "reference_file" not in result and not native.exists()
+    assert _heartbeat_receipt_count(runtime, turn) == 1 and _spend_run_count(runtime) == 0
 
 
 @pytest.mark.parametrize("activation", [[], ["--context-delivery", "off"], ["--context-delivery", "on"]])

From efb66e20e4cd57ea4d889db4d8bdea0236b902d9 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Thu, 5 Nov 2026 20:04:59 +0800
Subject: [PATCH 10/13] test(quota): exercise the original review after a
 successor loses its claim

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../control_plane/test_replan_staged_read.py  |  1 +
 .../test_replan_successor_guard_reentry.py    | 21 +++++++++++++------
 2 files changed, 16 insertions(+), 6 deletions(-)

diff --git a/tests/control_plane/test_replan_staged_read.py b/tests/control_plane/test_replan_staged_read.py
index bb86b09ca3..f8a2dd7eb1 100644
--- a/tests/control_plane/test_replan_staged_read.py
+++ b/tests/control_plane/test_replan_staged_read.py
@@ -265,6 +265,7 @@ def observe(command, **kwargs):
             presentation = json.loads(projected.stdout)
             assert presentation["ok"] and presentation["mode"] == "full"
             assert reference_metadata_oracle(presentation)["recovery"]["path"] == str(view_path)
+            assert read_cli_policy_payload(projected.stdout, returncode=projected.returncode) == view
         assert source["heartbeat_receipt"]["turn_instance_id"] == TURN
         assert source["heartbeat_receipt"] == view["heartbeat_receipt"]
         assert receipt["checkpoint"]["reason"] == "synthetic_no_native_capture"
diff --git a/tests/control_plane/test_replan_successor_guard_reentry.py b/tests/control_plane/test_replan_successor_guard_reentry.py
index 52dbc22aa5..a222a52e85 100644
--- a/tests/control_plane/test_replan_successor_guard_reentry.py
+++ b/tests/control_plane/test_replan_successor_guard_reentry.py
@@ -71,6 +71,12 @@ def _fixture(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, *,
     ), state_path=state, provider=provider)
 
     def call(*args: str, expected_code: int = 0) -> dict:
+        # This fixture inspects the authority decision; checkpoint tests cover
+        # presentation and follow its captured reader to the delivered policy.
+        if args[:2] == ("quota", "should-run") and not any(
+            arg == "--context-delivery" or arg.startswith("--context-delivery=") for arg in args
+        ):
+            args = (*args, "--context-delivery", "off")
         result = subprocess.run([sys.executable, "-m", "loopx.cli", "--registry", str(registry),
             "--runtime-root", str(runtime), "--format", "json", *args], cwd=ROOT,
             capture_output=True, text=True, timeout=60)
@@ -345,13 +351,16 @@ def test_invalidated_successor_cannot_rebind_or_close_original_turn(
             "unclaimed": ["--clear-claim"]}[invalidation]
     call("todo", "update", "--goal-id", GOAL, "--agent-id", AGENT,
          "--todo-id", added["todo_id"], *edit)
-    # An unclaimed successor loses its original-lane proof. Fail closed while
-    # retaining the original receipt; a stale creation ACK cannot settle it.
-    guarded = _guard(call, expected_code=1 if invalidation == "unclaimed" else 0)
+    # Losing the successor invalidates its settlement proof. The same Turn
+    # can still review the original obligation from current canonical facts.
+    guarded = _guard(call)
     assert guarded["heartbeat_receipt"]["settlement_identity"] == identity
-    if invalidation == "unclaimed":
-        assert "settlement identity conflicts" in guarded["reason"]
-        assert guarded["heartbeat_receipt"]["event_id"] == prior_guard["heartbeat_receipt"]["event_id"]
+    assert guarded["heartbeat_receipt"]["event_id"] == prior_guard["heartbeat_receipt"]["event_id"]
+    obligation = guarded["autonomous_replan_obligation"]
+    assert guarded["effective_action"] == "autonomous_replan_required"
+    assert obligation["selection_binding"] == "heartbeat_receipt"
+    assert obligation["obligation_id"] == identity["replan_obligation_id"]
+    assert obligation["replan_context"]["obligation_id"] == identity["replan_obligation_id"]
     assert guarded.get("selected_todo") is None
     assert (guarded.get("autonomous_replan_obligation") or {}).get("resolution_mode") != "receipt_bound_replan_settlement"
     assert (guarded.get("replan_action_packet") or {}).get("settlement_only") is not True

From 15509d2ba9f8d3324f56368fa9141a0ff1e57662 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Thu, 5 Nov 2026 20:06:54 +0800
Subject: [PATCH 11/13] refactor(context): reuse the quota action vocabulary
 for staged review

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 loopx/control_plane/quota/progressive_context.ts | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/loopx/control_plane/quota/progressive_context.ts b/loopx/control_plane/quota/progressive_context.ts
index 97315e2111..7764a6b481 100644
--- a/loopx/control_plane/quota/progressive_context.ts
+++ b/loopx/control_plane/quota/progressive_context.ts
@@ -1,3 +1,4 @@
+import {EffectiveAction} from "./effective_action.generated.ts";
 /** Model presentation: requirements stay inline, supporting material is
  * addressed in the immutable source. This plan changes no canonical state. */
 import {isDeepStrictEqual} from "node:util";
@@ -293,7 +294,7 @@ export function initialContextReadPointers(context: JsonObject, decisionFields:
   // A staged replan has already replaced inline acceptance with a pinned
   // review reader. Deliver its first action and ordering without writeback.
   const replan = object(context.replan_action_packet);
-  if (context.effective_action === "autonomous_replan_required" &&
+  if (context.effective_action === EffectiveAction.AUTONOMOUS_REPLAN_REQUIRED &&
       replan?.schema_version === "replan_action_packet_v0" && replan.decision === "replan_required" &&
       replan.settlement_only !== true && typeof replan.review === "string" && replan.review) {
     include("replan_action_packet", "review");

From 3cab4567f39857f9cf85eb4d9f463ac8bd77f557 Mon Sep 17 00:00:00 2001
From: huangruiteng <huangrt01@163.com>
Date: Fri, 6 Nov 2026 12:34:03 +0800
Subject: [PATCH 12/13] fix(types): widen context delivery source values

Signed-off-by: huangruiteng <huangrt01@163.com>
---
 loopx/control_plane/quota/turn_envelope.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/loopx/control_plane/quota/turn_envelope.py b/loopx/control_plane/quota/turn_envelope.py
index dd53b996aa..a1208ae1ca 100644
--- a/loopx/control_plane/quota/turn_envelope.py
+++ b/loopx/control_plane/quota/turn_envelope.py
@@ -196,7 +196,7 @@ def prepare_context_delivery(
     """
     from .cli_projection import _apply_context_projection_changes
 
-    source = dict(context=dict(context), scope=dict(scope), capture_id=capture_id, surface=surface,
+    source: dict[str, Any] = dict(context=dict(context), scope=dict(scope), capture_id=capture_id, surface=surface,
         baseline=dict(baseline) if baseline is not None else None,
         retained=dict(retained) if retained is not None else None,
         continuity=dict(continuity) if continuity is not None else None)

From 6403996185a477f3cc9474ae6ad7beea9b7850d0 Mon Sep 17 00:00:00 2001
From: huangruiteng <huangrt01@163.com>
Date: Fri, 6 Nov 2026 13:13:48 +0800
Subject: [PATCH 13/13] refactor(context): make prepared native delivery the
 sole projection path

Signed-off-by: huangruiteng <huangrt01@163.com>
---
 loopx/cli_commands/quota_capture.py           | 27 +++++++++++--------
 .../turn_driver/codex_context.py              | 11 --------
 .../test_codex_context_boundary.py            |  6 ++---
 .../test_codex_context_lifecycle.py           | 10 +++----
 .../control_plane/test_context_first_read.py  | 16 ++++++++---
 .../test_context_read_reference.py            | 12 ++++++++-
 .../test_native_context_delivery_cost.py      |  4 +--
 .../test_progressive_native_context.py        | 16 +++++------
 .../test_quota_execution_preparation.py       | 14 +++++-----
 9 files changed, 65 insertions(+), 51 deletions(-)

diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py
index 75d9487944..b67a92a059 100644
--- a/loopx/cli_commands/quota_capture.py
+++ b/loopx/cli_commands/quota_capture.py
@@ -516,21 +516,26 @@ def native_context_capture_view(projected: dict[str, Any], directory: Path, *,
     field_index = None
     initial_read_pointers = None
     initial_read_reference = None
+
+    def pin_initial_selection(plan: dict[str, Any]) -> None:
+        nonlocal initial_read_pointers, initial_read_reference
+        selections = plan["selections"]
+        initial_read_pointers = [item["pointer"] for item in selections]
+        selected = _context_selection_bytes(current_document, selections, list_fields=False, list_depth=None)
+        try:
+            initial_read_reference = write_context_read_reference(Path(current["path"]), current["sha256"], selection={
+                "select": initial_read_pointers, "initial_read": False, "list_fields": False, "list_depth": None,
+                "sha256": hashlib.sha256(selected).hexdigest()})
+        except OSError:
+            # Read-only capture roots keep the existing initial recipe;
+            # metadata is optional and never a read/authority receipt.
+            pass
+
     if initial_read_plan is not None:
         try:
             initial, plan = initial_read_plan["initial"], initial_read_plan["plan"]
             if initial:
-                selections = plan["selections"]
-                initial_read_pointers = [item["pointer"] for item in selections]
-                selected = _context_selection_bytes(current_document, selections, list_fields=False, list_depth=None)
-                try:
-                    initial_read_reference = write_context_read_reference(Path(current["path"]), current["sha256"], selection={
-                        "select": initial_read_pointers, "initial_read": False, "list_fields": False, "list_depth": None,
-                        "sha256": hashlib.sha256(selected).hexdigest()})
-                except OSError:
-                    # Read-only capture roots keep the existing initial recipe;
-                    # metadata is optional and never a read/authority receipt.
-                    pass
+                pin_initial_selection(plan)
             else:
                 field_index = _context_index_fields(current_document, plan["selections"][0]["children"])
         except (EffectRuntimeRemoteError, EffectRuntimeStartupError):
diff --git a/loopx/control_plane/turn_driver/codex_context.py b/loopx/control_plane/turn_driver/codex_context.py
index 6235b9101c..4f63aef7ad 100644
--- a/loopx/control_plane/turn_driver/codex_context.py
+++ b/loopx/control_plane/turn_driver/codex_context.py
@@ -612,17 +612,6 @@ class NativeContextPreparation:
     initial_read_plan: dict[str, Any] | None = None
 
 
-def project_codex_context(payload: dict[str, Any], *, directory: str | None, capture_directory: Path,
-                          goal_id: str, agent_id: str, surface: str, capture_id: str,
-                          continuity_generation: str | None = None,
-                          progressive_enabled: bool | None = None, hook_session: str | None = None) -> dict[str, Any]:
-    """Keep the historical projection API without requiring native navigation."""
-    return prepare_codex_context(payload, directory=directory, capture_directory=capture_directory,
-        goal_id=goal_id, agent_id=agent_id, surface=surface, capture_id=capture_id,
-        continuity_generation=continuity_generation, include_initial_read_plan=False,
-        progressive_enabled=progressive_enabled, hook_session=hook_session).delivery
-
-
 def prepare_codex_context(payload: dict[str, Any], *, directory: str | None, capture_directory: Path,
                           goal_id: str, agent_id: str, surface: str, capture_id: str,
                           continuity_generation: str | None = None,
diff --git a/tests/control_plane/test_codex_context_boundary.py b/tests/control_plane/test_codex_context_boundary.py
index 3295a28a6c..e26f8ef635 100644
--- a/tests/control_plane/test_codex_context_boundary.py
+++ b/tests/control_plane/test_codex_context_boundary.py
@@ -183,13 +183,13 @@ def test_native_reader_requires_complete_hook_output_even_when_model_saw_page(tm
     publish = workspace / "publish.py"
     publish.write_text(
         "import json, pathlib, time, subprocess, shlex\n"
-        "from loopx.control_plane.turn_driver.codex_context import project_codex_context\n"
+        "from loopx.control_plane.turn_driver.codex_context import prepare_codex_context\n"
         "from loopx.cli_commands.quota_capture import native_context_capture_view\n"
         f"capture=pathlib.Path({str(capture)!r})\n"
         "body={'complete':True,'sources':[{'source':'selected_todo','content':{'todo':"
         "{'todo_id':'todo_fixture','text':'Preserve the original requirement. '*40,'status':'open'}}}]}\n"
-        f"full=project_codex_context(body,directory={str(root)!r},capture_directory=capture,"
-        "goal_id='fixture',agent_id='agent',surface='work_context',capture_id='fixture-capture')\n"
+        f"full=prepare_codex_context(body,directory={str(root)!r},capture_directory=capture,"
+        "goal_id='fixture',agent_id='agent',surface='work_context',capture_id='fixture-capture').delivery\n"
         f"view=native_context_capture_view(full,capture,context_directory={str(root)!r})\n"
         # Read the returned reference through its real command. The test still
         # needs an actual page in model input to isolate complete-hook coverage.
diff --git a/tests/control_plane/test_codex_context_lifecycle.py b/tests/control_plane/test_codex_context_lifecycle.py
index dc63f08b0f..edcaf7be86 100644
--- a/tests/control_plane/test_codex_context_lifecycle.py
+++ b/tests/control_plane/test_codex_context_lifecycle.py
@@ -164,9 +164,9 @@ def project(root, capture_id="one", *, payload=None, generation=None):
     directory.mkdir(mode=0o700)
     if payload is None:
         payload = {"ok": True, "goal_id": "fixture", "requirements": "Preserve final stop condition."}
-    result = context.project_codex_context(payload, directory=str(root), capture_directory=directory,
+    result = context.prepare_codex_context(payload, directory=str(root), capture_directory=directory,
         goal_id="fixture", agent_id="agent", surface="quota_packet", capture_id=capture_id,
-        continuity_generation=generation)
+        continuity_generation=generation).delivery
     return payload, result
 
 
@@ -325,13 +325,13 @@ def test_command_workdir_preserves_native_session_delta_and_recovery(tmp_path, m
     script = """
 import json, sys
 from pathlib import Path
-from loopx.control_plane.turn_driver.codex_context import project_codex_context
+from loopx.control_plane.turn_driver.codex_context import prepare_codex_context
 root = Path(sys.argv[1])
 capture = root / 'two'
 capture.mkdir(mode=0o700)
-print(json.dumps(project_codex_context(json.load(sys.stdin), directory=str(root),
+print(json.dumps(prepare_codex_context(json.load(sys.stdin), directory=str(root),
     capture_directory=capture, goal_id='fixture', agent_id='agent',
-    surface='quota_packet', capture_id='two')))
+    surface='quota_packet', capture_id='two').delivery))
 """
     result = subprocess.run([sys.executable, "-c", script, str(tmp_path)],
         input=json.dumps(payload), text=True, capture_output=True,
diff --git a/tests/control_plane/test_context_first_read.py b/tests/control_plane/test_context_first_read.py
index 88272836e1..a189ec3350 100644
--- a/tests/control_plane/test_context_first_read.py
+++ b/tests/control_plane/test_context_first_read.py
@@ -76,6 +76,16 @@ def option(name):
     return page, selected_context_values(page["content"])
 
 
+def pinned_initial_view(payload, directory):
+    base = native_context_capture_view(payload, directory)
+    from loopx.control_plane.quota import turn_envelope
+
+    plan = turn_envelope._turn_envelope_result("context_select", context=payload, initial_read=True)
+    capture = {key: base["recovery"][key] for key in ("path", "sha256")}
+    return native_context_capture_view(payload, directory, captured_recovery=base["recovery"],
+        initial_read_plan={"initial": True, "plan": plan, "capture": capture})
+
+
 @pytest.mark.parametrize("required", [True, False])
 def test_initial_read_delivers_explicit_selection_only_at_its_gate(tmp_path, decision, required):
     cli = decision["interaction_contract"]["cli_channel"]
@@ -203,7 +213,7 @@ def test_prepared_initial_read_keeps_exact_body_without_retransmitting_unused_pr
     decision["unselected_evidence"] = "Historical evidence remains available. " * 10000
     decision["unselected_structure"] = {f"unread_branch_{index}": {"detail": index} for index in range(500)}
     decision["selected_todo"]["future_number"] = 10**45 + 17
-    view = native_context_capture_view(decision, tmp_path)
+    view = pinned_initial_view(decision, tmp_path)
     raw = Path(view["recovery"]["path"]).read_bytes()
     original = read_context_capture(Namespace(capture_file=view["recovery"]["path"],
         capture_sha256=view["recovery"]["sha256"], format="json", initial_read=True,
@@ -768,8 +778,8 @@ def test_warm_delta_initial_read_uses_current_action_without_the_old_base(tmp_pa
         decision["effective_action"] = "normal_run" if marker == "before" else "autonomous_replan_required"
         decision["interaction_contract"]["agent_channel"].update(
             primary_action=f"Read {marker} action.", delivery_allowed=marker == "before")
-        wire = codex_context.project_codex_context(decision, directory=str(tmp_path), capture_directory=directory,
-            goal_id="fixture", agent_id="agent", surface="quota_packet", capture_id=name)
+        wire = codex_context.prepare_codex_context(decision, directory=str(tmp_path), capture_directory=directory,
+            goal_id="fixture", agent_id="agent", surface="quota_packet", capture_id=name).delivery
         view = native_context_capture_view(wire, directory, context_directory=str(tmp_path))
     assert view["mode"] == "delta" and "first_read_command" in view
     current = json.loads(Path(view["recovery"]["path"]).read_bytes())
diff --git a/tests/control_plane/test_context_read_reference.py b/tests/control_plane/test_context_read_reference.py
index 46658b99c3..b66eec7eb1 100644
--- a/tests/control_plane/test_context_read_reference.py
+++ b/tests/control_plane/test_context_read_reference.py
@@ -23,12 +23,22 @@ def read(command, **overrides):
     return read_context_capture(args)
 
 
+def pinned_initial_view(payload, directory):
+    base = native_context_capture_view(payload, directory)
+    from loopx.control_plane.quota import turn_envelope
+
+    plan = turn_envelope._turn_envelope_result("context_select", context=payload, initial_read=True)
+    capture = {key: base["recovery"][key] for key in ("path", "sha256")}
+    return native_context_capture_view(payload, directory, captured_recovery=base["recovery"],
+        initial_read_plan={"initial": True, "plan": plan, "capture": capture})
+
+
 def fixture(tmp_path):
     payload = {"ok": True, "interaction_contract": {"schema_version": "loopx_interaction_contract_v0"},
         "goal_boundary": {"stop": "Do not extend the original budget."},
         "selected_todo": {"todo_id": "todo_fixture", "text": "Compare 原始条件。" * 900},
         "criterion": {"value": 9007199254740993, "stop": "No publication is authorized."}}
-    view = native_context_capture_view(payload, tmp_path)
+    view = pinned_initial_view(payload, tmp_path)
     return payload, view, model_context_reference(view)
 
 
diff --git a/tests/control_plane/test_native_context_delivery_cost.py b/tests/control_plane/test_native_context_delivery_cost.py
index 73ea1c45bf..9886f7c6f1 100644
--- a/tests/control_plane/test_native_context_delivery_cost.py
+++ b/tests/control_plane/test_native_context_delivery_cost.py
@@ -219,10 +219,10 @@ def test_native_cost_selection_keeps_exact_numbers_and_unknown_constraints(tmp_p
 def test_disabled_native_context_stays_unwrapped_and_exact(tmp_path):
     payload = {**_payload(), "measurement": 2**60 + 1}
     unused = tmp_path / "unused"
-    projected = context.project_codex_context(
+    projected = context.prepare_codex_context(
         payload, directory=None, capture_directory=unused,
         goal_id="fixture", agent_id="agent", surface="quota_packet", capture_id="off",
-    )
+    ).delivery
     assert projected is payload and not unused.exists()
     view = native_context_capture_view(projected, tmp_path)
     assert view["mode"] == "full" and "target" not in view and "baseline_reference" not in view
diff --git a/tests/control_plane/test_progressive_native_context.py b/tests/control_plane/test_progressive_native_context.py
index 94ba08193e..e56881d3d2 100644
--- a/tests/control_plane/test_progressive_native_context.py
+++ b/tests/control_plane/test_progressive_native_context.py
@@ -220,10 +220,10 @@ def test_unknown_semantics_are_never_hidden(tmp_path, monkeypatch, change):
 def test_off_and_unsupported_surface_remain_exact(tmp_path, monkeypatch):
     setup(tmp_path, monkeypatch)
     original, unchanged = payload(), payload()
-    assert context.project_codex_context(original, directory=None, capture_directory=tmp_path,
-        goal_id="fixture", agent_id="agent", surface="quota_packet", capture_id="one") is original
-    wire = context.project_codex_context(original, directory=str(tmp_path), capture_directory=tmp_path,
-        goal_id="fixture", agent_id="agent", surface="turn_envelope", capture_id="one")
+    assert context.prepare_codex_context(original, directory=None, capture_directory=tmp_path,
+        goal_id="fixture", agent_id="agent", surface="quota_packet", capture_id="one").delivery is original
+    wire = context.prepare_codex_context(original, directory=str(tmp_path), capture_directory=tmp_path,
+        goal_id="fixture", agent_id="agent", surface="turn_envelope", capture_id="one").delivery
     assert wire["context"] == original == unchanged
     assert not (tmp_path / "required-context.json").exists()
 
@@ -253,8 +253,8 @@ def test_saved_cli_references_keep_tail_constraints_and_recover_without_detail_r
     path = tmp_path / "original.json"
     raw = json.dumps(original).encode()
     path.write_bytes(raw)
-    assert context.project_codex_context(original, directory=None, capture_directory=tmp_path,
-        goal_id="fixture", agent_id="agent", surface="quota_packet", capture_id="off") is original
+    assert context.prepare_codex_context(original, directory=None, capture_directory=tmp_path,
+        goal_id="fixture", agent_id="agent", surface="quota_packet", capture_id="off").delivery is original
     digest = hashlib.sha256(raw).hexdigest()
     registry, runtime = tmp_path / "no-registry.json", tmp_path / "no-runtime"
     command = ["quota", "context-project", "--capture-file", str(path), "--capture-sha256", digest,
@@ -346,9 +346,9 @@ def test_delta_only_retains_original_material_and_recovers_changes(tmp_path, mon
         source = {**payload(), "stop": stop, "evidence": ["stable material " * 1000]}
         capture = tmp_path / str(turn)
         capture.mkdir(mode=0o700)
-        wire = context.project_codex_context(source, directory=str(tmp_path), capture_directory=capture,
+        wire = context.prepare_codex_context(source, directory=str(tmp_path), capture_directory=capture,
             goal_id="fixture", agent_id="agent", surface="quota_packet", capture_id=str(turn),
-            progressive_enabled=False)
+            progressive_enabled=False).delivery
         assert json.loads(Path(wire["recovery"]["path"]).read_bytes()) == source
         from loopx.cli_commands.quota_capture import native_context_capture_view
         view = native_context_capture_view(wire, capture, context_directory=str(tmp_path), progressive_enabled=False)
diff --git a/tests/control_plane/test_quota_execution_preparation.py b/tests/control_plane/test_quota_execution_preparation.py
index 7570fd3f26..2473c022d9 100644
--- a/tests/control_plane/test_quota_execution_preparation.py
+++ b/tests/control_plane/test_quota_execution_preparation.py
@@ -210,8 +210,8 @@ def test_native_hook_returns_readable_ready_source_and_original_claim(
     goal_file.write_text(goal_text)
     capture = tmp_path / "before-claim"
     capture.mkdir(mode=0o700)
-    codex_context.project_codex_context(before, directory=str(native), capture_directory=capture,
-        goal_id=GOAL, agent_id=AGENT, surface="quota_packet", capture_id="before-claim")
+    codex_context.prepare_codex_context(before, directory=str(native), capture_directory=capture,
+        goal_id=GOAL, agent_id=AGENT, surface="quota_packet", capture_id="before-claim").delivery
     hook = [sys.executable, "-m", "loopx.entrypoint", "--registry", str(tmp_path / "registry.json"),
         "--runtime-root", str(tmp_path / "runtime"), "--format", "json", "quota", "context-hook",
         "--context-delivery", "on", "--runtime-profile", scenario,
@@ -322,8 +322,8 @@ def test_prepared_hook_refuses_a_replaced_ready_capture_without_reclaim(
     assert before["execution_preparation"]["state"] == "claim_required"
     original = tmp_path / "preparation"
     original.mkdir(mode=0o700)
-    codex_context.project_codex_context(before, directory=str(native), capture_directory=original,
-        goal_id=GOAL, agent_id=AGENT, surface="quota_packet", capture_id="claim-required")
+    codex_context.prepare_codex_context(before, directory=str(native), capture_directory=original,
+        goal_id=GOAL, agent_id=AGENT, surface="quota_packet", capture_id="claim-required").delivery
     from loopx.control_plane.turn_driver import codex_execution_preparation
     record = codex_context.record_native_execution_preparation
     run = codex_execution_preparation._default_runner
@@ -500,7 +500,7 @@ def test_codex_lifecycle_hook_uses_captured_authority_from_another_project(
 ):
     from loopx.control_plane.turn_driver.codex_context import (
         codex_context_hook,
-        project_codex_context,
+        prepare_codex_context,
     )
 
     before = _run(
@@ -527,7 +527,7 @@ def test_codex_lifecycle_hook_uses_captured_authority_from_another_project(
         "source": "startup",
     }
     assert codex_context_hook(str(context_root), event) == {}
-    project_codex_context(
+    prepare_codex_context(
         before,
         directory=str(context_root),
         capture_directory=capture_root,
@@ -535,7 +535,7 @@ def test_codex_lifecycle_hook_uses_captured_authority_from_another_project(
         agent_id=AGENT,
         surface="quota_packet",
         capture_id="claim-required",
-    )
+    ).delivery
 
     command = [
         sys.executable, "-m", "loopx.entrypoint",