diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md
index 3bdd0728cb..0f69b6d3fd 100644
--- a/docs/reference/required-work-context.md
+++ b/docs/reference/required-work-context.md
@@ -155,6 +155,38 @@ model retention. These separate boundaries prevent transport loss from becoming
 either a repeated full-input loop or permission to act on missing requirements.
 No token savings or accepted outcome gains follow from instruction delivery alone.
 
+An enabled native **full** reference for a recognized decision can include
+`first_read_command`. It batches current admission/stop facts, the selected task,
+whole Goal/Todo narrative, source availability/revisions and read routes. Detailed
+history and optional authoring recipes remain expandable from the same capture.
+It also exposes existing outcome material addresses and the current settlement
+plan's obligations, including unknown step fields. Only attachments explicitly
+marked `required=false` defer their guidance/templates; their routes stay visible
+and the complete captured attachment must be read before use. Unknown plan
+versions remain whole. Initial pages do not repeat pointers already present in
+their content; the selection digest still prevents a full-input receipt.
+Its `context-read --initial-read` uses the same exact-address selector; paged
+continuations retain the short option and pin the selection digest. A changed
+profile rejects rather than mixing views. The option is read-only and exclusive with explicit
+selectors or discovery; unknown schemas use generic discovery.
+The concrete first read replaces the broad scalar preview and root index when it
+fits; unknown schemas or a tight reference budget keep generic discovery. Warm
+deltas keep their existing reader, and disabled native delivery is unchanged.
+This is a starting selection, not all required sources: read applicable criteria,
+evidence and the complete operation/settlement contract before depending on them.
+The original SHA-bound full recovery remains available; a selected page's EOF
+does not prove complete source delivery, model retention or fresh authority.
+
+中文：已启用原生传输的完整引用可给出 `first_read_command`，合并当前准入、停止条件、
+任务和完整 Goal/Todo 正文，以及来源状态、版本与读取入口；历史和可选格式说明仍在
+同一捕获中按需展开。它是起始选读，不代表已读全部必要来源；依赖具体验收、证据或
+执行/结算契约前仍须完整读取。未知 schema、过小引用预算和 warm delta 保留既有路线，
+关闭原生传输的输出不变；完整恢复和原 SHA 校验均保留。只读的 `--initial-read` 不与
+显式选读或索引混用；续页保留短选项和固定的选读摘要，profile 改变则拒绝混读。
+首读也给出现有验收材料地址和当前结算义务。仅明确 `required=false` 的附件延后完整
+说明与模板，使用前须读原附件；未知计划保留全文。首读分页不重复正文中的指针，
+选读摘要仍阻止将它记作完整输入。
+
 Checkpoint Host preflight uses the hidden `--defer-context-presentation` option
 with an explicitly enabled native JSON `should-run` capture. It collects the
 same complete Goal source for admitted work-context reads and writes the ordinary
diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py
index b1ac994eba..f98f155bb4 100644
--- a/loopx/cli_commands/quota_capture.py
+++ b/loopx/cli_commands/quota_capture.py
@@ -41,7 +41,8 @@ def _capture_page(path: Path, raw: bytes, digest: str, *, offset: int, size: int
                   select: list[str] | None = None,
                   selection_sha256: str | None = None,
                   list_fields: bool = False,
-                  list_depth: int | None = None) -> dict[str, Any]:
+                  list_depth: int | None = None,
+                  initial_read: bool = False) -> dict[str, Any]:
     end = min(offset + size, len(raw))
     while end < len(raw) and raw[end] & 0xC0 == 0x80:
         end -= 1
@@ -53,7 +54,12 @@ def page(text: str) -> dict[str, Any]:
             "offset": offset, "next_offset": cursor if cursor < len(raw) else None,
             "total_bytes": len(raw), "eof": cursor == len(raw), "content": text}
         if select is not None:
-            result.update(selection_sha256=selection_sha256, selected_pointers=select)
+            result["selection_sha256"] = selection_sha256
+            # The new initial view already names addresses in content; its
+            # short continuation pins that view's digest. Existing explicit
+            # selectors retain their original metadata and continuation.
+            if not initial_read:
+                result["selected_pointers"] = select
             if list_fields:
                 result["list_fields"] = True
             result["read_instruction"] = ("This is selected detail from the original capture, not a complete context read. "
@@ -68,7 +74,8 @@ def page(text: str) -> dict[str, Any]:
         if cursor < len(raw):
             result["next_command"] = context_capture_read_command(path, digest, offset=cursor, page_bytes=next_page_bytes or size,
                 max_page_output_bytes=next_page_output_bytes or output_limit,
-                select=select, selection_sha256=selection_sha256, list_fields=list_fields, list_depth=list_depth)
+                select=None if initial_read else select, selection_sha256=selection_sha256,
+                list_fields=list_fields, list_depth=list_depth, initial_read=initial_read)
         return result
 
     def fits(result: dict[str, Any]) -> bool:
@@ -165,12 +172,17 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
         if list_depth is not None and (not list_fields or type(list_depth) is not int or not 1 <= list_depth <= 8):
             raise ValueError("list-depth requires --list and an integer from 1 to 8")
         selectors = getattr(args, "select", None)
+        initial_read = getattr(args, "initial_read", False)
+        if initial_read and (selectors is not None or list_fields or list_depth is not None):
+            raise ValueError("initial-read cannot be combined with select or list")
         if list_fields and selectors is None:
             selectors = [""]
         selection_digest = getattr(args, "selection_sha256", None)
-        if selectors is None and selection_digest is not None:
+        if initial_read and offset > 0 and selection_digest is None:
+            raise ValueError("initial-read continuation requires selection-sha256; follow next_command")
+        if selectors is None and not initial_read and selection_digest is not None:
             raise ValueError("selection-sha256 requires --select")
-        if selectors is not None:
+        if selectors is not None or initial_read:
             if getattr(args, "format", None) != "json":
                 raise ValueError("selected context-read requires JSON output")
             if offset > 0 and selection_digest is None:
@@ -178,7 +190,8 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
             from ..control_plane.quota.turn_envelope import _turn_envelope_result
 
             depth_options = {"list_depth": list_depth} if list_depth is not None else {}
-            plan = _turn_envelope_result("context_select", context=payload, select=selectors, list_fields=list_fields, **depth_options)
+            selection_options = {"initial_read": True} if initial_read else {"select": selectors}
+            plan = _turn_envelope_result("context_select", context=payload, list_fields=list_fields, **selection_options, **depth_options)
             views: list[dict[str, Any]] = []
             for item in plan["selections"]:
                 selected: Any = payload
@@ -202,7 +215,7 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
             output_limit=output_limit, instruction=bool(context_dir),
             presentation=presentation, next_page_bytes=next_page_bytes,
             next_page_output_bytes=next_page_output_bytes, select=selectors, selection_sha256=selection_digest,
-            list_fields=list_fields, list_depth=list_depth)
+            list_fields=list_fields, list_depth=list_depth, initial_read=initial_read)
     except (OSError, ValueError, RecursionError, EffectRuntimeRemoteError, EffectRuntimeStartupError) as exc:
         # Recovery is presentation-only: keep failures separate from quota's
         # collection error path and retain the existing receipt/capture.
@@ -238,6 +251,7 @@ def _native_context_reference(projected: dict[str, Any], reference: dict[str, An
                               mode: str, current: dict[str, Any], current_document: dict[str, Any],
                               total_bytes: int, metadata: dict[str, Any] | None,
                               field_index: list[dict[str, Any]] | None,
+                              initial_read_pointers: list[str] | None,
                               page_bytes: int, output_bytes: int) -> dict[str, Any]:
     """Build the bounded native reference without publishing or registering it."""
     target = projected.get("target") if projected.get("schema_version") == "context_delivery_v1" else None
@@ -271,11 +285,21 @@ def _native_context_reference(projected: dict[str, Any], reference: dict[str, An
     # The reference must fit even under an explicit small caller ceiling.
     def fits() -> bool:
         return len(json.dumps(result, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= min(output_bytes, 7900)
-    # The same complete top-level index as `context-read --list`, bound to
-    # recovery (current facts), not a delta's patch keys. This saves an index
-    # tool roundtrip when it fits; it neither supplies source contents nor
-    # acknowledges their delivery. Deeper navigation keeps its existing route.
-    if field_index is not None:
+    if initial_read_pointers and mode == "full":
+        overview = result.pop("overview")
+        instruction = result["instruction"]
+        result["instruction"] = ("Start with first_read_command for current facts and source navigation. "
+            "Read each complete optional attachment from its captured step before using it. " + instruction)
+        result["first_read_command"] = context_capture_read_command(current_path, current["sha256"],
+            page_bytes=page_bytes, max_page_output_bytes=output_bytes, initial_read=True)
+        if not fits():
+            result.pop("first_read_command")  # A complete optional selection or the existing generic recovery.
+            result["overview"] = overview
+            result["instruction"] = instruction
+    # 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.
+    if field_index is not None and "first_read_command" not in result:
         result["field_index"] = field_index
         if not fits():
             result.pop("field_index")  # Optional metadata must never displace recovery or instructions.
@@ -359,21 +383,24 @@ def native_context_capture_view(projected: dict[str, Any], directory: Path, *,
 
     from ..control_plane.quota.turn_envelope import _turn_envelope_result
 
-    # Plan the existing optional index once for current facts, not once per
-    # transport candidate. Depth-one addressing needs keys, not source bodies.
+    # The same pure TS selector plans initial navigation and the optional index
+    # once for current facts, not once per full/delta transport candidate.
     field_index = None
+    initial_read_pointers = None
     try:
-        plan = _turn_envelope_result("context_select", context=dict.fromkeys(current_document), select=[""], list_fields=True)
+        plan = _turn_envelope_result("context_select", context=current_document, select=[""], list_fields=True)
     except (EffectRuntimeRemoteError, EffectRuntimeStartupError):
         # Optional discovery cannot invalidate the immutable file reference.
         pass
     else:
         field_index = _context_index_fields(current_document, plan["selections"][0]["children"])
+        initial_read_pointers = plan.get("initial_read_pointers")
 
     def view(kind: str, ref: dict[str, Any], body: bytes) -> dict[str, Any]:
         return _native_context_reference(projected, ref, mode=kind, current=current,
             current_document=current_document, total_bytes=len(body), metadata=metadata,
-            field_index=field_index, page_bytes=page_bytes, output_bytes=output_bytes)
+            field_index=field_index, initial_read_pointers=initial_read_pointers,
+            page_bytes=page_bytes, output_bytes=output_bytes)
 
     if typed and mode == "delta":
         result = None
@@ -737,6 +764,9 @@ def handoff_decision(directory: Path | None, socket_path: str | None,
 def handle_native_context_command(args: argparse.Namespace, *, print_payload: Callable[..., None],
                                   registry_path: Path | None = None, runtime_root_arg: str | None = None) -> int | None:
     """Dispatch native presentation/lifecycle IO before any quota authority read."""
+    if getattr(args, "initial_read", False) and args.quota_command != "context-read":
+        print_payload({"ok": False, "reason": "--initial-read is only supported by context-read"}, args.format, str)
+        return 1
     list_depth = getattr(args, "list_depth", None)
     if list_depth is not None and (args.quota_command != "context-read" or not getattr(args, "list_fields", False)
             or type(list_depth) is not int or not 1 <= list_depth <= 8):
diff --git a/loopx/cli_commands/quota_registration.py b/loopx/cli_commands/quota_registration.py
index 35d8d1d634..91969dc37d 100644
--- a/loopx/cli_commands/quota_registration.py
+++ b/loopx/cli_commands/quota_registration.py
@@ -228,6 +228,9 @@ def register_quota_command(
              "Use for shared Codex App hook configuration; cannot combine with --codex-context-dir. "
              "context-config renders native configuration for the exact Goal/Agent; no host configuration, trust, guard or spend are changed.")
     quota_parser.add_argument("--capture-sha256", help="For context-read/context-project: exact SHA-256 supplied with the capture reference.")
+    quota_parser.add_argument("--initial-read", action="store_true",
+        help="For JSON context-read: read current controls, task text and source navigation from a recognized capture. "
+             "Detailed criteria and operation contracts remain required before dependent work. Start at offset zero; follow next_command for continuation.")
     quota_parser.add_argument("--select", action="append",
         help="For context-read: repeat an exact RFC 6901 JSON pointer to read a batch from the original capture. "
              "Selected detail pages never acknowledge full-context retention; omit for unchanged full capture reads.")
diff --git a/loopx/control_plane/quota/context_selection.ts b/loopx/control_plane/quota/context_selection.ts
index 69e4d07b46..835a1c1503 100644
--- a/loopx/control_plane/quota/context_selection.ts
+++ b/loopx/control_plane/quota/context_selection.ts
@@ -3,9 +3,15 @@ import {requireJsonObject} from "../runtime_decode.ts";
 import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts";
 import type {JsonObject} from "../effect_program.ts";
 
-export function capturedContextSelection(request: JsonObject): JsonObject {
+export function capturedContextSelection(request: JsonObject, initialPointers: string[] = []): JsonObject {
   const context = requireJsonObject(request.context, "captured context");
-  const selectors = request.select;
+  if (request.initial_read !== undefined && typeof request.initial_read !== "boolean")
+    throw new EffectRuntimeRequestError("initial-read must be boolean");
+  if (request.initial_read === true && (request.select !== undefined || request.list_fields === true || request.list_depth !== undefined))
+    throw new EffectRuntimeRequestError("initial-read cannot be combined with select or list");
+  const selectors = request.initial_read === true ? initialPointers : request.select;
+  if (request.initial_read === true && !(selectors as string[]).length)
+    throw new EffectRuntimeRequestError("initial-read is unavailable for this capture schema; use generic discovery");
   if (!Array.isArray(selectors) || !selectors.length || selectors.some(value => typeof value !== "string"))
     throw new EffectRuntimeRequestError("context selection requires a nonempty array of JSON pointers");
   const depth = request.list_depth === undefined ? 1 : request.list_depth;
@@ -58,5 +64,7 @@ export function capturedContextSelection(request: JsonObject): JsonObject {
     }
     return {pointer, path};
   });
-  return {selections};
+  const initial = request.list_fields === true && pointers.length === 1 && pointers[0] === ""
+    ? initialPointers : [];
+  return {selections, ...(initial.length ? {initial_read_pointers: initial} : {})};
 }
diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts
index 680f592809..ed016e090a 100644
--- a/loopx/control_plane/quota/turn_envelope.ts
+++ b/loopx/control_plane/quota/turn_envelope.ts
@@ -1405,11 +1405,107 @@ export function turnEnvelopeRoute(envelope: JsonObject, historicalText?: JsonObj
   return action.quiet_noop_allowed === true ? "wait" : "blocked";
 }
 
+/** Navigation over captured facts, never a summary or a source-read receipt. */
+function initialReadPointers(context: JsonObject): string[] {
+  const object = (value: unknown): JsonObject | undefined =>
+    value !== null && typeof value === "object" && !Array.isArray(value) ? value as JsonObject : undefined;
+  const interaction = object(context.interaction_contract);
+  // Unknown envelopes retain generic discovery instead of guessing their policy.
+  if (interaction?.schema_version !== "loopx_interaction_contract_v0") return [];
+  const pointers: string[] = [];
+  const include = (...path: (string | number)[]): void => {
+    let value: unknown = context;
+    for (const key of path) {
+      if (value === null || typeof value !== "object" || !Object.hasOwn(value, key)) return;
+      value = (value as Record<string | number, unknown>)[key];
+    }
+    pointers.push("/" + path.map(key => String(key).replaceAll("~", "~0").replaceAll("/", "~1")).join("/"));
+  };
+  for (const field of ["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"])
+    include(field);
+  include("agent_lane_next_action", "next_step");
+  include("agent_lane_next_action", "next_action_basis");
+  include("interaction_contract", "user_channel");
+  include("interaction_contract", "agent_channel", "required_reads");
+  const work = ["interaction_contract", "agent_channel", "work_context"];
+  include(...work, "instruction");
+  include(...work, "user_todos");
+  const channel = object(interaction.agent_channel);
+  const sources = object(channel?.work_context)?.sources;
+  if (Array.isArray(sources)) sources.forEach((_source, index) => {
+    const source = [...work, "sources", index];
+    for (const field of ["source", "kind", "reason", "ordering", "command"])
+      include(...source, field);
+    // Current Goal/Todo text and source obligations stay whole. Expansive
+    // evidence pages, example recipes and historical bodies remain on demand.
+    include(...source, "content", "text");
+    include(...source, "content", "instruction");
+    for (const field of ["ok", "status", "reason", "error", "recovery", "revision", "provider_revision", "source_authority",
+      "decision_read_from_provider", "legacy_fallback_used", "graph_enabled", "harness_enabled", "boundary", "authority_read"])
+      include(...source, "content", field);
+    include(...source, "content", "current", "revision");
+    include(...source, "content", "current", "instruction");
+    include(...source, "content", "outcome_context");
+    const requirements = object(object(object(_source)?.content)?.goal_acceptance_contract)?.outcome_requirements;
+    const outcomes = object(requirements)?.outcomes;
+    if (Array.isArray(outcomes)) outcomes.forEach((_outcome, outcomeIndex) => {
+      const outcome = [...source, "content", "goal_acceptance_contract", "outcome_requirements", "outcomes", outcomeIndex];
+      include(...outcome, "id");
+      include(...outcome, "supporting_material_pointer");
+    });
+    include(...source, "content", "todo", "text");
+    include(...source, "content", "todo", "note");
+    include(...source, "content", "graph", "guidance");
+    include(...source, "content", "graph", "recent_findings");
+    include(...source, "content", "graph", "result_page");
+    include(...source, "content", "harness", "guidance");
+    include(...source, "content", "harness", "orchestration_gate");
+  });
+  for (const field of ["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);
+  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)) {
+    for (const field of Object.keys(plan)) if (field !== "ordered_steps") include(...planPath, field);
+    plan.ordered_steps.forEach((value, stepIndex) => {
+      const step = object(value), stepPath = [...planPath, "ordered_steps", stepIndex];
+      if (!step) { include(...stepPath); return; }
+      for (const field of Object.keys(step)) {
+        if (field !== "optional_attachments" || !Array.isArray(step[field])) { include(...stepPath, field); continue; }
+        step[field].forEach((attachment, attachmentIndex) => {
+          const path = [...stepPath, field, attachmentIndex];
+          if (object(attachment)?.required !== false) { include(...path); return; }
+          // Explicitly optional guidance is read before using that attachment.
+          // Keep its identity and actual read/authoring routes discoverable.
+          for (const key of Object.keys(object(attachment)!)) {
+            if (["guidance", "attachment_template", "linked_question_attachment_template", "path_delta_attachment_template",
+              "optional_measurement_example", "optional_question_evolution_example"].includes(key)) continue;
+            include(...path, key); // Unknown attachment controls are never hidden.
+          }
+        });
+      }
+    });
+  } else include(...planPath); // Unknown plan versions keep their complete obligations.
+  return pointers;
+}
+
+// Quota field vocabulary stays with its decision owner; the RFC 6901 reader
+// receives addresses and remains independent of Goal/permission policy.
+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 ? initialReadPointers(requireJsonObject(request.context, "captured context")) : []);
+}
+
 export function evaluateTurnEnvelope(value: unknown): JsonObject {
   const request = requireJsonObject(value, "turn envelope request");
   const operation = request.operation;
   if (operation === "replan_view") return replanView(request);
-  if (operation === "context_select") return capturedContextSelection(request);
+  if (operation === "context_select") return capturedDecisionSelection(request);
   if (operation === "replan_cli_view") {
     const payload = requireJsonObject(request.payload, "replan CLI payload");
     const reads = replanReadCommands(payload);
diff --git a/loopx/control_plane/turn_driver/decision_capture.py b/loopx/control_plane/turn_driver/decision_capture.py
index 584fa263b8..29487f30e5 100644
--- a/loopx/control_plane/turn_driver/decision_capture.py
+++ b/loopx/control_plane/turn_driver/decision_capture.py
@@ -211,7 +211,8 @@ def context_capture_read_command(path: Path, digest: str, *, offset: int = 0,
                                  select: list[str] | None = None,
                                  selection_sha256: str | None = None,
                                  list_fields: bool = False,
-                                 list_depth: int | None = None) -> str:
+                                 list_depth: int | None = None,
+                                 initial_read: bool = False) -> str:
     from ...command_invocation import loopx_cli_argv
 
     if page_bytes is None:
@@ -221,6 +222,7 @@ def context_capture_read_command(path: Path, digest: str, *, offset: int = 0,
     return shlex.join([*loopx_cli_argv(), "--format", "json", "quota", "context-read",
         "--capture-file", str(path), "--capture-sha256", digest,
         "--offset", str(offset), "--page-bytes", str(page_bytes),
+        *(["--initial-read"] if initial_read else []),
         *(["--list"] if list_fields else []),
         *(["--list-depth", str(list_depth)] if list_depth is not None else []),
         *(argument for pointer in select or [] for argument in ("--select", pointer)),
diff --git a/tests/control_plane/test_context_first_read.py b/tests/control_plane/test_context_first_read.py
new file mode 100644
index 0000000000..79a7c25108
--- /dev/null
+++ b/tests/control_plane/test_context_first_read.py
@@ -0,0 +1,248 @@
+"""Initial decision reading and progressive recovery share the original capture."""
+from __future__ import annotations
+
+from argparse import Namespace
+import hashlib
+import json
+from pathlib import Path
+import shlex
+
+import pytest
+
+from loopx.cli_commands.quota_capture import native_context_capture_view, read_context_capture
+from loopx.cli_commands.quota_capture import handle_native_context_command
+from loopx.control_plane.turn_driver import codex_context
+from loopx.control_plane.turn_driver.decision_capture import capture_page_range
+
+
+@pytest.fixture
+def decision():
+    return {
+        "ok": True,
+        "goal_id": "fixture",
+        "should_run": False,
+        "normal_delivery_allowed": False,
+        "workspace_repair_allowed": False,
+        "execution_obligation": {"must_attempt_work": False},
+        "goal_boundary": {"stop_condition": "No external write is authorized."},
+        "selected_todo": {"todo_id": "todo_current", "text": "Compare the retained evidence."},
+        "agent_lane_next_action": {"next_step": "Read the current criterion before judging completion.", "next_action_basis": "basis-current"},
+        "interaction_contract": {
+            "schema_version": "loopx_interaction_contract_v0",
+            "user_channel": {"notify": "DONT_NOTIFY", "action_required": False},
+            "agent_channel": {
+                "required_reads": ["read the applicable criterion, including its final stop"],
+                "work_context": {"instruction": "Sources remain mandatory when needed by the action.", "sources": [
+                    {"source": "goal", "command": "read the accepted requirements", "content": {
+                        "status": "unavailable", "revision": "current-revision", "recovery": "Restore this source before dependent work.",
+                        "text": "完整目标；完整条件。" * 120 + "STOP at the original budget.",
+                        "outcome_context": {"outcome_revision": 3},
+                        "full_criteria": ["the independently accepted requirement"],
+                    }},
+                    {"source": "explore", "content": {"graph": {
+                        "guidance": "Use existing evidence before repeating work.",
+                        "writeback_results": [{"detail": "older-evidence-" * 20000}],
+                    }}},
+                    {"source": "selected_todo", "content": {"todo": {"note": "Preserve the original failed attempt."}}},
+                ]},
+            },
+            "cli_channel": {"spend_allowed_now": False, "next_cli_actions": ["settle the original identity"],
+                "settlement_plan": {"schema_version": "quota_settlement_plan_v1", "identity": {"turn_instance_id": "original"},
+                    "ordered_steps": [{"kind": "durable_writeback", "precondition": "Validate first; preserve the original effect identity.",
+                        "unknown_obligation": "This new obligation must remain visible.", "optional_attachments": [
+                            {"required": False, "capability_id": "example", "option": "--example-json <file>",
+                                "new_gate": "Preserve this future attachment constraint.", "guidance": "optional recipe " * 3000},
+                            {"required": True, "guidance": "This attachment is required."},
+                        ]}]},
+            },
+        },
+    }
+
+
+def read_selection(command):
+    words = shlex.split(command)
+    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,
+        initial_read="--initial-read" in words,
+        offset=0, page_bytes=64000, max_page_output_bytes=64000)
+    page = read_context_capture(args)
+    assert page["ok"] and page["eof"]
+    return page, {item["pointer"]: item["value"] for item in json.loads(page["content"])["selections"]}
+
+
+def test_initial_read_keeps_current_constraints_whole_and_history_reachable(tmp_path, decision):
+    before = json.loads(json.dumps(decision))
+    view = native_context_capture_view(decision, tmp_path)
+    page, values = read_selection(view["first_read_command"])
+    assert values["/execution_obligation"] == {"must_attempt_work": False}
+    assert values["/should_run"] is False and values["/normal_delivery_allowed"] is False
+    assert values["/workspace_repair_allowed"] is False
+    assert values["/goal_boundary"] == decision["goal_boundary"]
+    assert values["/selected_todo"] == decision["selected_todo"]
+    assert values["/interaction_contract/agent_channel/required_reads"] == decision["interaction_contract"]["agent_channel"]["required_reads"]
+    assert values["/interaction_contract/cli_channel/spend_allowed_now"] is False
+    goal = "/interaction_contract/agent_channel/work_context/sources/0/content/text"
+    assert values[goal] == decision["interaction_contract"]["agent_channel"]["work_context"]["sources"][0]["content"]["text"]
+    assert values[goal.removesuffix("text") + "status"] == "unavailable"
+    assert values[goal.removesuffix("text") + "revision"] == "current-revision"
+    assert values[goal.removesuffix("text") + "recovery"] == "Restore this source before dependent work."
+    assert "writeback_results" not in page["content"]
+    assert "selected_pointers" not in page and page["selection_sha256"]
+    assert "first_read_command" in view and "overview" not in view and "field_index" not in view
+    raw = Path(view["recovery"]["path"]).read_bytes()
+    assert hashlib.sha256(raw).hexdigest() == view["recovery"]["sha256"]
+    assert json.loads(raw) == before == decision
+    assert capture_page_range(page, raw=raw, digest=page["sha256"]) is None
+    # Deferred criteria and history can still be selected, with the same SHA.
+    for pointer in [goal.removesuffix("text") + "full_criteria", "/interaction_contract/agent_channel/work_context/sources/1/content/graph/writeback_results"]:
+        recovery = read_context_capture(Namespace(capture_file=view["recovery"]["path"],
+            capture_sha256=view["recovery"]["sha256"], format="json", select=[pointer], offset=0,
+            page_bytes=64000, max_page_output_bytes=64000))
+        assert recovery["ok"] and recovery["sha256"] == page["sha256"]
+
+
+def test_initial_read_keeps_settlement_obligations_and_optional_routes(tmp_path, decision):
+    view = native_context_capture_view(decision, tmp_path)
+    page, values = read_selection(view["first_read_command"])
+    step = "/interaction_contract/cli_channel/settlement_plan/ordered_steps/0/"
+    assert values[step + "precondition"] == "Validate first; preserve the original effect identity."
+    assert values[step + "unknown_obligation"] == "This new obligation must remain visible."
+    assert values[step + "optional_attachments/0/required"] is False
+    assert values[step + "optional_attachments/0/option"] == "--example-json <file>"
+    assert values[step + "optional_attachments/0/new_gate"] == "Preserve this future attachment constraint."
+    assert values[step + "optional_attachments/1"]["guidance"] == "This attachment is required."
+    assert "optional recipe" not in page["content"]
+    # Existing explicit EOF selections retain their wire contract.
+    exact = read_context_capture(Namespace(capture_file=view["recovery"]["path"], capture_sha256=view["recovery"]["sha256"],
+        format="json", select=["/goal_boundary"], offset=0, page_bytes=64000, max_page_output_bytes=64000))
+    assert exact["selected_pointers"] == ["/goal_boundary"]
+
+
+def test_unknown_settlement_plan_is_not_compacted(tmp_path, decision):
+    plan = {"schema_version": "future", "new_stop": "Stop before every new effect."}
+    decision["interaction_contract"]["cli_channel"]["settlement_plan"] = plan
+    view = native_context_capture_view(decision, tmp_path)
+    _, values = read_selection(view["first_read_command"])
+    assert values["/interaction_contract/cli_channel/settlement_plan"] == plan
+
+
+def test_initial_read_exposes_existing_outcome_addresses_without_loading_all_criteria(tmp_path, decision):
+    pointer = "/interaction_contract/agent_channel/work_context/sources/0/content/goal_acceptance_contract/outcome_requirements/outcomes/0"
+    outcome = {"id": "current-work", "supporting_material_pointer": pointer,
+        "requirements": [{"description": "Read this whole criterion before claiming success. Final stop."}]}
+    decision["interaction_contract"]["agent_channel"]["work_context"]["sources"][0]["content"]["goal_acceptance_contract"] = {
+        "outcome_requirements": {"outcomes": [outcome]}}
+    view = native_context_capture_view(decision, tmp_path)
+    page, values = read_selection(view["first_read_command"])
+    assert values[pointer + "/id"] == "current-work"
+    assert values[pointer + "/supporting_material_pointer"] == pointer
+    assert "Read this whole criterion" not in page["content"]
+    detail = read_context_capture(Namespace(capture_file=view["recovery"]["path"], capture_sha256=view["recovery"]["sha256"],
+        format="json", select=[values[pointer + "/supporting_material_pointer"]], offset=0, page_bytes=64000, max_page_output_bytes=64000))
+    assert json.loads(detail["content"])["selections"][0]["value"] == outcome
+
+
+def test_small_reference_budget_omits_the_whole_initial_command(tmp_path, decision):
+    view = native_context_capture_view(decision, tmp_path, max_page_output_bytes=3000)
+    assert "first_read_command" not in view
+    assert len(json.dumps(view, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= 3000
+    assert Path(view["recovery"]["path"]).exists()
+
+
+def test_initial_command_rejects_tampered_capture_without_reentering_admission(tmp_path, decision):
+    view = native_context_capture_view(decision, tmp_path)
+    capture = Path(view["recovery"]["path"])
+    capture.write_text(json.dumps({**decision, "goal_boundary": {"stop_condition": "tampered"}}))
+    words = shlex.split(view["first_read_command"])
+    page = read_context_capture(Namespace(capture_file=str(capture),
+        capture_sha256=words[words.index("--capture-sha256") + 1], format="json", select=["/goal_boundary"],
+        offset=0, page_bytes=64000, max_page_output_bytes=64000))
+    assert not page["ok"]
+    assert "no guard, admission or spend" in page["recovery"]
+
+
+def test_initial_read_pages_pin_the_exact_selection_for_continuation(tmp_path, decision):
+    view = native_context_capture_view(decision, tmp_path)
+    args = Namespace(capture_file=view["recovery"]["path"], capture_sha256=view["recovery"]["sha256"],
+        format="json", initial_read=True, offset=0, page_bytes=256, max_page_output_bytes=24000)
+    page = read_context_capture(args)
+    assert page["ok"] and not page["eof"]
+    assert "--selection-sha256" in page["next_command"] and "--initial-read" in page["next_command"]
+    assert "--select " not in page["next_command"] and "selected_pointers" not in page
+    args.selection_sha256 = page["selection_sha256"]
+    args.offset = page["next_offset"]
+    following = read_context_capture(args)
+    assert following["ok"] and following["offset"] == 256
+    assert following["selection_sha256"] == page["selection_sha256"]
+    args.selection_sha256 = "0" * 64
+    changed = read_context_capture(args)
+    assert not changed["ok"] and "digest changed" in changed["reason"]
+
+
+def test_small_initial_pages_progress_without_repeating_the_pointer_vector(tmp_path, decision):
+    view = native_context_capture_view(decision, tmp_path, max_page_output_bytes=4500)
+    assert "first_read_command" in view
+    args = Namespace(capture_file=view["recovery"]["path"], capture_sha256=view["recovery"]["sha256"],
+        format="json", initial_read=True, offset=0, page_bytes=4500, max_page_output_bytes=4500)
+    content = []
+    for _ in range(30):
+        page = read_context_capture(args)
+        assert page["ok"] and len(json.dumps(page, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= 4500
+        content.append(page["content"])
+        if page["eof"]:
+            break
+        words = shlex.split(page["next_command"])
+        assert "--initial-read" in words and "--select" not in words
+        args.offset = int(words[words.index("--offset") + 1])
+        args.selection_sha256 = words[words.index("--selection-sha256") + 1]
+    assert page["eof"]
+    values = {item["pointer"]: item["value"] for item in json.loads("".join(content))["selections"]}
+    assert values["/goal_boundary"] == decision["goal_boundary"]
+    assert values["/interaction_contract/agent_channel/work_context/sources/0/content/text"].endswith("STOP at the original budget.")
+
+
+@pytest.mark.parametrize("options", [{"select": ["/goal_boundary"]}, {"list_fields": True},
+    {"offset": 256}, {"selection_sha256": "0" * 64}])
+def test_initial_read_rejects_ambiguous_scope_or_continuation(tmp_path, decision, options):
+    view = native_context_capture_view(decision, tmp_path)
+    args = 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, **options})
+    failure = read_context_capture(args)
+    assert not failure["ok"] and "no guard, admission or spend" in failure["recovery"]
+    assert any(text in failure["reason"] for text in ["cannot be combined", "requires selection-sha256", "digest changed"])
+
+
+def test_unknown_schema_keeps_generic_discovery_and_rejects_initial_read(tmp_path):
+    view = native_context_capture_view({"ok": True, "interaction_contract": {"schema_version": "future"}}, tmp_path)
+    assert "first_read_command" not in view and "index_command" in view
+    failure = 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 not failure["ok"] and "generic discovery" in failure["reason"]
+
+
+def test_initial_read_is_rejected_before_quota_or_machine_access(tmp_path):
+    output = []
+    code = handle_native_context_command(Namespace(initial_read=True, quota_command="should-run", format="json"),
+        print_payload=lambda value, *_args: output.append(value), registry_path=tmp_path / "unopened-registry.json")
+    assert code == 1 and not output[0]["ok"] and "only supported by context-read" in output[0]["reason"]
+    assert not (tmp_path / "unopened-registry.json").exists()
+
+
+def test_warm_delta_keeps_its_existing_reader_and_current_recovery(tmp_path, monkeypatch, decision):
+    tmp_path.chmod(0o700)
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setenv("CODEX_THREAD_ID", "fixture-session")
+    codex_context.codex_context_hook(str(tmp_path), {
+        "session_id": "fixture-session", "cwd": str(tmp_path), "hook_event_name": "SessionStart", "source": "startup"})
+    for name, marker in [("first", "before"), ("second", "after")]:
+        directory = tmp_path / name
+        directory.mkdir(mode=0o700)
+        decision["agent_lane_next_action"]["next_action_basis"] = marker
+        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)
+        view = native_context_capture_view(wire, directory, context_directory=str(tmp_path))
+    assert view["mode"] == "delta" and "first_read_command" not in view
+    current = json.loads(Path(view["recovery"]["path"]).read_bytes())
+    assert current["agent_lane_next_action"]["next_action_basis"] == "after"
+    assert view["sha256"] != view["recovery"]["sha256"]
diff --git a/tests/control_plane_ts/context_selection.test.ts b/tests/control_plane_ts/context_selection.test.ts
index d0ad4f6a58..856e4ba416 100644
--- a/tests/control_plane_ts/context_selection.test.ts
+++ b/tests/control_plane_ts/context_selection.test.ts
@@ -91,3 +91,45 @@ test("selection failures retain machine-readable causes at the runtime boundary"
     });
   }
 });
+
+test("full decision discovery points to current controls and whole source obligations before historical detail", () => {
+  const context = {
+    execution_obligation: {must_attempt_work: false},
+    goal_boundary: {stop_condition: "Do not execute the gated action."},
+    selected_todo: {todo_id: "todo_current"},
+    interaction_contract: {schema_version: "loopx_interaction_contract_v0", user_channel: {notify: "DONT_NOTIFY"},
+      agent_channel: {required_reads: ["read the complete accepted criterion"], work_context: {
+        instruction: "Read required sources before dependent effects.", sources: [
+          {source: "goal", command: "read the current source", content: {text: "A whole goal.\nFinal stop condition.",
+            outcome_context: {outcome_revision: 3}, full_criteria: ["Still required before claiming completion."]}},
+          {source: "explore", content: {graph: {guidance: "Use relevant evidence.", writeback_results: ["old detailed evidence"]}}},
+        ]}}, cli_channel: {next_cli_actions: ["the exact original settlement"], spend_allowed_now: false}},
+  };
+  const before = structuredClone(context);
+  const result = evaluateTurnEnvelope({operation: "context_select", context, select: [""], list_fields: true});
+  const pointers = result.initial_read_pointers as string[];
+  for (const pointer of ["/execution_obligation", "/goal_boundary", "/selected_todo",
+    "/interaction_contract/agent_channel/required_reads", "/interaction_contract/cli_channel/spend_allowed_now",
+    "/interaction_contract/agent_channel/work_context/sources/0/content/text"])
+    assert.ok(pointers.includes(pointer), pointer);
+  assert.ok(!pointers.some(pointer => /writeback_results|full_criteria/.test(pointer)));
+  assert.doesNotThrow(() => select(context, pointers));
+  // Discovery is not a claim that omitted criteria were read or are optional.
+  assert.deepEqual(select(context, ["/interaction_contract/agent_channel/work_context/sources/0/content/full_criteria"]),
+    {selections: [{pointer: "/interaction_contract/agent_channel/work_context/sources/0/content/full_criteria",
+      path: ["interaction_contract", "agent_channel", "work_context", "sources", 0, "content", "full_criteria"]}]});
+  assert.deepEqual(context, before);
+  assert.deepEqual(evaluateTurnEnvelope({operation: "context_select", context, initial_read: true}), select(context, pointers));
+});
+
+test("unknown decision schemas and ordinary selections keep generic discovery", () => {
+  for (const context of [{value: true}, {interaction_contract: {schema_version: "future"}}]) {
+    const result = evaluateTurnEnvelope({operation: "context_select", context, select: [""], list_fields: true});
+    assert.equal(result.initial_read_pointers, undefined);
+    assert.throws(() => evaluateTurnEnvelope({operation: "context_select", context, initial_read: true}), /generic discovery/);
+  }
+  const context = {interaction_contract: {schema_version: "loopx_interaction_contract_v0"}};
+  assert.equal(select(context, [""]).initial_read_pointers, undefined);
+  for (const options of [{select: [""]}, {list_fields: true}, {list_depth: 2}])
+    assert.throws(() => evaluateTurnEnvelope({operation: "context_select", context, initial_read: true, ...options}), /cannot be combined/);
+});