huangruiteng · Nov 6, 2026 · Nov 5, 2026 · Nov 5, 2026 · Nov 5, 2026 · Nov 5, 2026
@@ -686,6 +686,15 @@ 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.
 
+In a native progressive session, the initial read can reference a previously
+delivered preferences block or acceptance outcome. The first page lists those
+pointers in `previously_delivered_fields`. Their requirements still apply.
+Use `recovery.command --select <pointer>` for the current body when needed.
+Reuse requires complete observed tool delivery in the same task, session and
+epoch, and a matching digest of the current visible field. A revised field
+appears in full. Losing the observation restores the ordinary initial read.
+A saved delta on its own provides no evidence of model retention.
+
 Each progressive `context_details.source_path` names this Turn's complete immutable
 `projection.json`, including when a warm wire omits unchanged material. The current
 `unread_context_index` preserves unobserved or changed supporting pointers and

@@ -206,6 +206,26 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
                     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")
+            if observation := reference.get("observed_read"):
+                from ..control_plane.turn_driver.native_first_read import observation_current
+
+                if not observation_current(observation, path, digest):
+                    if offset:
+                        return {"ok": False, "reason": "Observed first-read scope changed during paging.",
+                            "recovery_command": context_capture_read_command(path, digest, initial_read=True,
+                                page_bytes=size, max_page_output_bytes=output_limit, reader_argv=reader_argv)}
+                    # Restart in the ordinary complete view of this same capture.
+                    # Subsequent pages must pin that view even if the journal changes.
+                    pinned_selection, initial_read = None, True
+                    try:
+                        reference_file = write_context_read_reference(path, digest)
+                    except OSError:
+                        reference_file = None
+                elif offset == 0:
+                    presentation = {**(presentation or {}), "previously_delivered_fields": list(observation["fields"]),
+                        "field_recovery": "These unchanged fields still apply. If needed or uncertain, append --select "
+                            "and the field pointer to recovery.command for its complete current body."}
+        if pinned_selection is not None:
             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"]
@@ -531,7 +551,7 @@ def pin_initial_selection(plan: dict[str, Any]) -> None:
             # metadata is optional and never a read/authority receipt.
             pass
 
-    if initial_read_plan is not None:
+    if progressive_enabled and initial_read_plan is not None:
         try:
             initial, plan = initial_read_plan["initial"], initial_read_plan["plan"]
             if initial:
@@ -592,8 +612,14 @@ def view(kind: str, ref: dict[str, Any], body: bytes) -> dict[str, Any]:
         write_decision_capture(wire, projected)
     from ..control_plane.turn_driver.codex_context import register_native_context_view
 
-    register_native_context_view(context_directory,
+    observed = register_native_context_view(context_directory,
         {**result, "ancestry": ancestry}, hook_session=hook_session)
+    if (progressive_enabled and observed and "first_read_command" in result
+            and observed["pointer_bytes"] <= output_bytes // 4):
+        first = Path(observed["path"])
+        result["first_read_reference"] = str(first)
+        result["first_read_command"] = context_capture_read_command(Path(current["path"]), current["sha256"],
+            reference_file=first, page_bytes=page_bytes, max_page_output_bytes=output_bytes, reader_argv=reader_argv)
     return result
 
 

@@ -68,6 +68,9 @@ export function updateNativeContextLifecycle(request: JsonObject): JsonObject {
       delete state.candidate;
       delete state.pending;
       delete state.active_call;
+      delete state.consumed_pointers;
+      delete state.confirmed_read_sections;
+      delete state.section_reads;
       return state;
     }
     invalidate();

@@ -10,24 +10,86 @@ import {FULL_OUTCOME_READ_GUIDANCE, PROGRESSIVE_OUTCOME_READ_GUIDANCE} from "../
 import {quotaSummaryReferences, contextSourceIdentity, contextProjectionChanges,
   type ProjectionSourceValues} from "./diagnostic_projection.ts";
 import {encodeContextDelivery, fileNavigationPaths} from "./context_delivery.ts";
+import {capturedContextSelection} from "./context_selection.ts";
 import {createHash} from "node:crypto";
 
-import {capturedContextSelection} from "./context_selection.ts";
+
 
 type Path = (string | number)[];
+const MIN_REUSABLE_BODY_BYTES = 1024;
 const pointer = (path: Path): string => path.length ? "/" + path.map(key =>
   String(key).replaceAll("~", "~0").replaceAll("/", "~1")).join("/") : "";
 const at = (root: unknown, path: Path): unknown => path.reduce<unknown>((row, key) =>
   Array.isArray(row) ? row[Number(key)] : jsonObject(row)?.[key], root);
 const known = (row: JsonObject, fields: string[]): boolean => Object.keys(row).every(key => fields.includes(key));
 
+/** One registry for bodies eligible for observed reuse, regardless of whether
+ * the reader delivered the body itself or packaged it in a complete parent. */
+function reusableFirstReadFields(context: JsonObject): {path: Path; freshKeys: string[]}[] {
+  const base: Path = ["interaction_contract", "agent_channel", "work_context", "sources"];
+  const sources = at(context, base);
+  if (!Array.isArray(sources)) return [];
+  return sources.flatMap((value, i) => {
+    const source = jsonObject(value), content = jsonObject(source?.content);
+    if (!content) return [];
+    const path = [...base, i, "content"];
+    if (source?.kind === "agent_preferences")
+      return [{path, freshKeys: ["ok", "revision", "instruction"]}];
+    if (source?.source !== "goal_acceptance") return [];
+    const outcomesPath = [...path, "goal_acceptance_contract", "outcome_requirements", "outcomes"];
+    const outcomes = at(context, outcomesPath);
+    return Array.isArray(outcomes) ? outcomes.flatMap((outcome, index) =>
+      typeof jsonObject(outcome)?.id === "string"
+        ? [{path: [...outcomesPath, index], freshKeys: ["id", "revision"]}] : []) : [];
+  });
+}
+
 /** Digests stay in the private observed-read journal, never repeated to the model.
  * The existing exact-source identity preserves non-JS numeric spellings. */
 export function contextReadSectionDigests(request: JsonObject, paths: Path[]): JsonObject {
   const context = requireJsonObject(request.context, "read section source");
   const identity = contextSourceIdentity(request, request.source_number_tokens);
-  return Object.fromEntries(paths.map(path => [pointer(path),
-    createHash("sha256").update(identity(at(context, path), ["context", ...path])).digest("hex")]));
+  const selected = new Set(paths.map(pointer));
+  const covered = new Map(paths.map(path => [pointer(path), path]));
+  for (const {path} of reusableFirstReadFields(context))
+    if (paths.some(parent => pathStartsWith(path, parent))) covered.set(pointer(path), path);
+  return Object.fromEntries([...covered].flatMap(([address, path]) => {
+    const content = identity(at(context, path), ["context", ...path]);
+    // Parent delivery must not multiply tiny bodies into a larger journal.
+    return selected.has(address) || Buffer.byteLength(content) >= MIN_REUSABLE_BODY_BYTES
+      ? [[address, createHash("sha256").update(content).digest("hex")]] : [];
+  }));
+}
+
+/** Only previously delivered supplementary bodies may use a reference. Current
+ * decisions, stops, selection, settlement and unknown source shapes stay inline. */
+function observedFirstRead(request: JsonObject, pointers: string[]): JsonObject | undefined {
+  const context = requireJsonObject(request.context, "current first read");
+  const seen = requireJsonObject(request.consumed_pointers, "observed sections");
+  if (typeof jsonObject(context.selected_todo)?.todo_id !== "string") return undefined;
+  const fields = new Map(reusableFirstReadFields(context).map(field => [pointer(field.path), field]));
+  const identity = contextSourceIdentity(request, request.source_number_tokens);
+  const reused: JsonObject = {};
+  // Match semantic fields before the reader groups complete parents. Otherwise
+  // selecting every outcome hides the individual unchanged bodies from reuse.
+  const selections = capturedContextSelection({context, select: pointers,
+    preserve_selection_order: true}).selections as {pointer: string; path: Path}[];
+  const current = selections.flatMap(row => {
+    const {path, pointer: address} = row;
+    const field = fields.get(address);
+    if (typeof seen[address] !== "string" || !field) return [row];
+    const body = jsonObject(at(context, path));
+    if (!body) return [row];
+    const content = identity(body, ["context", ...path]);
+    // A reference and its recovery instructions should cost less than the field.
+    if (Buffer.byteLength(content) < MIN_REUSABLE_BODY_BYTES || createHash("sha256").update(content).digest("hex") !== seen[address])
+      return [row];
+    reused[address] = seen[address];
+    return field.freshKeys.filter(key =>
+      Object.hasOwn(body, key)).map(key => ({pointer: address + "/" + key, path: [...path, key]}));
+  });
+  return Object.keys(reused).length ? {selections: capturedContextSelection({context, initial_read: true},
+    current.map(row => row.pointer)).selections, reused} : undefined;
 }
 
 function inlineReadSections(request: JsonObject, required: JsonObject): JsonObject {
@@ -251,8 +313,12 @@ export function prepareProgressiveContextDelivery(request: JsonObject, legacyFie
   const codec = nativeContextContinuity({...request, context: required.context,
     ...(request.source_number_tokens === undefined ? {} : {source_number_tokens: required.tokens})});
   const delivery = encodeContextDelivery(codec, legacyFields, sourceValues, fileNavigationPaths(required.context));
+  const observe = request.observe_first_read === true && !!jsonObject(required.context.selected_todo)?.todo_id;
+  const pointers = observe ? initialContextReadPointers(required.context, decisionFields) : [];
+  const observed = pointers.length ? observedFirstRead(codec, pointers) : undefined;
   const {operation: _operation, source_edits: _edits, source_number_tokens: _numbers, ...source} = request;
   return contextProjectionChanges(source, {context: required.context, delivery, projected: required.projected,
+    ...(observed ? {observed_first_read: observed} : {}),
     read_sections: inlineReadSections(request, required.context),
     ...(request.include_initial_read_plan === true
       ? {initial_read_plan: nativeContextReadPlan(required.context, decisionFields)} : {})}, sourceValues);

@@ -186,7 +186,7 @@ def prepare_context_delivery(
     retained: Mapping[str, Any] | None = None,
     continuity: Mapping[str, Any] | None = None,
     surface: str = "work_context", native_projection: Mapping[str, Any] | None = None,
-    consumed_pointers: Mapping[str, str] | None = None,
+    consumed_pointers: Mapping[str, str] | None = None, observe_first_read: bool = False,
     include_initial_read_plan: bool = False,
 ) -> dict[str, Any]:
     """Prepare under the Host's explicit continuity basis, not a model assertion.
@@ -208,6 +208,8 @@ def prepare_context_delivery(
             source["include_initial_read_plan"] = True
     if consumed_pointers is not None:
         source["consumed_pointers"] = dict(consumed_pointers)
+        if consumed_pointers and observe_first_read:
+            source["observe_first_read"] = True
     result = _turn_envelope_result("context_delivery", **source, source_edits=True)
     return _apply_context_projection_changes(source, result)
 

@@ -588,10 +588,11 @@ def _native_context_request(state: dict[str, Any], scope: dict[str, Any], captur
     return request
 
 
-def register_native_context_view(directory: str | None, view: dict[str, Any], *, hook_session: str | None = None) -> None:
+def register_native_context_view(directory: str | None, view: dict[str, Any], *,
+                                 hook_session: str | None = None) -> dict[str, Any] | None:
     """Bind the existing pending delivery to its actual paged wire, under lock."""
     if directory is None or not isinstance(view.get("target"), dict):
-        return
+        return None
     with _locked(directory) as root:
         state = _state(root)
         pending = state.get("pending")
@@ -602,6 +603,8 @@ def register_native_context_view(directory: str | None, view: dict[str, Any], *,
             ancestry=view.get("ancestry", 0),
             read_sections=pending.get("full_read_sections", {}) if view.get("mode") == "full" else {})
         _save(root, state)
+        observed = pending.get("observed_first_read")
+        return observed if isinstance(observed, dict) else None
 
 
 @dataclass(frozen=True)
@@ -655,10 +658,13 @@ def prepare_codex_context(payload: dict[str, Any], *, directory: str | None, cap
                 scope = dict(goal_id=goal_id, agent_id=agent_id, session_id=session, context_epoch=epoch)
                 if continuity_generation is not None:
                     scope["continuity_generation"] = continuity_generation
-                if state.get("read_scope") != scope:
+                selected = payload.get("selected_todo")
+                read_scope = {**scope, "surface": surface,
+                    "todo_id": selected.get("todo_id") if isinstance(selected, dict) else None}
+                if state.get("read_scope") != read_scope:
                     for key in ("consumed_pointers", "confirmed_read_sections", "section_reads"):
                         state.pop(key, None)
-                state["read_scope"] = scope
+                state["read_scope"] = read_scope
                 request = _native_context_request(state, scope, capture_id)
                 recovery = capture_context_projection(payload, directory=capture_directory,
                     read_guidance=CONTEXT_CAPTURE_PAGE_GUIDANCE)
@@ -695,7 +701,8 @@ def publish_recovery(reference: dict[str, Any]) -> None:
                             "command_template": context_capture_read_command(Path(recovery["path"]), recovery["sha256"],
                                 page_bytes=page_bytes, max_page_output_bytes=output_bytes)
                                 + " --select '<supporting_material_pointer>/sources'",
-                        } if progressive else None, consumed_pointers=state.get("consumed_pointers"))
+                        } if progressive else None, consumed_pointers=state.get("consumed_pointers"),
+                        observe_first_read=progressive)
                 except (EffectRuntimeRemoteError, EffectRuntimeStartupError):
                     return NativeContextPreparation(payload)
                 if progressive and transaction["projected"]:
@@ -717,6 +724,13 @@ def publish_recovery(reference: dict[str, Any]) -> None:
                     state["pending"].update(target=result["target"], ancestry=result["ancestry"],
                         full_read_sections=read_sections,
                         read_sections=read_sections if result["mode"] == "full" else {})
+                    if plan := transaction.get("observed_first_read"):
+                        from .native_first_read import observed_first_read_reference
+
+                        reference = observed_first_read_reference(directory, result["target"], recovery,
+                            transaction["context"], plan)
+                        if reference is not None:
+                            state["pending"]["observed_first_read"] = reference
                     _save(root, state)
                 read_plan = transaction.get("initial_read_plan")
                 if read_plan is not None:

@@ -299,7 +299,7 @@ def context_capture_read_command(path: Path, digest: str, *, offset: int = 0,
 
 
 def write_context_read_reference(path: Path, digest: str, *, selection: dict[str, Any] | None = None,
-                                 navigation: dict[str, Any] | None = None) -> Path:
+                                 navigation: dict[str, Any] | None = None, observed_read: 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
@@ -310,6 +310,8 @@ def write_context_read_reference(path: Path, digest: str, *, selection: dict[str
         payload["selection"] = selection
     if navigation is not None:
         payload["navigation"] = navigation
+    if observed_read is not None:
+        payload["observed_read"] = observed_read
     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")

@@ -0,0 +1,40 @@
+"""Pin a smaller initial selection to previously observed native tool output."""
+from __future__ import annotations
+
+import hashlib
+import os
+from pathlib import Path
+from typing import Any
+
+from .decision_capture import context_selection_bytes, write_context_read_reference
+
+
+def observed_first_read_reference(directory: str, target: dict[str, Any], recovery: dict[str, Any],
+                                  context: dict[str, Any], plan: dict[str, Any]) -> dict[str, Any] | None:
+    """Write the typed delivery plan once. The host only renders original Python values."""
+    try:
+        selected = plan["selections"]
+        raw = context_selection_bytes(context, selected, list_fields=False, list_depth=None)
+        reference = write_context_read_reference(Path(recovery["path"]), recovery["sha256"], selection={
+            "select": [item["pointer"] for item in selected], "initial_read": False,
+            "list_fields": False, "list_depth": None, "sha256": hashlib.sha256(raw).hexdigest()},
+            observed_read={"directory": directory, "target": target, "fields": plan["reused"]})
+        return {"path": str(reference), "pointer_bytes": sum(len(key.encode()) + 4 for key in plan["reused"])}
+    except (OSError, ValueError):
+        return None  # The ordinary first read remains available.
+
+
+def observation_current(observation: dict[str, Any], path: Path, digest: str) -> bool:
+    from .codex_context import _private_root, _state
+
+    try:
+        state = _state(_private_root(observation["directory"]))
+        recovery = state["recovery"]
+        target = observation["target"]
+        return (target == recovery.get("target") and target["context_epoch"] == state["context_epoch"]
+                and target["session_id"] == state["session_id"] == os.environ.get("CODEX_THREAD_ID")
+                and recovery["sha256"] == digest and Path(recovery["path"]).resolve() == path.resolve()
+                and bool(observation["fields"]) and all(state.get("consumed_pointers", {}).get(pointer) == expected
+                    for pointer, expected in observation["fields"].items()))
+    except (OSError, ValueError, KeyError, TypeError, AttributeError):
+        return False