fix(context): keep native reader digests in private references

benchmark/tests/test_native_checkpoint_handoff.py

@@ -384,7 +384,9 @@ def test_actual_worker_captures_pending_replan_before_native_model(tmp_path, exp
     from benchmark.runtime.worker import run_once
     from tests.control_plane import test_quota_settlement_cli as cli
     from tests.control_plane.canonical_authority_fixture import initialize_canonical_source_fixture
-    from tests.control_plane.codex_checkpoint_fixture import NativeContextReader, native_context_command
+    from tests.control_plane.codex_checkpoint_fixture import (
+        NativeContextReader, native_context_command, native_process_continuation,
+    )
 
     project, runtime, registry = cli._write_fixture(tmp_path / "authority")
     document = json.loads(registry.read_text())
@@ -425,6 +427,8 @@ def run(*args):
 
     def response(payload, index, message):
         requests.append(payload)
+        if continuation := native_process_continuation(payload, f"checkpoint_{index}"):
+            return continuation
         if context_delivery == "on" and phase["kind"] != "initial":
             follow = reader.follow(payload)
             if follow:

docs/reference/required-work-context.md

@@ -880,6 +880,30 @@ journal，防止短暂 IO 故障恢复后复活旧基线。不可变 capture 与
 
 ### Selected capture details / 按需读取捕获详情
 
+Enabled native navigation uses `quota context-read --reference-file <file>`.
+The immutable, private reader document keeps the full capture digest; selected
+continuations also keep the exact selection and its full digest there. The CLI
+checks both on every page, including after a later Turn or a changed reader
+profile. Its short filename is only a locator, never a truncated integrity check,
+permission or freshness claim. Missing, malformed, public or symlinked reader
+documents and capture/selection digest mismatches fail closed without admission. Selection reads may publish a derived
+private reader document beside the capture; they do not write Goal state.
+Compact selected pages use `context_capture_selection_page_v1`, with a reader
+reference and byte cursors instead of repeated hashes or pointer vectors. They
+are never full-input observations. Explicit file/hash readers and full-page
+observation receipts keep their existing schema and complete digest fields.
+The navigation document also retains the complete machine navigation metadata.
+Very small output budgets may keep the existing explicit hash reference when
+the compact navigation cannot fit; no navigation or source is truncated.
+
+已启用的原生导航通过 `--reference-file` 读取不可变私有引用；完整 capture 摘要、
+选读路径与 selection 摘要保存在本地，CLI 每页校验，跨 Turn 或 reader profile
+改变也不会混读。短文件名只负责定位，不替代完整校验，也不代表权限或新鲜度。
+引用缺失、格式错误、公开可读、为符号链接或摘要不符时拒绝读取，不重新 admission。选读可在
+捕获旁生成派生私有引用，不修改 Goal 状态。精简选读页返回引用及字节游标，使用
+独立 schema，不能作为完整输入证明；原显式文件/摘要接口及完整页证明保持兼容。
+导航引用同时保存完整机器元数据；极小输出预算保留原显式摘要引用，不截断正文或导航。
+
 `quota context-read --select /path --select /another/path` reads an explicit
 batch of RFC 6901 JSON pointers from the original digest-bound capture. Supply
 the existing `--capture-file` and `--capture-sha256` reference, or start through

loopx/cli_commands/context_machine.py

@@ -23,7 +23,7 @@ def has_explicit_context_read_source(args: argparse.Namespace) -> bool:
     """
     return getattr(args, "quota_command", None) == "context-read" and any(
         getattr(args, key, None) for key in (
-            "capture_file", "capture_sha256", "codex_context_dir", "codex_context_root"
+            "capture_file", "capture_sha256", "reference_file", "codex_context_dir", "codex_context_root"
         )
     )
 

loopx/cli_commands/quota.py

@@ -15,7 +15,7 @@
 from .quota_capture import (
     bind_capture_selection_transport, capture_decision, handoff_decision,
     handle_native_context_command, handle_replan_view, prepare_quota_capture, native_context_capture_view,
-    native_context_activation,
+    native_context_activation, model_context_reference,
 )
 from ..control_plane.turn_driver.decision_capture import (
     context_capture_read_command,
@@ -431,6 +431,7 @@ def _emit_quota_result(
             payload = native_context_capture_view(payload, capture_directory, context_directory=directory,
                 metadata={"capture_sha256": capture_id, "recovery": recovery},
                 max_page_output_bytes=getattr(args, "max_page_output_bytes", None))
+            payload = model_context_reference(payload)
         except (QuotaDecisionTransportError, OSError, ValueError) as exc:
             if not isinstance(exc, QuotaDecisionTransportError):
                 exc = QuotaDecisionTransportError(

loopx/cli_commands/quota_capture.py

@@ -21,6 +21,7 @@
     NATIVE_CONTEXT_PAGE_BYTES, NATIVE_CONTEXT_PAGE_OUTPUT_BYTES, CONTEXT_CAPTURE_NAVIGATION_GUIDANCE,
     native_context_page_budget,
     read_context_capture_file as _read_capture,
+    read_context_read_reference, write_context_read_reference,
 )
 
 
@@ -42,7 +43,8 @@ def _capture_page(path: Path, raw: bytes, digest: str, *, offset: int, size: int
                   selection_sha256: str | None = None,
                   list_fields: bool = False,
                   list_depth: int | None = None,
-                  initial_read: bool = False) -> dict[str, Any]:
+                  initial_read: bool = False,
+                  reference_file: Path | None = None) -> dict[str, Any]:
     end = min(offset + size, len(raw))
     while end < len(raw) and raw[end] & 0xC0 == 0x80:
         end -= 1
@@ -71,11 +73,26 @@ def page(text: str) -> dict[str, Any]:
             # first bounded page rather than repeating them in file metadata.
             # Historical unbounded readers retain their exact wire shape.
             result["read_instruction"] = CONTEXT_CAPTURE_PAGE_GUIDANCE
+        if reference_file is not None and select is not None:
+            # Selected pages have never been full-input receipts. Give this
+            # compact transport a distinct schema so old observers cannot
+            # mistake an omitted digest for a full capture observation.
+            result["schema_version"] = "context_capture_selection_page_v1"
+            for key in ("sha256", "selection_sha256", "selected_pointers"):
+                result.pop(key, None)
+            result["reference_file"] = str(reference_file)
+            result["read_instruction"] = ("Selected detail only. Follow next_command for contiguous bytes; "
+                "the immutable reference pins both complete digests, which the CLI verifies on every page. "
+                "Offsets and EOF describe only this selected view. "
+                "EOF is not full-context retention, authority or freshness.")
         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=None if initial_read else select, selection_sha256=selection_sha256,
-                list_fields=list_fields, list_depth=list_depth, initial_read=initial_read)
+                select=None if initial_read or reference_file is not None else select,
+                selection_sha256=selection_sha256 if reference_file is None else None,
+                list_fields=list_fields if reference_file is None else False,
+                list_depth=list_depth if reference_file is None else None,
+                initial_read=initial_read if reference_file is None else False, reference_file=reference_file)
         return result
 
     def fits(result: dict[str, Any]) -> bool:
@@ -133,6 +150,16 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
     try:
         digest = getattr(args, "capture_sha256", None)
         value = getattr(args, "capture_file", None)
+        reference_file = getattr(args, "reference_file", None)
+        pinned_selection = None
+        if reference_file:
+            if value or digest or getattr(args, "codex_context_dir", None) or getattr(args, "codex_context_root", None):
+                raise ValueError("reference-file cannot be combined with capture or native session lookup")
+            if args.format != "json":
+                raise ValueError("reference-file requires JSON output")
+            reference_file, reference = read_context_read_reference(reference_file)
+            value, digest = reference["path"], reference["sha256"]
+            pinned_selection = reference.get("selection")
         offset, size = args.offset, args.page_bytes
         output_limit = getattr(args, "max_page_output_bytes", None)
         if output_limit is not None and (type(output_limit) is not int or output_limit <= 0
@@ -173,11 +200,24 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
             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)
+        selection_digest = getattr(args, "selection_sha256", None)
+        if pinned_selection is not None:
+            if selectors is not None or initial_read or list_fields or list_depth is not None or selection_digest is not None:
+                raise ValueError("a pinned selection reference cannot change selectors; use the original capture reference")
+            if (set(pinned_selection) != {"select", "initial_read", "list_fields", "list_depth", "sha256"}
+                    or type(pinned_selection["initial_read"]) is not bool or type(pinned_selection["list_fields"]) is not bool
+                    or not isinstance(pinned_selection["sha256"], str)
+                    or BARE_SHA256_PATTERN.fullmatch(pinned_selection["sha256"]) is None):
+                raise ValueError("reader selection reference is incomplete or unsupported")
+            selectors, initial_read = pinned_selection["select"], pinned_selection["initial_read"]
+            list_fields, list_depth = pinned_selection["list_fields"], pinned_selection["list_depth"]
+            selection_digest = pinned_selection["sha256"]
+            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("reader selection list-depth must be between 1 and 8 and requires list")
         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 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:
@@ -209,13 +249,22 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
                 raise ValueError("selected view digest changed; restore the original selection and reader")
             selection_digest = expected
             selectors = [item["pointer"] for item in views]
+            if reference_file is not None and pinned_selection is None:
+                try:
+                    reference_file = write_context_read_reference(path, digest, selection={
+                        "select": None if initial_read else selectors, "initial_read": initial_read,
+                        "list_fields": list_fields, "list_depth": list_depth, "sha256": expected})
+                except OSError:
+                    # A verified read must not require a writable capture root.
+                    # Legacy cursors pin the same capture, selection and digest.
+                    reference_file = None
         if offset > len(raw) or (offset < len(raw) and raw[offset] & 0xC0 == 0x80):
             raise ValueError("capture offset is outside the file or not a UTF-8 boundary; use the returned next offset")
         return _capture_page(path, raw, digest, offset=offset, size=size,
             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, initial_read=initial_read)
+            list_fields=list_fields, list_depth=list_depth, initial_read=initial_read, reference_file=reference_file)
     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.
@@ -247,6 +296,49 @@ def _capture_read_reference(value: dict[str, Any], *, page_bytes: int,
             page_bytes=page_bytes, max_page_output_bytes=output_bytes)}
 
 
+def model_context_reference(view: dict[str, Any]) -> dict[str, Any]:
+    """Present native navigation; machine references and journal bindings stay full.
+
+    The descriptor holds complete digests locally. Legacy explicit readers and
+    full-page observation receipts retain their original transport shape.
+    """
+    page_bytes, output_bytes = native_context_page_budget()
+    # Retain an explicitly smaller caller budget from the generated command.
+    words = shlex.split(view["command"])
+    for option in ("--page-bytes", "--max-page-output-bytes"):
+        if option in words:
+            if option == "--page-bytes":
+                page_bytes = int(words[words.index(option) + 1])
+            else:
+                output_bytes = int(words[words.index(option) + 1])
+
+    def reference(value: dict[str, Any], *, navigation: bool = False) -> dict[str, str]:
+        path = Path(value["path"])
+        reader = write_context_read_reference(path, value["sha256"], navigation=value if navigation else None)
+        return {"reference_file": str(reader), "command": context_capture_read_command(path, value["sha256"],
+            page_bytes=page_bytes, max_page_output_bytes=output_bytes, reference_file=reader)}
+
+    result = {"ok": True, "schema_version": "context_capture_navigation_v1", "mode": view["mode"],
+        **reference(view, navigation=True), "total_bytes": view["total_bytes"], "recovery": reference(view["recovery"]),
+        "instruction": view["instruction"]}
+    if "context_delivery" in view:
+        result["context_delivery"] = view["context_delivery"]
+    if "baseline_reference" in view:
+        result["baseline_reference"] = reference(view["baseline_reference"])
+        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"]))
+    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.")
+    if len(_native_output_bytes(result)) > min(output_bytes, 7900):
+        return view  # Preserve usable legacy recovery under a very small ceiling.
+    return result
+
+
 def _native_context_reference(projected: dict[str, Any], reference: dict[str, Any], *,
                               mode: str, current: dict[str, Any], current_document: dict[str, Any],
                               total_bytes: int, metadata: dict[str, Any] | None,
@@ -529,7 +621,13 @@ def project_saved_context_capture(args: argparse.Namespace) -> dict[str, Any]:
             max_page_output_bytes=getattr(args, "max_page_output_bytes", None))
         reference = {key: native_view[key] for key in ("path", "sha256", "command")}
         mode, target = native_view["mode"], native_view.get("target")
+        native_view = model_context_reference(native_view)
     except (QuotaDecisionTransportError, OSError, ValueError, TypeError, RecursionError) as exc:
+        if native_view is not None and not isinstance(exc, QuotaDecisionTransportError):
+            # The native view and journal are already valid. Publishing the
+            # shorter reader metadata is optional; retain their exact legacy
+            # recovery rather than losing the original capture or admitting again.
+            return native_view
         if isinstance(exc, QuotaDecisionTransportError) or (activation["enabled"] and not native_prepared):
             if not isinstance(exc, QuotaDecisionTransportError):
                 exc = QuotaDecisionTransportError(
@@ -563,7 +661,7 @@ def project_saved_context_capture(args: argparse.Namespace) -> dict[str, Any]:
         "instruction": "Presentation of the original decision only; authority and freshness are unchanged. "
             + (CONTEXT_CAPTURE_NAVIGATION_GUIDANCE if activation["enabled"] else CONTEXT_CAPTURE_READ_GUIDANCE)}
     if native_view is not None:
-        result.update(native_view)
+        return native_view
     return result
 
 

loopx/cli_commands/quota_registration.py

@@ -207,6 +207,7 @@ def register_quota_command(
               "Waits at most 120s for a digest-bound release; failure retains the original receipt."),
     )
     quota_parser.add_argument("--capture-file", help="For context-read/context-project: original captured JSON file; no guard is evaluated.")
+    quota_parser.add_argument("--reference-file", help="For context-read: immutable private reader reference holding the complete capture/selection digests. No admission or live Goal lookup.")
     # Host preflight captures the complete native input, then presents that same
     # immutable view after the model's lifecycle hooks establish continuity.
     quota_parser.add_argument("--defer-context-presentation", action="store_true", help=argparse.SUPPRESS)