fix(context): deliver saved files for autonomous progressive reading

docs/architecture/rfcs/controller-executor-unified-contract-v0.md

@@ -881,6 +881,15 @@ are stable design references, not claims that a canonical Todo is complete.
 | **C8 — Product configuration and operational readback; C3/C6** | Existing App/settings/Chat projections: one effective configuration owner, role-aware activity, failure/recovery, Stop and artifact return | Packaged managed and supported attached heartbeat journeys, including unavailable host and late results; installed qualification separate; no new packet-format toggle |
 | **C9 — Retire old paths and qualify adoption; C1–C8** | Contract/runtime maintainers: remove superseded producers, format switches, duplicated prompts and tests; update protocols, guides and RFC delivery facts | Caller/field/decoder retirement inventory, historical replay and rollback evidence; App + benchmark acceptance and matched benefit before any default promotion |
 
+The existing native context entry now supports a bounded C2 precursor: complete
+captures remain on disk, the normal response exposes file navigation, and exact
+file deltas do not require model retention. After compaction the next work Turn
+establishes a new full saved baseline, without an immediate full reread.
+Required facts still precede dependent decisions; admission and settlement are
+unchanged. This does not qualify C1's unified producer/consumer cutover or the
+matched effectiveness/full-cost result in §9. Rollback uses the existing
+context-delivery off path; staged material composition remains separately owned.
+
 C1's inventory is part of its usable replacement, not a standalone architecture
 framework. C3/C4/C5 can proceed independently once C1's contract is stable;
 C7 can prepare measurements before C6 but cannot claim two-session outcomes.

docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md

@@ -663,6 +663,13 @@ backoff 继续归当前 scheduler owner。本文不引入模型空轮询或新
 | **C8 — 产品配置与运行 readback；C3/C6** | 既有 App/settings/Chat 投影：同一 effective configuration owner，按角色展示 activity、失败/恢复、Stop 和产物返回 | 打包 managed 及受支持 attached heartbeat 流程，包括 host 不可用和结果迟到；安装资格另验；不新增 packet 格式开关 |
 | **C9 — 旧路径退役与采用资格；C1–C8** | Contract/runtime 维护者：删除被替代 producer、格式开关、重复提示/测试，更新 protocol、guide 和 RFC 交付事实 | 调用方/字段/decoder 退役清单、历史 replay 与回退证据；App + benchmark 验收及可比收益成立后再决定默认提升 |
 
+现有原生上下文入口增加一个有界 C2 前置改进：完整捕获保存在磁盘，常规响应提供
+文件导航，精确文件差分不依赖模型保留；压缩后下一个新 Turn 建立新的完整文件基线，不要求立即
+全量重读。必要事实仍须在
+依赖性决策前理解；准入、结算不变。这不代表 C1 统一 producer/consumer 切换或
+第 9 节可比效果/完整成本验收完成。沿既有 context-delivery off 路径回滚；材料分阶段
+组合仍由原 owner 独立收敛。
+
 C1 盘点是可用替换的一部分，不单独建设架构框架。C1 契约稳定后，C3/C4/C5 可以独立
 推进；C7 可以提前准备测量，但不能提前宣称双会话成果。实现 PR 列明替换的调用方、剩余
 缺口、实际测试与回退。完整语言迁移或无关 manager 重写不是前置条件；新抽象能编译

docs/development/README.md

@@ -246,6 +246,13 @@ cross-work dependencies and contrary or unresolved evidence require expansion
 before a dependent decision. Reuse an existing typed scope owner instead of adding
 keyword classifiers or another relevance engine.
 
+Keep file availability separate from model retention. Saving a complete capture
+permits exact file comparison, not a claim that the model read it. In native
+context delivery, a new compaction epoch starts with a full saved baseline on
+the next work Turn; ordinary resume and selective reads preserve the epoch.
+Recovery continues the original Turn and reads missing necessary facts, without
+replaying admission or draining an entire packet to enable later deltas.
+
 Separate **acceptance criteria** from **acceptance evidence**. Criteria and current
 unmet/unknown status guide execution; historical reviewer notes, full provenance,
 logs and detailed evidence support a specific verification, dispute, replan or

loopx/cli_commands/quota_capture.py

@@ -18,7 +18,7 @@
 from ..control_plane.quota.error_codes import QuotaCommandValidationError, QuotaDecisionTransportError
 from ..control_plane.turn_driver.decision_capture import (
     validate_handoff_socket, CONTEXT_CAPTURE_READ_GUIDANCE, CONTEXT_CAPTURE_PAGE_GUIDANCE, context_capture_read_command,
-    NATIVE_CONTEXT_PAGE_BYTES, NATIVE_CONTEXT_PAGE_OUTPUT_BYTES,
+    NATIVE_CONTEXT_PAGE_BYTES, NATIVE_CONTEXT_PAGE_OUTPUT_BYTES, CONTEXT_CAPTURE_NAVIGATION_GUIDANCE,
     native_context_page_budget,
     read_context_capture_file as _read_capture,
 )
@@ -39,7 +39,8 @@ def _capture_page(path: Path, raw: bytes, digest: str, *, offset: int, size: int
                   next_page_bytes: int | None = None,
                   next_page_output_bytes: int | None = None,
                   select: list[str] | None = None,
-                  selection_sha256: str | None = None) -> dict[str, Any]:
+                  selection_sha256: str | None = None,
+                  list_fields: bool = False) -> dict[str, Any]:
     end = min(offset + size, len(raw))
     while end < len(raw) and raw[end] & 0xC0 == 0x80:
         end -= 1
@@ -52,6 +53,8 @@ def page(text: str) -> dict[str, Any]:
             "total_bytes": len(raw), "eof": cursor == len(raw), "content": text}
         if select is not None:
             result.update(selection_sha256=selection_sha256, selected_pointers=select)
+            if list_fields:
+                result["list_fields"] = True
             result["read_instruction"] = ("This is selected detail from the original capture, not a complete context read. "
                 "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.")
@@ -60,7 +63,7 @@ def page(text: str) -> dict[str, Any]:
         if cursor < len(raw):
             result["next_command"] = context_capture_read_command(path, digest, offset=cursor, page_bytes=next_page_bytes or size,
                 max_page_output_bytes=next_page_output_bytes or output_limit,
-                select=select, selection_sha256=selection_sha256)
+                select=select, selection_sha256=selection_sha256, list_fields=list_fields)
         return result
 
     def fits(result: dict[str, Any]) -> bool:
@@ -134,7 +137,10 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
             raise ValueError("context-read requires a nonnegative byte offset and page-bytes between 256 and 8192; "
                 "larger pages require a sufficient max-page-output-bytes within the capture limit")
         path, raw, payload = _read_capture(value, digest)
+        list_fields = getattr(args, "list_fields", False)
         selectors = getattr(args, "select", None)
+        if list_fields and selectors is None:
+            selectors = [""]
         selection_digest = getattr(args, "selection_sha256", None)
         if selectors is None and selection_digest is not None:
             raise ValueError("selection-sha256 requires --select")
@@ -145,14 +151,30 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
                 raise ValueError("selected continuation requires selection-sha256; follow next_command")
             from ..control_plane.quota.turn_envelope import _turn_envelope_result
 
-            plan = _turn_envelope_result("context_select", context=payload, select=selectors)
+            plan = _turn_envelope_result("context_select", context=payload, select=selectors, list_fields=list_fields)
             views: list[dict[str, Any]] = []
             for item in plan["selections"]:
                 selected: Any = payload
                 for key in item["path"]:
                     selected = selected[key]
-                views.append({"pointer": item["pointer"], "value": selected})
-            raw = json.dumps({"schema_version": "context_capture_selection_v1", "selections": views},
+                if list_fields:
+                    fields = []
+                    for child in item["children"]:
+                        value = payload
+                        for key in child["path"]:
+                            value = value[key]
+                        field = {"pointer": child["pointer"], "type":
+                            "object" if isinstance(value, dict) else "array" if isinstance(value, list)
+                            else "null" if value is None else "boolean" if isinstance(value, bool)
+                            else "string" if isinstance(value, str) else "number",
+                            "bytes": len(json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode("utf-8"))}
+                        if isinstance(value, (dict, list)):
+                            field["count"] = len(value)
+                        fields.append(field)
+                    views.append({"pointer": item["pointer"], "fields": fields})
+                else:
+                    views.append({"pointer": item["pointer"], "value": selected})
+            raw = json.dumps({"schema_version": "context_capture_index_v1" if list_fields else "context_capture_selection_v1", "selections": views},
                 ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") + b"\n"
             expected = hashlib.sha256(raw).hexdigest()
             if selection_digest is not None and selection_digest != expected:
@@ -164,7 +186,7 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
         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)
+            next_page_output_bytes=next_page_output_bytes, select=selectors, selection_sha256=selection_digest, list_fields=list_fields)
     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.
@@ -178,10 +200,10 @@ def native_context_capture_view(projected: dict[str, Any], directory: Path, *,
                                 context_directory: str | None = None,
                                 metadata: dict[str, Any] | None = None,
                                 max_page_output_bytes: int | None = None) -> dict[str, Any]:
-    """One paged presentation for ordinary and already-captured native calls.
+    """Publish file references for an already-decided native capture.
 
-    The guard/projector already ran. Only serialize its wire and prefetch the
-    existing digest-checked reader; never collect or admit another decision.
+    Persist the complete wire and expose navigation without eagerly importing
+    content into model history. Reading never collects or admits a decision.
     """
     from ..control_plane.turn_driver.decision_capture import write_decision_capture
 
@@ -208,34 +230,51 @@ def native_context_capture_view(projected: dict[str, Any], directory: Path, *,
         result["recovery"] = {**recovery, "command": context_capture_read_command(
             Path(recovery["path"]), recovery["sha256"], page_bytes=page_bytes,
             max_page_output_bytes=output_bytes)}
-    # The codec puts native continuity semantics before its change tree. Keep
-    # this shared transport frame neutral instead of copying that instruction
-    # or replacing an assumption with a claim of confirmed final input.
-    presentation = ("This is the full current capture; after complete reading, continue from it."
-        if mode == "full" else "Read the current continuity basis and instruction before its change tree.")
-    instruction = (
-        "Presentation of the original decision only; authority and freshness are unchanged. "
-        + presentation + " ")
-    result["instruction"] = (instruction + "This result includes page zero; consume it before next_command. "
-        "command restarts at zero if the inline page is incomplete or unsupported. " + CONTEXT_CAPTURE_PAGE_GUIDANCE)
-    # Bound the complete stdout, including caller-owned recovery and activation
-    # fields. Capping only content leaves the guard itself vulnerable to native
-    # truncation before the consumer can discover its continuation command.
-    page = read_context_capture(argparse.Namespace(capture_file=reference["path"],
-        capture_sha256=reference["sha256"], offset=0, page_bytes=page_bytes,
-        format="json", max_page_output_bytes=output_bytes),
-        presentation=result, next_page_bytes=page_bytes, next_page_output_bytes=output_bytes)
-    if page["ok"]:
-        result = page
-        from ..control_plane.turn_driver.codex_context import register_native_context_view
-
-        register_native_context_view(context_directory,
-            {**result, "ancestry": projected.get("ancestry", 0)})
-    else:
-        result["instruction"] = instruction + CONTEXT_CAPTURE_PAGE_GUIDANCE
-        if len(json.dumps(result, ensure_ascii=False, separators=(",", ":")).encode()) + 1 > output_bytes:
-            raise QuotaDecisionTransportError("Native context metadata exceeds the output limit; "
-                "read the saved original decision and receipt without rerunning admission.")
+    # 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)})
+    current_raw = current_path.read_bytes()
+    current_document = json.loads(current_raw)
+    if projected.get("continuity_basis") == "captured_file":
+        result["continuity_basis"] = "captured_file"
+        result["baseline_reference"] = dict(projected["baseline_reference"])
+    result["total_bytes"] = Path(reference["path"]).stat().st_size
+    result["index_command"] = context_capture_read_command(current_path, current["sha256"],
+        page_bytes=min(page_bytes, 24000), max_page_output_bytes=min(output_bytes, 24000), list_fields=True)
+    result["select_command"] = context_capture_read_command(current_path, current["sha256"],
+        page_bytes=page_bytes, max_page_output_bytes=output_bytes) + " --select '<pointer>'"
+    # A complete scalar preview helps orientation without inventing a second
+    # decision owner. Oversized or structured values remain discoverable in the
+    # index; no prose is truncated or treated as an authorization summary.
+    result["overview"] = {key: value for key, value in current_document.items()
+        if value is None or isinstance(value, (str, bool, int, float))
+        and len(json.dumps(value, ensure_ascii=False).encode("utf-8")) <= 240}
+    result["instruction"] = CONTEXT_CAPTURE_NAVIGATION_GUIDANCE
+    if mode != "full":
+        result["instruction"] += (" command reads the delta; baseline_reference identifies its exact saved base when file-backed. "
+            "index_command/select_command read current facts directly without that baseline.")
+    # The reference must fit even under an explicit small caller ceiling.
+    def fits():
+        return len(json.dumps(result, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= min(output_bytes, 7900)
+    if not fits():
+        result.pop("overview")
+    if not fits():
+        result.pop("select_command")  # --select also works on the current recovery command.
+    if not fits() and "baseline_reference" in result:
+        result["baseline_reference"].pop("command", None)  # Exact path/hash remain usable.
+    if not fits():
+        raise QuotaDecisionTransportError("Native context reference exceeds the output limit; "
+            "read the saved original decision and receipt without rerunning admission.")
+    from ..control_plane.turn_driver.codex_context import register_native_context_view
+
+    register_native_context_view(context_directory,
+        {**result, "ancestry": projected.get("ancestry", 0)})
     return result
 
 
@@ -354,7 +393,7 @@ def recovery_reference(value: dict[str, Any]) -> dict[str, str]:
         "capture_sha256": args.capture_sha256,
         **({"context_delivery": activation} if activation["enabled"] else {}),
         "instruction": "Presentation of the original decision only; authority and freshness are unchanged. "
-            + CONTEXT_CAPTURE_READ_GUIDANCE}
+            + (CONTEXT_CAPTURE_NAVIGATION_GUIDANCE if activation["enabled"] else CONTEXT_CAPTURE_READ_GUIDANCE)}
     if native_view is not None:
         result.update(native_view)
     return result
@@ -557,6 +596,9 @@ def handoff_decision(directory: Path | None, socket_path: str | None,
 def handle_native_context_command(args: argparse.Namespace, *, print_payload: Callable[..., None],
                                   registry_path: Path | None = None, runtime_root_arg: str | None = None) -> int | None:
     """Dispatch native presentation/lifecycle IO before any quota authority read."""
+    if getattr(args, "list_fields", False) and args.quota_command != "context-read":
+        print_payload({"ok": False, "reason": "--list is only supported by context-read"}, args.format, str)
+        return 1
     has_output_limit = getattr(args, "max_page_output_bytes", None) is not None
     budget_scope_error = ("max-page-output-bytes is only supported by JSON context-read "
         "or enabled native should-run/context-project")
@@ -619,7 +661,7 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca
                        "Omit per-command activation and location flags to retain that inheritance. " if machine_runtime else
                        "Pass --context-delivery on with the same declared host mode, context location and a decision-output-root to heartbeat-prompt/should-run. ")
                     +
-                    "Normal completed full reads establish an explicit continuity assumption; no input diagnostics or provider change is required. "
+                    "Verified saved files support progressive reads and exact deltas without a full model read; compaction starts a new full baseline on the next work Turn. "
                     "The generated output profile uses 64000-byte continuations and a 20000-token native history limit; "
                     "the same page profile applies without the shell override. Explicit smaller reader budgets remain available for truncated-range recovery. "
                     "Set --context-delivery off and remove the generated context configuration, including its output overrides, to restore ordinary output. "