Reuse already-read Goal narrative in progressive first reads

docs/guides/context-delivery.md

@@ -129,6 +129,11 @@ from file deltas and provider caching; it does not prove model retention or
 lower total cost. Partial reads and lost continuity cannot qualify this reuse.
 Explicit same-capture selections always recover the complete current value.
 
+An unchanged Goal narrative can also be referenced after an exact read of the
+same registered source in the current session and epoch. The current source
+revision and Turn controls still appear. A change to the narrative or its
+source, an incomplete read, or compaction restores the current text.
+
 ### Read and recover the right object
 
 Use the path, digest and command returned by the current response. A native

docs/guides/context-delivery.zh-CN.md

@@ -102,6 +102,10 @@ Full 表示完整保存视图可用，不要求判断任何问题前先遍历所
 这种工具投递假设与文件差分、provider 缓存分别成立，不证明模型保留或完整成本下降。
 未读完或连续性丢失不能取得该复用资格；显式同 capture 选择仍可恢复完整当前值。
 
+当前原生会话在同一上下文周期内完整读取同一注册来源后，也可引用未变的 Goal 正文。
+首读仍交付当前来源修订和 Turn 控制条件。正文或来源变化、读取不完整或发生压缩时，
+读者恢复当前正文。
+
 ### 读对对象，再恢复
 
 使用当前响应返回的路径、摘要与命令。原生响应的 `recovery` 指向当前完整必要视图；

loopx/cli_commands/quota_capture.py

@@ -143,6 +143,17 @@ def _validate_context_page_window(offset: int, size: int, output_limit: int | No
             "larger pages require a sufficient --max-page-output-bytes within the capture limit.")
 
 
+def _validate_pinned_selection(selection: dict[str, Any]) -> None:
+    if (set(selection) - {"present_goal_narrative"} != {"select", "initial_read", "list_fields", "list_depth", "sha256"}
+            or type(selection["initial_read"]) is not bool or type(selection["list_fields"]) is not bool
+            or ("present_goal_narrative" in selection and (
+                selection["present_goal_narrative"] is not True
+                or selection["list_fields"] or selection["initial_read"]))
+            or not isinstance(selection["sha256"], str)
+            or BARE_SHA256_PATTERN.fullmatch(selection["sha256"]) is None):
+        raise ValueError("reader selection reference is incomplete or unsupported")
+
+
 def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, Any] | None = None,
                          next_page_bytes: int | None = None,
                          next_page_output_bytes: int | None = None,
@@ -204,30 +215,31 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
         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) - {"present_goal_narrative"} !=
-                    {"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 ("present_goal_narrative" in pinned_selection and (
-                        pinned_selection["present_goal_narrative"] is not True
-                        or pinned_selection["list_fields"] or pinned_selection["initial_read"]))
-                    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")
+            _validate_pinned_selection(pinned_selection)
             if observation := reference.get("observed_read"):
-                from ..control_plane.turn_driver.native_first_read import observation_current
+                from ..control_plane.turn_driver.native_first_read import observation_current, observed_first_read_fallback
 
                 if not observation_current(observation, path, digest):
+                    fallback = observed_first_read_fallback(observation, path, digest)
+                    if fallback is not None:
+                        try:
+                            _validate_pinned_selection(fallback[1])
+                        except ValueError:
+                            fallback = None
                     if offset:
                         return {"ok": False, "reason": "Observed first-read scope changed during paging.",
-                            "recovery_command": context_capture_read_command(path, digest, initial_read=True,
+                            "recovery_command": context_capture_read_command(path, digest, initial_read=fallback is None,
+                                reference_file=fallback[0] if fallback is not None else None,
                                 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
+                    if fallback is not None:
+                        reference_file, pinned_selection = fallback
+                    else:
+                        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 "

loopx/control_plane/quota/progressive_context.ts

@@ -3,6 +3,7 @@ import {EffectiveAction} from "./effective_action.generated.ts";
  * addressed in the immutable source. This plan changes no canonical state. */
 import {isDeepStrictEqual} from "node:util";
 import {visionAuthoringContract} from "../goals/vision_checkpoint.ts";
+import {completeGoalNarrative} from "../work_items/goal_context_source.ts";
 import type {JsonObject} from "../effect_program.ts";
 import {jsonObject, requireBoolean, requireJsonObject} from "../runtime_decode.ts";
 import {GOAL_OUTCOMES_SCHEMA, OUTCOME_CONTEXT_NAVIGATION_SCHEMA} from "../goals/outcome_requirements.ts";
@@ -17,6 +18,7 @@ import {createHash} from "node:crypto";
 
 
 type Path = (string | number)[];
+type ReusableReadField = {path: Path; freshKeys: string[]; minimumBytes: number; sourceFile?: string};
 const MIN_REUSABLE_BODY_BYTES = 1024;
 const pointer = (path: Path): string => path.length ? "/" + path.map(key =>
   String(key).replaceAll("~", "~0").replaceAll("/", "~1")).join("/") : "";
@@ -26,14 +28,21 @@ const known = (row: JsonObject, fields: string[]): boolean => Object.keys(row).e
 
 /** 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[]; minimumBytes: number}[] {
+function reusableFirstReadFields(context: JsonObject): ReusableReadField[] {
   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) => {
+  return sources.flatMap<ReusableReadField>((value, i) => {
     const source = jsonObject(value), content = jsonObject(source?.content);
     if (!content) return [];
     const path = [...base, i, "content"];
+    // The current source and full-file revision still reach the first read.
+    // Only the literal narrative may carry forward across generated Todo edits.
+    if (source?.source === "goal_state" && content.ok !== false &&
+        (content.status === undefined || content.status === "loaded") &&
+        typeof content.state_file === "string" && completeGoalNarrative(content))
+      return [{path: [...path, "text"], freshKeys: [], minimumBytes: MIN_REUSABLE_BODY_BYTES,
+        sourceFile: content.state_file}];
     if (source?.kind === "agent_preferences")
       return [{path, freshKeys: ["ok", "revision", "instruction"], minimumBytes: MIN_REUSABLE_BODY_BYTES}];
     if (source?.source !== "goal_acceptance") return [];
@@ -66,15 +75,22 @@ function reusableFirstReadFields(context: JsonObject): {path: Path; freshKeys: s
   });
 }
 
+function observedSectionDigest(body: string, field?: ReusableReadField): string {
+  // Same text at a reused array index is insufficient after a Goal file rebind.
+  const value = field?.sourceFile === undefined ? body : JSON.stringify([field.sourceFile, body]);
+  return createHash("sha256").update(value).digest("hex");
+}
+
 /** 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);
   const selected = new Set(paths.map(pointer));
+  const fields = new Map(reusableFirstReadFields(context).map(field => [pointer(field.path), field]));
   const covered = new Map(paths.map(path => [pointer(path), {path, minimumBytes: MIN_REUSABLE_BODY_BYTES}]));
-  for (const field of reusableFirstReadFields(context))
-    if (paths.some(parent => pathStartsWith(field.path, parent))) covered.set(pointer(field.path), field);
+  for (const [address, field] of fields)
+    if (paths.some(parent => pathStartsWith(field.path, parent))) covered.set(address, field);
   const inherited = jsonObject(request.consumed_pointers), coverage = new Set(paths.map(pointer));
   // The IO owner authenticates the reader's native observation. Recheck each
   // inherited value against this exact source before combining it with the
@@ -86,7 +102,7 @@ export function contextReadSectionDigests(request: JsonObject, paths: Path[]): J
       const read = capturedContextSelection({context, select: [address], recover_selection_errors: true});
       if (read.rejected_selection) continue;
       const path = (read.selections as {path: Path}[])[0].path;
-      if (createHash("sha256").update(identity(at(context, path), ["context", ...path])).digest("hex") === expected)
+      if (observedSectionDigest(identity(at(context, path), ["context", ...path]), fields.get(address)) === expected)
         coverage.add(address);
     }
     const complete = (path: Path): boolean => {
@@ -95,18 +111,18 @@ export function contextReadSectionDigests(request: JsonObject, paths: Path[]): J
       return value !== null && typeof value === "object" && Object.keys(value).length > 0
         && Object.keys(value).every(key => complete([...path, Array.isArray(value) ? Number(key) : key]));
     };
-    for (const field of reusableFirstReadFields(context))
-      if (complete(field.path)) covered.set(pointer(field.path), field);
+    for (const [address, field] of fields)
+      if (complete(field.path)) covered.set(address, field);
   }
   return Object.fromEntries([...covered].flatMap(([address, {path, minimumBytes}]) => {
     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) >= minimumBytes
-      ? [[address, createHash("sha256").update(content).digest("hex")]] : [];
+      ? [[address, observedSectionDigest(content, fields.get(address))]] : [];
   }));
 }
 
-/** Only previously delivered supplementary bodies may use a reference. Current
+/** Only registered, previously delivered 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");
@@ -126,7 +142,7 @@ function observedFirstRead(request: JsonObject, pointers: string[]): JsonObject
     const body = at(context, path);
     if (typeof seen[address] === "string" && field) {
       const content = identity(body, ["context", ...path]);
-      if (Buffer.byteLength(content) >= field.minimumBytes && createHash("sha256").update(content).digest("hex") === seen[address]) {
+      if (Buffer.byteLength(content) >= field.minimumBytes && observedSectionDigest(content, field) === seen[address]) {
         reused[address] = seen[address];
         return field.freshKeys.filter(key => body !== null && typeof body === "object" &&
           Object.hasOwn(body, key)).map(key => ({pointer: pointer([...path, key]), path: [...path, key]}));
@@ -148,8 +164,22 @@ function observedFirstRead(request: JsonObject, pointers: string[]): JsonObject
     JSON.stringify({pointer: row.pointer}).slice(0, -1) + ',"value":' + identity(at(context, row.path), ["context", ...row.path]) + '}'), 0);
   // Splitting a container repeats addresses. Count those and the reuse index;
   // retain a margin for the existing recovery guidance, or deliver it whole.
-  return bytes(grouped) + Buffer.byteLength(JSON.stringify(Object.keys(reused))) + MIN_REUSABLE_BODY_BYTES < bytes(original)
-    ? {selections: grouped, reused} : undefined;
+  if (bytes(grouped) + Buffer.byteLength(JSON.stringify(Object.keys(reused))) + MIN_REUSABLE_BODY_BYTES >= bytes(original))
+    return undefined;
+  if (request.include_initial_read_plan === true) {
+    // The IO adapter already receives the ordinary plan. Send the changed
+    // middle of that list, keeping its exact order and coalescing decisions.
+    let start = 0, end = original.length, stop = grouped.length;
+    while (start < end && start < stop && original[start].pointer === grouped[start].pointer) start++;
+    while (end > start && stop > start && original[end - 1].pointer === grouped[stop - 1].pointer) {
+      end--;
+      stop--;
+    }
+    const selection_splice = {start, end, selections: grouped.slice(start, stop)};
+    if (Buffer.byteLength(JSON.stringify(selection_splice)) < Buffer.byteLength(JSON.stringify(grouped)))
+      return {selection_splice, reused};
+  }
+  return {selections: grouped, reused};
 }
 
 function inlineReadSections(request: JsonObject, required: JsonObject): JsonObject {

loopx/control_plane/turn_driver/codex_context.py

@@ -785,7 +785,7 @@ def publish_recovery(reference: dict[str, Any]) -> None:
                         from .native_first_read import observed_first_read_reference
 
                         reference = observed_first_read_reference(directory, result["target"], recovery,
-                            transaction["context"], plan)
+                            transaction["context"], plan, transaction.get("initial_read_plan"))
                         if reference is not None:
                             state["pending"]["observed_first_read"] = reference
                     _save(root, state)

loopx/control_plane/turn_driver/native_first_read.py

@@ -1,25 +1,63 @@
 """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 write_initial_context_read_reference
+from .decision_capture import read_context_read_reference, write_initial_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 renders it from the immutable source."""
+                                  context: dict[str, Any], plan: dict[str, Any],
+                                  initial_read_plan: dict[str, Any] | None = None) -> dict[str, Any] | None:
+    """Pin the reduced view and its ordinary fallback from the same capture."""
     try:
-        reference, _ = write_initial_context_read_reference(Path(recovery["path"]), recovery["sha256"],
-            context, plan["selections"],
-            observed_read={"directory": directory, "target": target, "fields": plan["reused"]})
+        path, digest = Path(recovery["path"]), recovery["sha256"]
+        observation = {"directory": directory, "target": target, "fields": plan["reused"]}
+        selections = plan.get("selections")
+        if initial_read_plan is not None and initial_read_plan.get("initial") is True:
+            ordinary = initial_read_plan["plan"]["selections"]
+            if splice := plan.get("selection_splice"):
+                start, end = splice["start"], splice["end"]
+                if type(start) is not int or type(end) is not int or not 0 <= start <= end <= len(ordinary):
+                    return None
+                selections = [*ordinary[:start], *splice["selections"], *ordinary[end:]]
+            fallback, body = write_initial_context_read_reference(path, digest, context, ordinary)
+            observation["fallback_reference"] = {"filename": fallback.name,
+                "selection_sha256": hashlib.sha256(body).hexdigest()}
+        if selections is None:
+            return None
+        reference, _ = write_initial_context_read_reference(path, digest, context, selections,
+            observed_read=observation)
         return {"path": str(reference), "pointer_bytes": sum(len(key.encode()) + 4 for key in plan["reused"])}
-    except (OSError, ValueError):
+    except (OSError, ValueError, KeyError, TypeError):
         return None  # The ordinary first read remains available.
 
 
+def observed_first_read_fallback(observation: dict[str, Any], path: Path, digest: str) -> tuple[Path, dict[str, Any]] | None:
+    """Locate the ordinary pinned selection of this exact capture."""
+    fallback = observation.get("fallback_reference") if isinstance(observation, dict) else None
+    if not isinstance(fallback, dict):
+        return None
+    filename = fallback.get("filename")
+    if not isinstance(filename, str) or not filename.startswith(path.stem + ".") or not filename.endswith(".reader.json"):
+        return None
+    try:
+        reference, metadata = read_context_read_reference(str(path.with_name(filename)))
+        selection = metadata.get("selection")
+        source = Path(metadata["path"]).expanduser().absolute()
+        if (metadata["sha256"] == digest and source.parent.resolve(strict=True) / source.name == path
+                and "observed_read" not in metadata and isinstance(selection, dict)
+                and isinstance(fallback.get("selection_sha256"), str)
+                and selection.get("sha256") == fallback["selection_sha256"]):
+            return reference, selection
+    except (OSError, ValueError):
+        pass
+    return None
+
+
 def observation_current(observation: dict[str, Any], path: Path, digest: str, *,
                         hook_session: str | None = None) -> bool:
     from .codex_context import _private_root, _state