From a2aa667461b6e8ba3fe1fecee8b59746b687f147 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sun, 1 Nov 2026 16:14:27 +0530
Subject: [PATCH 1/5] fix(context): keep reader digests behind native
 references

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 docs/reference/required-work-context.md       |  24 ++++
 loopx/cli_commands/context_machine.py         |   2 +-
 loopx/cli_commands/quota.py                   |   3 +-
 loopx/cli_commands/quota_capture.py           |  99 +++++++++++++-
 loopx/cli_commands/quota_registration.py      |   1 +
 .../turn_driver/decision_capture.py           |  56 +++++++-
 .../test_context_delivery_cli.py              |  23 ++--
 .../test_context_machine_configuration.py     |   7 +-
 .../test_context_read_reference.py            | 125 ++++++++++++++++++
 9 files changed, 320 insertions(+), 20 deletions(-)
 create mode 100644 tests/control_plane/test_context_read_reference.py

diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md
index 0f69b6d3fd..842b03ccb1 100644
--- a/docs/reference/required-work-context.md
+++ b/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
diff --git a/loopx/cli_commands/context_machine.py b/loopx/cli_commands/context_machine.py
index 50d9417f81..2fe133254b 100644
--- a/loopx/cli_commands/context_machine.py
+++ b/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"
         )
     )
 
diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py
index 0f20b8249b..b3026a9664 100644
--- a/loopx/cli_commands/quota.py
+++ b/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(
diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py
index f98f155bb4..7507ef7152 100644
--- a/loopx/cli_commands/quota_capture.py
+++ b/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,17 @@ 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:
+                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})
         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 +291,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,
@@ -563,7 +650,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 model_context_reference(native_view)
     return result
 
 
diff --git a/loopx/cli_commands/quota_registration.py b/loopx/cli_commands/quota_registration.py
index 91969dc37d..d6ce128f8d 100644
--- a/loopx/cli_commands/quota_registration.py
+++ b/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)
diff --git a/loopx/control_plane/turn_driver/decision_capture.py b/loopx/control_plane/turn_driver/decision_capture.py
index 29487f30e5..74aad434c3 100644
--- a/loopx/control_plane/turn_driver/decision_capture.py
+++ b/loopx/control_plane/turn_driver/decision_capture.py
@@ -212,7 +212,8 @@ def context_capture_read_command(path: Path, digest: str, *, offset: int = 0,
                                  selection_sha256: str | None = None,
                                  list_fields: bool = False,
                                  list_depth: int | None = None,
-                                 initial_read: bool = False) -> str:
+                                 initial_read: bool = False,
+                                 reference_file: Path | None = None) -> str:
     from ...command_invocation import loopx_cli_argv
 
     if page_bytes is None:
@@ -220,7 +221,8 @@ def context_capture_read_command(path: Path, digest: str, *, offset: int = 0,
         if max_page_output_bytes is None:
             max_page_output_bytes = NATIVE_CONTEXT_PAGE_OUTPUT_BYTES
     return shlex.join([*loopx_cli_argv(), "--format", "json", "quota", "context-read",
-        "--capture-file", str(path), "--capture-sha256", digest,
+        *(["--reference-file", str(reference_file)] if reference_file is not None
+          else ["--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 []),
@@ -230,6 +232,56 @@ def context_capture_read_command(path: Path, digest: str, *, offset: int = 0,
         *(["--max-page-output-bytes", str(max_page_output_bytes)] if max_page_output_bytes is not None else [])])
 
 
+def write_context_read_reference(path: Path, digest: str, *, selection: dict[str, Any] | None = None,
+                                 navigation: dict[str, Any] | None = None) -> Path:
+    """Immutable reader IO metadata, not a new identity or authority store.
+
+    A short filename locates the document; only its complete stored digests are
+    checked. A collision or replaced document fails closed, never overwrites.
+    """
+    payload = {"schema_version": "context_capture_read_reference_v1", "path": str(path), "sha256": digest}
+    if selection is not None:
+        payload["selection"] = selection
+    if navigation is not None:
+        payload["navigation"] = navigation
+    suffix = ("." + hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()).hexdigest()[:16]
+              if selection is not None or navigation is not None else "")
+    reference = path.with_name(path.stem + suffix + ".reader.json")
+    try:
+        write_decision_capture(reference, payload)
+    except FileExistsError:
+        existing = read_context_read_reference(str(reference))[1]
+        if json.dumps(existing, sort_keys=True) != json.dumps(payload, sort_keys=True):
+            raise ValueError("original reader reference changed; restore it without rerunning admission") from None
+    return reference
+
+
+def read_context_read_reference(value: str) -> tuple[Path, dict[str, Any]]:
+    """Read one private physical descriptor; the caller verifies capture bytes."""
+    from ..effect_runtime import MAX_REQUEST_BYTES
+
+    path = Path(value).expanduser().absolute()
+    path = path.parent.resolve(strict=True) / path.name
+    if path.is_symlink():
+        raise ValueError("reader reference must be a physical file, not a symlink")
+    descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0))
+    with os.fdopen(descriptor, "rb") as stream:
+        info = os.fstat(stream.fileno())
+        if (not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() or info.st_mode & 0o077
+                or info.st_size > MAX_REQUEST_BYTES):
+            raise ValueError("reader reference must be a private regular file owned by this user")
+        raw = stream.read(MAX_REQUEST_BYTES + 1)
+    if len(raw) > MAX_REQUEST_BYTES:
+        raise ValueError("reader reference exceeds the existing transport limit")
+    payload = json.loads(raw)
+    if (not isinstance(payload, dict) or payload.get("schema_version") != "context_capture_read_reference_v1"
+            or not isinstance(payload.get("path"), str) or not isinstance(payload.get("sha256"), str)
+            or BARE_SHA256_PATTERN.fullmatch(payload["sha256"]) is None
+            or ("selection" in payload and not isinstance(payload["selection"], dict))):
+        raise ValueError("reader reference is incomplete or unsupported")
+    return path, payload
+
+
 def read_context_capture_file(value: str, digest: str, *, private: bool = False) -> tuple[Path, bytes, dict[str, Any]]:
     """One bounded physical-file/hash check for readers and native observations."""
     from ..effect_runtime import MAX_REQUEST_BYTES
diff --git a/tests/control_plane/test_context_delivery_cli.py b/tests/control_plane/test_context_delivery_cli.py
index 5974effe93..25b8dd728d 100644
--- a/tests/control_plane/test_context_delivery_cli.py
+++ b/tests/control_plane/test_context_delivery_cli.py
@@ -17,6 +17,7 @@
 
 from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime, read_fixture_todos
 from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
+from loopx.control_plane.turn_driver.decision_capture import read_context_read_reference
 from loopx.control_plane.quota.turn_envelope import (
     build_turn_envelope, prepare_context_delivery, restore_context_delivery,
 )
@@ -495,12 +496,13 @@ def consume(page):
         args = build_parser().parse_args(shlex.split(wire["recovery"]["command"])[1:])
         assert args.max_page_output_bytes == limit
         observed = consume(read_context_capture(args))
-        assert hashlib.sha256(observed).hexdigest() == wire["recovery"]["sha256"]
+        recovery = read_context_read_reference(wire["recovery"]["reference_file"])[1]
+        assert hashlib.sha256(observed).hexdigest() == recovery["sha256"]
         assert json.loads(observed)["heartbeat_receipt"]["turn_instance_id"] == "native-full"
         if output_limit is not None:
             rc, projected = _run_cli(registry, runtime, "quota", "context-project", "--codex-app",
                 "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--turn-instance-id", "native-full",
-                "--capture-file", wire["recovery"]["path"], "--capture-sha256", wire["recovery"]["sha256"],
+                "--capture-file", recovery["path"], "--capture-sha256", recovery["sha256"],
                 "--decision-output-root", str(projection_root), "--context-delivery", "on",
                 "--codex-context-dir", str(root), "--max-page-output-bytes", str(limit), cwd=project)
             assert rc == 0 and projected["ok"], projected
@@ -552,7 +554,8 @@ def consume(view):
             if page["eof"]:
                 break
             page = read_context_capture(build_parser().parse_args(shlex.split(page["next_command"])[1:]))
-        assert hashlib.sha256(observed).hexdigest() == view["recovery"]["sha256"]
+        recovery = read_context_read_reference(view["recovery"]["reference_file"])[1]
+        assert hashlib.sha256(observed).hexdigest() == recovery["sha256"]
         return json.loads(observed)
 
     first = guard("native-work-first")
@@ -571,7 +574,8 @@ def consume(view):
         "--agent-id", AGENT_ID, "--todo-id", TODO_ID, "--turn-instance-id", "native-work-first",
         "--source", "heartbeat", "--execute", cwd=project)
     assert rc == 0 and spent["ok"], spent
-    original = Path(first["recovery"]["path"]).read_bytes()
+    original_reference = read_context_read_reference(first["recovery"]["reference_file"])[1]
+    original = Path(original_reference["path"]).read_bytes()
     context.codex_context_hook(str(root), {**event, "hook_event_name": "PreCompact"})
     context.codex_context_hook(str(root), {**event, "source": "compact"})
     assert "candidate" not in context._state(root)
@@ -580,12 +584,14 @@ def consume(view):
     assert current["mode"] == "full" and "candidate" not in context._state(root)
     body = consume(current)
     assert body["heartbeat_receipt"]["turn_instance_id"] == "native-work-next"
-    assert context._state(root)["candidate"]["target"] == current["target"]
-    assert Path(first["recovery"]["path"]).read_bytes() == original
+    current_machine = read_context_read_reference(current["reference_file"])[1]["navigation"]
+    assert context._state(root)["candidate"]["target"] == current_machine["target"]
+    assert Path(original_reference["path"]).read_bytes() == original
     command_workspace = tmp_path / "independent-checkout"
     command_workspace.mkdir()
     warm = guard("native-work-next", cwd=command_workspace)
-    assert warm["mode"] == "delta" and warm["target"] != current["target"]
+    warm_machine = read_context_read_reference(warm["reference_file"])[1]["navigation"]
+    assert warm["mode"] == "delta" and warm_machine["target"] != current_machine["target"]
     assert context._state(root)["cwd"] == str(project)
     assert _heartbeat_receipt_count(runtime, "native-work-first") == 1
     assert _heartbeat_receipt_count(runtime, "native-work-next") == 1
@@ -634,7 +640,8 @@ def unavailable_directory(**kwargs):
         assert printed[0]["captured_decision"] == {"path": str(tmp_path / "decision.json")}
     else:
         assert len(printed) == 1 and "content" not in printed[0]
-        assert json.loads(Path(printed[0]["path"]).read_text()) == {"native_projection": original}
+        reference = read_context_read_reference(printed[0]["reference_file"])[1]
+        assert json.loads(Path(reference["path"]).read_text()) == {"native_projection": original}
     assert json.loads((tmp_path / "decision.json").read_text()) == original
 
 
diff --git a/tests/control_plane/test_context_machine_configuration.py b/tests/control_plane/test_context_machine_configuration.py
index 8b85cf0479..d9981136ba 100644
--- a/tests/control_plane/test_context_machine_configuration.py
+++ b/tests/control_plane/test_context_machine_configuration.py
@@ -13,6 +13,7 @@
 import pytest
 
 from loopx.capabilities.machine_configuration.context_delivery import CONTEXT_DELIVERY_MACHINE_SCHEMA
+from loopx.control_plane.turn_driver.decision_capture import read_context_read_reference
 from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime
 from tests.control_plane.test_quota_settlement_cli import (
     AGENT_ID, GOAL_ID, TODO_ID, _heartbeat_receipt_count, _initialize_fixture_authority,
@@ -240,12 +241,13 @@ def test_machine_on_reaches_real_guard_and_same_capture_readback_without_readmis
         "--agent-id", AGENT_ID, "--todo-id", TODO_ID, "--turn-instance-id", "machine-original-turn"]
     code, wire = _run_cli(registry, runtime, *args, cwd=project)
     assert code == 0 and wire["mode"] == "full", wire
-    recovery = wire["recovery"]
+    recovery = read_context_read_reference(wire["recovery"]["reference_file"])[1]
     raw = Path(recovery["path"]).read_bytes()
     assert hashlib.sha256(raw).hexdigest() == recovery["sha256"]
     assert json.loads(raw)["goal_id"] == GOAL_ID
 
     def complete_reader(view):
+        view = read_context_read_reference(view["reference_file"])[1]["navigation"]
         # Native delivery advertises a file; only its reader pages are content.
         observed = bytearray()
         while True:
@@ -267,13 +269,14 @@ def complete_reader(view):
     code, warm = _run_cli(registry, runtime, *args, cwd=project)
     assert code == 0 and warm["mode"] == "delta", warm
     complete_reader(warm)
-    old_generation = warm["target"]["continuity_generation"]
+    old_generation = read_context_read_reference(warm["reference_file"])[1]["navigation"]["target"]["continuity_generation"]
     old_configuration = (runtime / "machine/configuration.json").read_bytes()
     assert configure(registry, runtime, tmp_path, "off")[0] == 0
     assert configure(registry, runtime, tmp_path, "on")[0] == 0
     assert (runtime / "machine/configuration.json").read_bytes() == old_configuration
     code, fresh = _run_cli(registry, runtime, *args, cwd=project)
     assert code == 0 and fresh["mode"] == "full", fresh
+    fresh = read_context_read_reference(fresh["reference_file"])[1]["navigation"]
     assert fresh["target"]["continuity_generation"] != old_generation
     code, page = _run_cli(registry, runtime, "quota", "context-read", "--goal-id", GOAL_ID,
         "--agent-id", AGENT_ID, cwd=project)
diff --git a/tests/control_plane/test_context_read_reference.py b/tests/control_plane/test_context_read_reference.py
new file mode 100644
index 0000000000..b699e6029d
--- /dev/null
+++ b/tests/control_plane/test_context_read_reference.py
@@ -0,0 +1,125 @@
+"""Reader references hide repeated digests without weakening byte verification."""
+from argparse import Namespace
+import hashlib
+import json
+from pathlib import Path
+import shlex
+
+import pytest
+
+from loopx.cli_commands.quota_capture import model_context_reference, native_context_capture_view, read_context_capture
+from loopx.control_plane.turn_driver.codex_context import _context_pages
+from loopx.control_plane.turn_driver.decision_capture import (
+    capture_page_range, read_context_read_reference, write_context_read_reference,
+)
+
+
+def read(command, **overrides):
+    words = shlex.split(command)
+    args = dict(format="json", offset=0, page_bytes=64000, max_page_output_bytes=64000)
+    for name in ("reference-file", "capture-file", "capture-sha256", "offset", "page-bytes", "max-page-output-bytes"):
+        if "--" + name in words:
+            value = words[words.index("--" + name) + 1]
+            args[name.replace("-", "_")] = int(value) if name in {"offset", "page-bytes", "max-page-output-bytes"} else value
+    args["initial_read"] = "--initial-read" in words
+    args.update(overrides)
+    return read_context_capture(Namespace(**args))
+
+
+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)
+    return payload, view, model_context_reference(view)
+
+
+def test_native_navigation_keeps_digests_in_immutable_reference(tmp_path):
+    payload, machine, model = fixture(tmp_path)
+    raw = Path(machine["path"]).read_bytes()
+    assert json.loads(raw) == payload
+    assert machine["sha256"] == hashlib.sha256(raw).hexdigest()
+    assert machine["sha256"] not in json.dumps(model)
+    assert "--capture-sha256" not in json.dumps(model)
+    _, descriptor = read_context_read_reference(model["reference_file"])
+    assert descriptor["path"] == machine["path"] and descriptor["sha256"] == machine["sha256"]
+    assert descriptor["navigation"] == machine  # complete machine metadata remains recoverable
+    assert Path(model["reference_file"]).stat().st_mode & 0o077 == 0
+    # Same immutable source and full-page observer compatibility.
+    page = read(model["command"])
+    assert page["schema_version"] == "context_capture_page_v1"
+    assert capture_page_range(page, raw=raw, digest=machine["sha256"]) == (0, len(raw))
+
+
+def test_selected_continuations_hide_both_digests_and_preserve_exact_values(tmp_path):
+    payload, machine, model = fixture(tmp_path)
+    page = read(model["recovery"]["command"], select=["/selected_todo", "/criterion"],
+        max_page_output_bytes=3000, page_bytes=3000)
+    parts = []
+    while True:
+        assert page["ok"] and page["schema_version"] == "context_capture_selection_page_v1"
+        assert not any(key in page for key in ("sha256", "selection_sha256", "selected_pointers"))
+        assert _context_pages(page) == []  # selected EOF is never a full-input receipt
+        parts.append(page["content"])
+        if page["eof"]:
+            break
+        command = page["next_command"]
+        assert "--selection-sha256" not in command and "--capture-sha256" not in command
+        assert page["next_offset"] == len("".join(parts).encode())
+        page = read(command)
+    values = {item["pointer"]: item["value"] for item in json.loads("".join(parts))["selections"]}
+    assert values == {"/selected_todo": payload["selected_todo"], "/criterion": payload["criterion"]}
+    _, descriptor = read_context_read_reference(page["reference_file"])
+    assert descriptor["sha256"] == machine["sha256"] and len(descriptor["selection"]["sha256"]) == 64
+    changed = read(model["command"], reference_file=page["reference_file"], select=["/criterion"])
+    assert changed["ok"] is False and "cannot change selectors" in changed["reason"]
+
+
+def test_initial_reference_rejects_changed_selection_profile(tmp_path, monkeypatch):
+    _, _, model = fixture(tmp_path)
+    first = read(model["first_read_command"], max_page_output_bytes=3000, page_bytes=3000)
+    assert first["ok"] and not first["eof"]
+    from loopx.control_plane.quota import turn_envelope
+
+    original = turn_envelope._turn_envelope_result
+
+    def changed(operation, **kwargs):
+        result = original(operation, **kwargs)
+        result["selections"].pop()
+        return result
+
+    monkeypatch.setattr(turn_envelope, "_turn_envelope_result", changed)
+    rejected = read(first["next_command"])
+    assert rejected["ok"] is False and "digest changed" in rejected["reason"]
+    assert "content" not in rejected
+
+
+@pytest.mark.parametrize("damage", ["capture", "selection", "symlink", "permissions"])
+def test_reference_damage_fails_closed(tmp_path, damage):
+    _, machine, model = fixture(tmp_path)
+    page = read(model["command"], select=["/criterion"])
+    reference = Path(page["reference_file"])
+    if damage == "capture":
+        Path(machine["path"]).write_text('{"changed":true}\n')
+    elif damage == "selection":
+        payload = json.loads(reference.read_text())
+        payload["selection"]["sha256"] = "0" * 64
+        reference.write_text(json.dumps(payload))
+    elif damage == "symlink":
+        original = reference.with_suffix(".saved")
+        reference.rename(original)
+        reference.symlink_to(original)
+    else:
+        reference.chmod(0o644)
+    rejected = read(model["command"], reference_file=str(reference))
+    assert rejected["ok"] is False and "content" not in rejected
+
+
+def test_reference_publication_never_overwrites_an_existing_document(tmp_path):
+    path = tmp_path / "capture.json"
+    path.write_text('{}\n')
+    original = write_context_read_reference(path, "a" * 64)
+    assert write_context_read_reference(path, "a" * 64) == original
+    with pytest.raises(ValueError, match="reference changed"):
+        write_context_read_reference(path, "b" * 64)

From 0de0888927574751bd4d454ac0aa8db2890dc6dc Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sun, 1 Nov 2026 16:58:01 +0530
Subject: [PATCH 2/5] test(context): follow compact references through native
 recovery

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../control_plane/codex_checkpoint_fixture.py | 16 +++++++-
 .../test_context_capture_selection.py         | 13 ++++++-
 .../test_saved_context_project.py             | 39 ++++++++++++-------
 3 files changed, 51 insertions(+), 17 deletions(-)

diff --git a/tests/control_plane/codex_checkpoint_fixture.py b/tests/control_plane/codex_checkpoint_fixture.py
index 5ac1996930..4d22358a7f 100644
--- a/tests/control_plane/codex_checkpoint_fixture.py
+++ b/tests/control_plane/codex_checkpoint_fixture.py
@@ -45,7 +45,7 @@ def delivered_context(request, digest):
     return raw.decode().rstrip("\n")
 
 
-def last_context_page(request, *, allow_reference=False):
+def last_context_page(request, *, allow_reference=False, allow_selection=False):
     """Observe the latest tool result; a saved-file reference is not a page."""
     for item in reversed(request["input"]):
         output = item.get("output")
@@ -59,6 +59,8 @@ def last_context_page(request, *, allow_reference=False):
                     value = json.loads(value["output"])
                 if isinstance(value, dict) and (
                     value.get("schema_version") == "context_capture_page_v1"
+                    or (allow_selection and value.get("schema_version") == "context_capture_selection_page_v1")
+                    or (allow_reference and value.get("schema_version") == "context_capture_navigation_v1")
                     or (allow_reference and value.get("mode") in {"full", "delta"}
                         and all(key in value for key in ("path", "sha256", "command", "target", "recovery")))
                 ):
@@ -113,7 +115,8 @@ def follow(self, request):
         if self.pending["descriptor"] is None:
             assert page["offset"] == 0
             if reference := self.pending.get("reference"):
-                assert page["sha256"] == reference["sha256"]
+                if "sha256" in reference:
+                    assert page["sha256"] == reference["sha256"]
                 # The generic saved-file reader returns bytes and cursors;
                 # delivery identity stays in the original visible reference.
                 self.pending["descriptor"] = {**reference, **page}
@@ -126,6 +129,15 @@ def follow(self, request):
             return self.tool(page["next_command"])
         raw = delivered_context(request, first["sha256"])
         assert len(raw.encode()) in {page["total_bytes"], page["total_bytes"] - 1}
+        if self.pending.get("reference", {}).get("schema_version") == "context_capture_navigation_v1":
+            # Resolve oracle metadata only after complete model-visible byte
+            # coverage; a reference never seeds the retained-input baseline.
+            from loopx.control_plane.turn_driver.decision_capture import read_context_read_reference
+
+            _, descriptor = read_context_read_reference(first["reference_file"])
+            navigation = descriptor["navigation"]
+            assert navigation["sha256"] == first["sha256"]
+            first = {**first, **navigation}
         # The original reader owns digest verification. This separate oracle
         # checks the actual complete model input against its declared source.
         assert raw == Path(first["path"]).read_text().rstrip("\n")
diff --git a/tests/control_plane/test_context_capture_selection.py b/tests/control_plane/test_context_capture_selection.py
index 2fc82a5602..8610b210be 100644
--- a/tests/control_plane/test_context_capture_selection.py
+++ b/tests/control_plane/test_context_capture_selection.py
@@ -102,17 +102,26 @@ def unavailable(*args, **kwargs):
     assert page["ok"] and json.loads(page["content"]) == original
 
 
-def test_native_reference_reader_uses_real_pages_before_recording_delivery(tmp_path):
+@pytest.mark.parametrize("compact_reference", [False, True])
+def test_native_reference_reader_uses_real_pages_before_recording_delivery(tmp_path, compact_reference):
     path, raw, _ = fixture_capture(tmp_path)
     options = args(path, raw)
     options.select = None
     reference = {"ok": True, "mode": "full", "path": str(path),
         "sha256": options.capture_sha256, "command": "read saved capture",
-        "target": {"capture_id": "synthetic-source"}, "recovery": {"path": str(path)}}
+        "target": {"capture_id": "synthetic-source"},
+        "total_bytes": len(raw), "instruction": "Read the current saved input.",
+        "recovery": {"path": str(path), "sha256": options.capture_sha256}}
+    if compact_reference:
+        from loopx.cli_commands.quota_capture import model_context_reference
+
+        reference = model_context_reference(reference)
+        assert "sha256" not in reference and "target" not in reference
     request = {"input": [{"type": "function_call_output", "output": json.dumps(reference)}]}
     reader = NativeContextReader()
     reader.begin("admit current Turn", "source")
     reader.follow(request)
+    assert not reader.deliveries and not reader.retained
     page_count = 0
     while True:
         page = read_context_capture(options)
diff --git a/tests/control_plane/test_saved_context_project.py b/tests/control_plane/test_saved_context_project.py
index e1ac717efb..be096a0e06 100644
--- a/tests/control_plane/test_saved_context_project.py
+++ b/tests/control_plane/test_saved_context_project.py
@@ -11,6 +11,15 @@
 from tests.control_plane.test_quota_settlement_cli import _run_cli
 
 
+def machine_reference_oracle(view):
+    """Inspect complete metadata as an oracle, never as a model read receipt."""
+    if view.get("schema_version") == "context_capture_navigation_v1":
+        from loopx.control_plane.turn_driver.decision_capture import read_context_read_reference
+
+        return read_context_read_reference(view["reference_file"])[1]["navigation"]
+    return view
+
+
 @pytest.mark.parametrize("failure", [None, "digest", "goal", "agent", "turn", "symlink", "object"])
 def test_saved_projection_checks_original_before_any_publication(tmp_path, failure):
     original = {"ok": True, "goal_id": "saved-goal", "agent_identity": {"agent_id": "saved-agent"},
@@ -116,8 +125,9 @@ def test_saved_projection_manages_private_cache_without_authority_io(tmp_path, m
     assert path.read_bytes() == raw and not registry.exists() and not runtime.exists()
     if enabled:
         assert "content" not in result and "offset" not in result
-        assert "index_command" in result and "select_command" in result
-        assert json.loads(Path(result["path"]).read_bytes()) == original
+        machine = machine_reference_oracle(result)
+        assert "index_command" in machine and "select_command" in machine
+        assert json.loads(Path(machine["path"]).read_bytes()) == original
         actual_root = context if cache_location == "default" else root
         assert actual_root.stat().st_mode & 0o077 == 0
         assert list(actual_root.glob("context-*"))
@@ -171,9 +181,10 @@ def test_saved_projection_reference_allows_optional_exact_read_and_restart(tmp_p
     # read remains lossless and paginated, without changing the saved decision.
     assert "content" not in result and "offset" not in result
     assert len(json.dumps(result, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= 8192
-    delivered = Path(result["path"]).read_bytes()
+    machine = machine_reference_oracle(result)
+    delivered = Path(machine["path"]).read_bytes()
     delivered_digest = hashlib.sha256(delivered).hexdigest()
-    assert result["sha256"] == delivered_digest and json.loads(delivered) == json.loads(raw)
+    assert machine["sha256"] == delivered_digest and json.loads(delivered) == json.loads(raw)
     combined = b""
     command = result["command"]
     while True:
@@ -228,7 +239,8 @@ def fail_optional_wire(path, payload, **kwargs):
         "--capture-file", str(path), "--capture-sha256", digest, "--goal-id", "saved-goal",
         "--agent-id", "saved-agent", "--turn-instance-id", "saved-turn", "--context-delivery", "on",
         "--codex-app", "--codex-context-dir", str(context), "--decision-output-root", str(root)])
-    result = project_saved_context_capture(args)
+    wire = project_saved_context_capture(args)
+    result = machine_reference_oracle(wire)
     assert result["ok"] and result["mode"] == "full"
     recovered_path, recovered_digest = codex_context.read_codex_context_reference(str(context),
         goal_id="saved-goal", agent_id="saved-agent")
@@ -237,7 +249,7 @@ def fail_optional_wire(path, payload, **kwargs):
         assert result["path"] == recovered_path and result["sha256"] == recovered_digest
     else:
         assert "content" not in result and result["recovery"]["sha256"] == recovered_digest
-        argv = shlex.split(result["command"])
+        argv = shlex.split(wire["command"])
         code, page = _run_cli(tmp_path / "absent.json", tmp_path / "absent-runtime",
             *argv[argv.index("quota"):], cwd=tmp_path)
         assert code == 0 and page["offset"] == 0 and json.loads(page["content"]) == original
@@ -442,7 +454,7 @@ def consume(wire):
             data.extend(page["content"].encode())
             codex_context_hook(str(native), {**event, "hook_event_name": "PostToolUse", "tool_response": json.dumps(page)})
             if page["eof"]:
-                assert hashlib.sha256(data).hexdigest() == wire["sha256"]
+                assert hashlib.sha256(data).hexdigest() == machine_reference_oracle(wire)["sha256"]
                 return json.loads(data)
             argv = shlex.split(page["next_command"])
             rc, page = _run_cli(registry, runtime, *argv[argv.index("quota"):], cwd=project)
@@ -474,7 +486,7 @@ def consume(wire):
         assert len(json.dumps(bound, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= 32000
         delivered = consume(bound)
         if not compacted:
-            assert delivered["base"]["capture_id"] == cold["target"]["capture_id"]
+            assert delivered["base"]["capture_id"] == machine_reference_oracle(cold)["target"]["capture_id"]
     else:
         assert bound["mode"] == "should-run" and not list(native.iterdir())
     assert _heartbeat_receipt_count(runtime, turn) == 2 and _spend_run_count(runtime) == 0
@@ -572,10 +584,11 @@ def response(request, number, message):
                     if item.get("type") in ("function_call_output", "custom_tool_call_output")][-1]
                 if isinstance(output, list):
                     output = [part["text"] for part in output if part.get("type") == "input_text"][-1]
-                descriptor = json.loads(output.split("\nOutput:\n", 1)[-1])
+                navigation = json.loads(output.split("\nOutput:\n", 1)[-1])
+                descriptor = machine_reference_oracle(navigation)
                 assert descriptor["ok"] and descriptor["mode"] == ("full" if name == "cold" else "delta")
                 if benchmark:
-                    assert len(json.dumps(descriptor, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= 32000
+                    assert len(json.dumps(navigation, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= 32000
                 data = Path(descriptor["path"]).read_bytes()
                 assert hashlib.sha256(data).hexdigest() == descriptor["sha256"]
                 published = json.loads(data)
@@ -586,12 +599,12 @@ def response(request, number, message):
                 # Follow the public reader's actual page cursor rather than
                 # guessing offsets from the capture size. This keeps each
                 # request's real page budget and avoids overlapping reads.
-                queue.append(descriptor["command"])
+                queue.append(navigation["command"])
                 if progressive:
                     # 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 = [descriptor["index_command"], descriptor["recovery"]["command"]
+                    queue = [navigation["recovery"]["command"] + " --list", navigation["recovery"]["command"]
                         + " --select /goal_id --select /heartbeat_receipt"]
                 phase = name + "_read"
                 if not queue:
@@ -600,7 +613,7 @@ def response(request, number, message):
             else:
                 name = phase.removesuffix("_read")
                 descriptor = observed[name][1]
-                page = last_context_page(request)
+                page = last_context_page(request, allow_selection=progressive)
                 assert page and page["ok"]
                 if progressive:
                     values = {item["pointer"]: item["value"] for item in json.loads(page["content"])["selections"]}

From 809398af749ab6a0be210a830e4fe45c41fc436f Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sun, 1 Nov 2026 17:09:58 +0530
Subject: [PATCH 3/5] test(context): await yielded native commands before
 accepting work

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../tests/test_native_checkpoint_handoff.py   |  6 +++-
 .../control_plane/codex_checkpoint_fixture.py | 34 +++++++++++++++++++
 .../test_context_capture_selection.py         | 27 ++++++++++++++-
 .../test_saved_context_project.py             |  7 +++-
 4 files changed, 71 insertions(+), 3 deletions(-)

diff --git a/benchmark/tests/test_native_checkpoint_handoff.py b/benchmark/tests/test_native_checkpoint_handoff.py
index 53a0e303ed..26c65ca93a 100644
--- a/benchmark/tests/test_native_checkpoint_handoff.py
+++ b/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:
diff --git a/tests/control_plane/codex_checkpoint_fixture.py b/tests/control_plane/codex_checkpoint_fixture.py
index 4d22358a7f..0055f876fe 100644
--- a/tests/control_plane/codex_checkpoint_fixture.py
+++ b/tests/control_plane/codex_checkpoint_fixture.py
@@ -2,6 +2,7 @@
 import json
 import os
 from pathlib import Path
+import re
 import shlex
 import threading
 import time
@@ -74,6 +75,36 @@ def last_context_page(request, *, allow_reference=False, allow_selection=False):
     return None
 
 
+def native_process_continuation(request, identity, *, code_mode=False):
+    """Follow a yielded native command, using only its latest visible output."""
+    for item in reversed(request["input"]):
+        if (output := item.get("output")) is None:
+            continue
+        parts = [p.get("text", "") for p in output] if isinstance(output, list) else [output]
+        for part in reversed(parts):
+            if not isinstance(part, str):
+                continue
+            try:
+                wrapper = json.loads(part)
+                if isinstance(wrapper, dict) and isinstance(wrapper.get("output"), str):
+                    part = wrapper["output"]
+            except ValueError:
+                pass
+            match = re.search(r"(?m)^Process running with session ID (\d+)\s*$", part)
+            if match:
+                options = {"session_id": int(match[1]), "chars": "", "yield_time_ms": 1000,
+                           "max_output_tokens": 30000}
+                base = {"id": f"fc_process_{identity}", "call_id": f"call_process_{identity}"}
+                if code_mode:
+                    options["max_output_tokens"] = 22000
+                    return {**base, "type": "custom_tool_call", "name": "exec",
+                            "input": '// @exec: {"max_output_tokens":24000}\ntext((await tools.write_stdin('
+                                + json.dumps(options) + ')).output);'}
+                return {**base, "type": "function_call", "name": "write_stdin", "arguments": json.dumps(options)}
+        return None  # Never borrow a session from an older result.
+    return None
+
+
 class NativeContextReader:
     """Static inference which follows only model-visible native reader cursors.
 
@@ -103,6 +134,9 @@ def begin(self, command, label):
     def follow(self, request):
         if self.pending is None:
             return None
+        if continuation := native_process_continuation(request, f"{self.identity}_{self.calls + 1}"):
+            self.calls += 1
+            return continuation
         page = last_context_page(request, allow_reference=True)
         assert page and page["ok"], "Native reader output must reach the model intact"
         if page.get("schema_version") != "context_capture_page_v1":
diff --git a/tests/control_plane/test_context_capture_selection.py b/tests/control_plane/test_context_capture_selection.py
index 8610b210be..0521909e48 100644
--- a/tests/control_plane/test_context_capture_selection.py
+++ b/tests/control_plane/test_context_capture_selection.py
@@ -7,7 +7,9 @@
 
 import pytest
 
-from tests.control_plane.codex_checkpoint_fixture import NativeContextReader, last_context_page
+from tests.control_plane.codex_checkpoint_fixture import (
+    NativeContextReader, last_context_page, native_process_continuation,
+)
 from loopx.cli_commands.quota_capture import read_context_capture
 from loopx.control_plane.effect_runtime import EffectRuntimeRemoteError, EffectRuntimeStartupError
 from loopx.control_plane.turn_driver.decision_capture import capture_page_range
@@ -47,6 +49,25 @@ def fixture_capture(tmp_path):
     return path, raw, original
 
 
+@pytest.mark.parametrize("code_mode", [False, True])
+@pytest.mark.parametrize("wrapped", [False, True])
+def test_native_process_continuation_uses_only_the_latest_tool_result(code_mode, wrapped):
+    # Literal native output from a slow command; completion must not be
+    # inferred from an earlier page, an EOF, or the command's mere dispatch.
+    output = "Wall time: 10.0832 seconds\nProcess running with session ID 92241\nOutput:\n"
+    if wrapped:
+        output = [{"type": "input_text", "text": json.dumps({"output": output})}]
+    request = {"input": [{"type": "function_call_output", "output": output}]}
+    call = native_process_continuation(request, "fixture", code_mode=code_mode)
+    assert call and call["name"] == ("exec" if code_mode else "write_stdin")
+    if code_mode:
+        assert "tools.write_stdin" in call["input"] and '"session_id": 92241' in call["input"]
+    else:
+        assert json.loads(call["arguments"])["session_id"] == 92241
+    request["input"].append({"type": "function_call_output", "output": "Process exited with code 0"})
+    assert native_process_continuation(request, "complete", code_mode=code_mode) is None
+
+
 def args(path, raw):
     return argparse.Namespace(**dict(capture_file=str(path), capture_sha256=hashlib.sha256(raw).hexdigest(),
         offset=0, page_bytes=256, max_page_output_bytes=2000, format="json", select=["/rows/0"]))
@@ -122,6 +143,10 @@ def test_native_reference_reader_uses_real_pages_before_recording_delivery(tmp_p
     reader.begin("admit current Turn", "source")
     reader.follow(request)
     assert not reader.deliveries and not reader.retained
+    request["input"].append({"type": "function_call_output", "output": "Process running with session ID 92241\n"})
+    call = reader.follow(request)
+    assert call["name"] == "write_stdin" and json.loads(call["arguments"])["session_id"] == 92241
+    assert not reader.deliveries and not reader.retained
     page_count = 0
     while True:
         page = read_context_capture(options)
diff --git a/tests/control_plane/test_saved_context_project.py b/tests/control_plane/test_saved_context_project.py
index be096a0e06..ceee65283a 100644
--- a/tests/control_plane/test_saved_context_project.py
+++ b/tests/control_plane/test_saved_context_project.py
@@ -500,7 +500,9 @@ def test_native_saved_preflight_uses_lifecycle_without_guard_replay(tmp_path, mo
     from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime, read_fixture_todos
     from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
     from loopx.control_plane.quota.turn_envelope import prepare_context_delivery, restore_context_delivery
-    from tests.control_plane.codex_checkpoint_fixture import last_context_page, local_inference_provider
+    from tests.control_plane.codex_checkpoint_fixture import (
+        last_context_page, local_inference_provider, native_process_continuation,
+    )
     from tests.control_plane.test_codex_context_boundary import _app_server, _configure_provider, _native_fixture, _input_text
     from tests.control_plane.test_codex_context_capture import _delivered
     from tests.control_plane.test_quota_settlement_cli import (
@@ -556,6 +558,9 @@ def test_native_saved_preflight_uses_lifecycle_without_guard_replay(tmp_path, mo
     def response(request, number, message):
         nonlocal phase, queue, actual_turn, receipts
         try:
+            if continuation := native_process_continuation(request, f"projection_{number}",
+                    code_mode=host == "app_code_mode"):
+                return continuation
             if queue:
                 command = queue.pop(0)
             elif phase in ("cold", "warm"):

From 6642dd30a318482e43986db183941052aab72b31 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sun, 1 Nov 2026 17:17:19 +0530
Subject: [PATCH 4/5] fix(context): preserve committed recovery when reader
 metadata is unavailable

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 loopx/cli_commands/quota_capture.py           |  8 ++++-
 .../control_plane/codex_checkpoint_fixture.py | 14 +++++---
 .../test_context_delivery_cli.py              |  9 +++--
 .../test_goal_state_context_transport.py      |  8 ++---
 .../test_progressive_native_context.py        | 13 ++++---
 .../test_saved_context_project.py             | 35 ++++++++++---------
 6 files changed, 55 insertions(+), 32 deletions(-)

diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py
index 7507ef7152..6e3c4c60be 100644
--- a/loopx/cli_commands/quota_capture.py
+++ b/loopx/cli_commands/quota_capture.py
@@ -616,7 +616,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(
@@ -650,7 +656,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:
-        return model_context_reference(native_view)
+        return native_view
     return result
 
 
diff --git a/tests/control_plane/codex_checkpoint_fixture.py b/tests/control_plane/codex_checkpoint_fixture.py
index 0055f876fe..6b1139c88a 100644
--- a/tests/control_plane/codex_checkpoint_fixture.py
+++ b/tests/control_plane/codex_checkpoint_fixture.py
@@ -105,6 +105,15 @@ def native_process_continuation(request, identity, *, code_mode=False):
     return None
 
 
+def reference_metadata_oracle(view):
+    """Inspect complete metadata as an oracle, never as a model read receipt."""
+    if view.get("schema_version") == "context_capture_navigation_v1":
+        from loopx.control_plane.turn_driver.decision_capture import read_context_read_reference
+
+        return read_context_read_reference(view["reference_file"])[1]["navigation"]
+    return view
+
+
 class NativeContextReader:
     """Static inference which follows only model-visible native reader cursors.
 
@@ -166,10 +175,7 @@ def follow(self, request):
         if self.pending.get("reference", {}).get("schema_version") == "context_capture_navigation_v1":
             # Resolve oracle metadata only after complete model-visible byte
             # coverage; a reference never seeds the retained-input baseline.
-            from loopx.control_plane.turn_driver.decision_capture import read_context_read_reference
-
-            _, descriptor = read_context_read_reference(first["reference_file"])
-            navigation = descriptor["navigation"]
+            navigation = reference_metadata_oracle(self.pending["reference"])
             assert navigation["sha256"] == first["sha256"]
             first = {**first, **navigation}
         # The original reader owns digest verification. This separate oracle
diff --git a/tests/control_plane/test_context_delivery_cli.py b/tests/control_plane/test_context_delivery_cli.py
index 25b8dd728d..5984a44c3c 100644
--- a/tests/control_plane/test_context_delivery_cli.py
+++ b/tests/control_plane/test_context_delivery_cli.py
@@ -598,10 +598,10 @@ def consume(view):
     assert _spend_run_count(runtime) == 1
 
 
-@pytest.mark.parametrize("native_failure", [None, "projection", "directory"])
+@pytest.mark.parametrize("native_failure", [None, "projection", "directory", "reader_reference"])
 def test_native_projection_never_layers_host_projection(tmp_path, monkeypatch, native_failure):
     from loopx.cli import build_parser
-    from loopx.cli_commands import quota
+    from loopx.cli_commands import quota, quota_capture
     from loopx.control_plane.turn_driver import codex_context
 
     args = build_parser().parse_args(["--format", "json", "quota", "should-run",
@@ -625,10 +625,15 @@ def native(payload, **kwargs):
     def unavailable_directory(**kwargs):
         raise OSError("native directory unavailable")
 
+    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)
     if native_failure == "directory":
         monkeypatch.setattr(codex_context, "codex_context_directory", unavailable_directory)
+    elif native_failure == "reader_reference":
+        monkeypatch.setattr(quota_capture, "write_context_read_reference", unavailable_reference)
     printed = []
     assert quota._emit_quota_result(original, args, context=None, registry_path=tmp_path / "absent.json",
         heartbeat_turn_id=None, usage_quota_started=0, capture_directory=tmp_path,
diff --git a/tests/control_plane/test_goal_state_context_transport.py b/tests/control_plane/test_goal_state_context_transport.py
index 7a734abef4..3bc2c9afd1 100644
--- a/tests/control_plane/test_goal_state_context_transport.py
+++ b/tests/control_plane/test_goal_state_context_transport.py
@@ -16,6 +16,7 @@
 from loopx.control_plane.effect_runtime import MAX_REQUEST_BYTES
 from loopx.control_plane.testing.canary_harness import write_fixture_registry
 from loopx.control_plane.work_items import context_readback
+from tests.control_plane.codex_checkpoint_fixture import reference_metadata_oracle
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
@@ -85,11 +86,10 @@ def guard():
             "--decision-output-root", str(captures)])
     page, raw = json.loads(output.getvalue()), bytearray()
     assert code == 0 and page["ok"], (page.get("reason"), page.get("error"), page.get("error_code"))
-    digest = page["sha256"]
+    digest = reference_metadata_oracle(page)["sha256"]
     if page.get("schema_version") != "context_capture_page_v1":
-        page = read_context_capture(build_parser().parse_args([
-            "--format", "json", "quota", "context-read", "--capture-file", page["path"],
-            "--capture-sha256", digest, "--offset", "0", "--page-bytes", "3072"]))
+        page = read_context_capture(build_parser().parse_args(
+            shlex.split(page["command"])[1:] + ["--page-bytes", "3072"]))
         assert page["ok"], page
     while True:
         assert page["offset"] == len(raw) and page["sha256"] == digest
diff --git a/tests/control_plane/test_progressive_native_context.py b/tests/control_plane/test_progressive_native_context.py
index 2920beffc9..0ed6d2e857 100644
--- a/tests/control_plane/test_progressive_native_context.py
+++ b/tests/control_plane/test_progressive_native_context.py
@@ -10,6 +10,7 @@
 
 from loopx.cli_commands.quota_capture import read_context_capture
 from loopx.control_plane.turn_driver import codex_context as context
+from tests.control_plane.codex_checkpoint_fixture import reference_metadata_oracle
 from test_codex_context_lifecycle import setup, project, deliver, read_pages
 from test_goal_acceptance_cli import acceptance_goal as acceptance_goal
 from test_quota_settlement_cli import _run_cli
@@ -141,13 +142,14 @@ def fail_wire(path, value, **kwargs):
     assert code == 0 and view["ok"] and view["mode"] == "full", view
 
     def check_recovery(view):
+        wire, view = view, reference_metadata_oracle(view)
         journal = context._state(tmp_path)["recovery"]
         assert view["capture_sha256"] == digest
         assert {key: view["recovery"][key] for key in ("path", "sha256")} == {
             key: journal[key] for key in ("path", "sha256")}
-        argv = shlex.split(view["recovery"]["command"])
+        argv = shlex.split(wire["recovery"]["command"])
         if output_limit is not None:
-            assert len((json.dumps(view, ensure_ascii=False, separators=(",", ":")) + "\n").encode()) <= output_limit
+            assert len((json.dumps(wire, ensure_ascii=False, separators=(",", ":")) + "\n").encode()) <= output_limit
             assert int(argv[argv.index("--max-page-output-bytes") + 1]) <= output_limit
             # Discovery and selection remain usable from the complete current
             # recovery, independently of optional shortcut command fields.
@@ -181,11 +183,12 @@ def check_recovery(view):
     recovered = check_recovery(view)
     if failure is None:
         deliver(tmp_path, event, recovered)
-        assert context._state(tmp_path)["candidate"]["capture_file"] == view["recovery"]["path"]
+        assert context._state(tmp_path)["candidate"]["capture_file"] == reference_metadata_oracle(view)["recovery"]["path"]
         code, warm = _run_cli(registry, runtime, *command, cwd=tmp_path)
         assert code == 0 and warm["ok"], warm
-        if "target" in warm:
-            assert warm["target"]["context_epoch"] == view["target"]["context_epoch"]
+        warm_metadata = reference_metadata_oracle(warm)
+        if "target" in warm_metadata:
+            assert warm_metadata["target"]["context_epoch"] == reference_metadata_oracle(view)["target"]["context_epoch"]
         check_recovery(warm)
         context.codex_context_hook(str(tmp_path), {**event, "source": "compact"})
         code, compacted = _run_cli(registry, runtime, *command, cwd=tmp_path)
diff --git a/tests/control_plane/test_saved_context_project.py b/tests/control_plane/test_saved_context_project.py
index ceee65283a..d6ecc1beab 100644
--- a/tests/control_plane/test_saved_context_project.py
+++ b/tests/control_plane/test_saved_context_project.py
@@ -9,15 +9,7 @@
 import pytest
 
 from tests.control_plane.test_quota_settlement_cli import _run_cli
-
-
-def machine_reference_oracle(view):
-    """Inspect complete metadata as an oracle, never as a model read receipt."""
-    if view.get("schema_version") == "context_capture_navigation_v1":
-        from loopx.control_plane.turn_driver.decision_capture import read_context_read_reference
-
-        return read_context_read_reference(view["reference_file"])[1]["navigation"]
-    return view
+from tests.control_plane.codex_checkpoint_fixture import reference_metadata_oracle as machine_reference_oracle
 
 
 @pytest.mark.parametrize("failure", [None, "digest", "goal", "agent", "turn", "symlink", "object"])
@@ -206,10 +198,11 @@ def test_saved_projection_reference_allows_optional_exact_read_and_restart(tmp_p
     assert source.read_bytes() == raw and not registry.exists() and not runtime.exists()
 
 
-@pytest.mark.parametrize("failure", ["wire", "inline_read"])
+@pytest.mark.parametrize("failure", ["wire", "inline_read", "reader_reference", "reader_recovery_reference"])
 def test_saved_projection_wire_failure_keeps_committed_current_recovery(tmp_path, monkeypatch, failure):
-    from loopx.cli import build_parser
-    from loopx.cli_commands.quota_capture import project_saved_context_capture
+    import errno
+
+    from loopx.cli_commands import quota_capture
     from loopx.control_plane.turn_driver import codex_context, decision_capture
 
     path, root, context = tmp_path / "original.json", tmp_path / "cache", tmp_path / "native"
@@ -232,14 +225,24 @@ def fail_optional_wire(path, payload, **kwargs):
 
     monkeypatch.setattr(decision_capture, "write_decision_capture", fail_optional_wire)
     if failure == "inline_read":
-        from loopx.cli_commands import quota_capture
         monkeypatch.setattr(quota_capture, "read_context_capture", lambda args, **kwargs: {"ok": False})
+    elif failure.startswith("reader_"):
+        publish = quota_capture.write_context_read_reference
+        calls = []
+
+        def fail_reference(*args, **kwargs):
+            calls.append(args)
+            if len(calls) == (2 if failure == "reader_recovery_reference" else 1):
+                raise OSError(errno.ENOSPC, "synthetic reference publication failure")
+            return publish(*args, **kwargs)
+
+        monkeypatch.setattr(quota_capture, "write_context_read_reference", fail_reference)
     digest = hashlib.sha256(raw).hexdigest()
-    args = build_parser().parse_args(["--format", "json", "quota", "context-project",
+    code, wire = _run_cli(tmp_path / "absent.json", tmp_path / "absent-runtime", "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", "on",
-        "--codex-app", "--codex-context-dir", str(context), "--decision-output-root", str(root)])
-    wire = project_saved_context_capture(args)
+        "--codex-app", "--codex-context-dir", str(context), "--decision-output-root", str(root), cwd=tmp_path)
+    assert code == 0, wire  # Actual CLI must return JSON, never an IO traceback.
     result = machine_reference_oracle(wire)
     assert result["ok"] and result["mode"] == "full"
     recovered_path, recovered_digest = codex_context.read_codex_context_reference(str(context),

From 78f02bc3cc644baa91d9109938822ade2b08ce75 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sun, 1 Nov 2026 17:40:38 +0530
Subject: [PATCH 5/5] fix: preserve verified context reads when metadata is
 unwritable

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 loopx/cli_commands/quota_capture.py           | 11 ++--
 .../test_context_read_reference.py            | 52 +++++++++++++++----
 2 files changed, 50 insertions(+), 13 deletions(-)

diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py
index 6e3c4c60be..a635ec1122 100644
--- a/loopx/cli_commands/quota_capture.py
+++ b/loopx/cli_commands/quota_capture.py
@@ -250,9 +250,14 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
             selection_digest = expected
             selectors = [item["pointer"] for item in views]
             if reference_file is not None and pinned_selection is None:
-                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})
+                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,
diff --git a/tests/control_plane/test_context_read_reference.py b/tests/control_plane/test_context_read_reference.py
index b699e6029d..bc19cb2b60 100644
--- a/tests/control_plane/test_context_read_reference.py
+++ b/tests/control_plane/test_context_read_reference.py
@@ -1,5 +1,4 @@
 """Reader references hide repeated digests without weakening byte verification."""
-from argparse import Namespace
 import hashlib
 import json
 from pathlib import Path
@@ -15,15 +14,12 @@
 
 
 def read(command, **overrides):
-    words = shlex.split(command)
-    args = dict(format="json", offset=0, page_bytes=64000, max_page_output_bytes=64000)
-    for name in ("reference-file", "capture-file", "capture-sha256", "offset", "page-bytes", "max-page-output-bytes"):
-        if "--" + name in words:
-            value = words[words.index("--" + name) + 1]
-            args[name.replace("-", "_")] = int(value) if name in {"offset", "page-bytes", "max-page-output-bytes"} else value
-    args["initial_read"] = "--initial-read" in words
-    args.update(overrides)
-    return read_context_capture(Namespace(**args))
+    from loopx.cli import build_parser
+
+    args = build_parser().parse_args(shlex.split(command)[1:])
+    for key, value in overrides.items():
+        setattr(args, key, value)
+    return read_context_capture(args)
 
 
 def fixture(tmp_path):
@@ -95,6 +91,42 @@ def changed(operation, **kwargs):
     assert "content" not in rejected
 
 
+@pytest.mark.parametrize("selection", ["fields", "index", "initial"])
+def test_selected_read_survives_unwritable_reference_metadata(tmp_path, monkeypatch, selection):
+    import errno
+    from loopx.cli_commands import quota_capture
+
+    payload, machine, model = fixture(tmp_path)
+    before = Path(machine["path"]).read_bytes()
+
+    def unavailable(*args, **kwargs):
+        raise OSError(errno.ENOSPC, "synthetic metadata unavailable")
+
+    monkeypatch.setattr(quota_capture, "write_context_read_reference", unavailable)
+    options = {"fields": {"select": ["/selected_todo", "/criterion"]},
+               "index": {"list_fields": True}, "initial": {"initial_read": True}}[selection]
+    page = read(model["command"], max_page_output_bytes=3500, page_bytes=3000, **options)
+    content = []
+    while True:
+        assert page["ok"] and page["schema_version"] == "context_capture_page_v1"
+        assert page["sha256"] == machine["sha256"] and page["selection_sha256"]
+        assert capture_page_range(page, raw=before, digest=machine["sha256"]) is None
+        content.append(page["content"])
+        if page["eof"]:
+            break
+        assert "--capture-sha256" in page["next_command"] and "--selection-sha256" in page["next_command"]
+        page = read(page["next_command"])
+    values = json.loads("".join(content))["selections"]
+    if selection == "fields":
+        assert {item["pointer"]: item["value"] for item in values} == {
+            "/selected_todo": payload["selected_todo"], "/criterion": payload["criterion"]}
+    elif selection == "index":
+        assert {item["pointer"] for item in values[0]["fields"]} == {"/" + key for key in payload}
+    else:
+        assert next(item["value"] for item in values if item["pointer"] == "/goal_boundary") == payload["goal_boundary"]
+    assert Path(machine["path"]).read_bytes() == before
+
+
 @pytest.mark.parametrize("damage", ["capture", "selection", "symlink", "permissions"])
 def test_reference_damage_fails_closed(tmp_path, damage):
     _, machine, model = fixture(tmp_path)