fix(context): keep bounded native references readable

docs/reference/required-work-context.md

@@ -456,6 +456,12 @@ ceiling); if metadata cannot fit, recovery retains the current capture. Reader
 pages remain independently bounded. A complete raw tool log is not proof that
 the Host retained the same bytes in model history.
 
+File references carry the original path, digest and budgeted reader command;
+the outer view supplies navigation. The first bounded full-file page supplies
+the complete paging and truncation-recovery instructions, including when read
+directly by capture/hash in a fresh shell. Selected pages keep their scoped
+read instructions. Historical unbounded page responses retain their wire shape.
+
 The native codec's `captured_file` basis verifies the previous full file and
 compares exact versions, including unchanged control fields. It does not depend
 on a full read, page EOF, a model ACK or a retained-input claim. Goal, Agent,
@@ -748,6 +754,10 @@ SSH Goal 使用 `--runtime-profile codex_app_ssh_goal` 生成 context-config 和
 元数据仍放不下时保留当前捕获恢复入口。读取页另受输出预算约束。
 原始工具日志完整，不证明宿主保留给模型的字节也完整。
 
+文件引用保留原始路径、摘要和遵循预算的读取命令；外层视图提供导航。首个有界全文
+分页返回完整的分页与截断恢复说明，直接在新 shell 中按 capture/hash 读取也一样。
+选读页仍提供其作用域内的读取说明；历史无界分页响应保留原有 wire 结构。
+
 原生 `captured_file` 基线来自校验后的前一完整文件，可差分复用所有未变字段，包括
 控制字段；不依赖通读、EOF、模型 ACK 或保留输入声明。Goal、Agent、会话、surface、
 投影版本、配置 generation 和上下文 epoch 仍隔离；压缩后下一个新 Turn 建立完整

loopx/cli_commands/quota_capture.py

@@ -59,7 +59,11 @@ def page(text: str) -> dict[str, Any]:
             result["read_instruction"] = ("This is selected detail from the original capture, not a complete context read. "
                 "Offsets and EOF describe only this selected view. Follow next_command to finish it; "
                 "keep both digests. Detail delivery does not establish a full-context baseline or grant authority.")
-        elif instruction:
+        elif instruction or (offset == 0 and output_limit is not None):
+            # A digest-bound native reference may be read in a fresh shell
+            # without its session lookup. Deliver paging obligations with the
+            # 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 cursor < len(raw):
             result["next_command"] = context_capture_read_command(path, digest, offset=cursor, page_bytes=next_page_bytes or size,
@@ -203,6 +207,18 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
             "recovery": "Restore the original digest-bound capture; no guard, admission or spend was performed."}
 
 
+def _capture_read_reference(value: dict[str, Any], *, page_bytes: int,
+                            output_bytes: int) -> dict[str, str]:
+    """Keep file identity and the budgeted reader at the transport boundary.
+
+    Navigation lives on the outer view; complete paging obligations accompany
+    the actual reader response. Neither belongs in every nested file reference.
+    """
+    return {"path": value["path"], "sha256": value["sha256"],
+        "command": context_capture_read_command(Path(value["path"]), value["sha256"],
+            page_bytes=page_bytes, max_page_output_bytes=output_bytes)}
+
+
 def native_context_capture_view(projected: dict[str, Any], directory: Path, *,
                                 context_directory: str | None = None,
                                 metadata: dict[str, Any] | None = None,
@@ -232,24 +248,20 @@ def native_context_capture_view(projected: dict[str, Any], directory: Path, *,
     if max_page_output_bytes is not None:
         output_bytes = min(output_bytes, max_page_output_bytes)
         page_bytes = min(page_bytes, output_bytes)
-    reference["command"] = context_capture_read_command(Path(reference["path"]), reference["sha256"],
-        page_bytes=page_bytes, max_page_output_bytes=output_bytes)
+    reference = _capture_read_reference(reference, page_bytes=page_bytes, output_bytes=output_bytes)
     result = {"ok": True, **(metadata or {}), "mode": mode, **reference,
         **({"target": target} if target else {})}
-    if max_page_output_bytes is not None and isinstance(result.get("recovery"), dict):
-        recovery = result["recovery"]
-        result["recovery"] = {**recovery, "command": context_capture_read_command(
-            Path(recovery["path"]), recovery["sha256"], page_bytes=page_bytes,
-            max_page_output_bytes=output_bytes)}
+    if isinstance(result.get("recovery"), dict):
+        result["recovery"] = _capture_read_reference(result["recovery"],
+            page_bytes=page_bytes, output_bytes=output_bytes)
     # Persist first. A normal guard returns a small reference, never an eager
     # page that can be cut by an unrelated Host/history budget. The optional
     # delta remains readable in full; navigation always addresses current facts.
     current = projected.get("recovery") if projected.get("schema_version") == "context_delivery_v1" else reference
     current = current or reference
     current_path = Path(current["path"])
-    result.setdefault("recovery", {"path": str(current_path), "sha256": current["sha256"],
-        "command": context_capture_read_command(current_path, current["sha256"],
-            page_bytes=page_bytes, max_page_output_bytes=output_bytes)})
+    result.setdefault("recovery", _capture_read_reference(current,
+        page_bytes=page_bytes, output_bytes=output_bytes))
     current_raw = current_path.read_bytes()
     current_document = json.loads(current_raw)
     if projected.get("continuity_basis") == "captured_file":
@@ -340,14 +352,8 @@ def project_saved_context_capture(args: argparse.Namespace) -> dict[str, Any]:
             output_bytes = min(output_bytes, output_limit)
             page_bytes = min(page_bytes, output_bytes)
 
-        def recovery_reference(value: dict[str, Any]) -> dict[str, str]:
-            # Preserve the caller's read budget on success and failure without
-            # duplicating instructions already carried by the outer view.
-            return {"path": value["path"], "sha256": value["sha256"],
-                "command": context_capture_read_command(Path(value["path"]), value["sha256"],
-                    page_bytes=page_bytes, max_page_output_bytes=output_bytes)}
-
-        full_reference = recovery_reference(full_reference)
+        full_reference = _capture_read_reference(full_reference,
+            page_bytes=page_bytes, output_bytes=output_bytes)
         reference = dict(full_reference)
         directory = codex_context_directory(directory=args.codex_context_dir,
             root=args.codex_context_root, create=True)
@@ -378,7 +384,8 @@ def recovery_reference(value: dict[str, Any]) -> dict[str, str]:
             # Native publication owns the complete required view. Keep its
             # committed recovery even if optional wire/inline delivery fails;
             # the input capture digest below remains the original identity.
-            full_reference = recovery_reference(projected["recovery"])
+            full_reference = _capture_read_reference(projected["recovery"],
+                page_bytes=page_bytes, output_bytes=output_bytes)
             reference = dict(full_reference)
         native_view = native_context_capture_view(projected, cache, context_directory=directory,
             metadata={"recovery": full_reference, "capture_sha256": args.capture_sha256,

tests/control_plane/test_selection_replan_reentry.py

@@ -461,12 +461,13 @@ def concurrent_receipt(*args, **kwargs):
     assert _spend_run_count(runtime) == 0
 
 
-def test_native_first_selection_publishes_only_final_bound_capture(tmp_path, monkeypatch):
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_native_first_selection_publishes_only_final_bound_capture(tmp_path, monkeypatch, provider):
     from loopx.control_plane.turn_driver.codex_context import codex_context_hook
 
     project, runtime, registry = _write_fixture(tmp_path)
     _configure_selected_todo_replan_fixture(project, registry)
-    _initialize_fixture_authority(project, runtime, provider="sqlite")
+    _initialize_fixture_authority(project, runtime, provider=provider)
     turn = "turn-native-first-selection"
     capture = tmp_path / "capture"
     native = tmp_path / "native"
@@ -481,6 +482,7 @@ def test_native_first_selection_publishes_only_final_bound_capture(tmp_path, mon
         "--codex-context-dir", str(native), "--decision-output-dir", str(capture),
         "--max-page-output-bytes", "4096")
     assert rc == 0 and presented["ok"], presented
+    assert len(json.dumps(presented, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= 4096
     original = json.loads((capture / "decision.json").read_text())
     assert original["ok"] and original["effective_action"] == "autonomous_replan_required"
     assert original["heartbeat_receipt"]["status"] == "committed"
@@ -489,3 +491,22 @@ def test_native_first_selection_publishes_only_final_bound_capture(tmp_path, mon
     assert not original["interaction_contract"]["cli_channel"].get("selection_required")
     assert _heartbeat_receipt_count(runtime, turn) == 1
     assert _spend_run_count(runtime) == 0
+    # The bounded reference reaches the original receipt without another
+    # admission. Reading a selected section does not certify the whole input.
+    from loopx.cli import build_parser
+    from loopx.cli_commands.quota_capture import read_context_capture
+    from loopx.control_plane.turn_driver.decision_capture import CONTEXT_CAPTURE_PAGE_GUIDANCE
+
+    reader = shlex.split(presented["recovery"]["command"])[1:]
+    first = read_context_capture(build_parser().parse_args(reader))
+    assert first["ok"] and first["offset"] == 0
+    assert first["read_instruction"] == CONTEXT_CAPTURE_PAGE_GUIDANCE
+    assert len(json.dumps(first, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= 4096
+    args = build_parser().parse_args(reader
+        + ["--select", "/heartbeat_receipt"])
+    read = read_context_capture(args)
+    assert read["ok"] and read["eof"]
+    assert read["sha256"] == presented["recovery"]["sha256"]
+    assert json.loads(read["content"])["selections"][0]["value"] == original["heartbeat_receipt"]
+    assert _heartbeat_receipt_count(runtime, turn) == 1
+    assert _spend_run_count(runtime) == 0