fix(context): provide a focused first read for native packets

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

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,20 +172,26 @@ 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:
                 raise ValueError("selected continuation requires selection-sha256; follow next_command")
             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):

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.")

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} : {})};
 }