fix(context): reuse unchanged criteria across progressive first reads

docs/guides/context-delivery.md

@@ -109,6 +109,18 @@ the model has retained it. Retrieve necessary current facts when uncertain.
 Default-off and signed-envelope callers retain their own read contracts; the
 progressive native example does not relax them.
 
+In a continuing native session, a progressive first read can reference exact
+criteria or requirement descriptions delivered by an earlier complete read. An appended
+criterion, revised review or supporting-material projection need not replay
+every unchanged description. Current revisions, changed descriptions, review
+states and new controls invalidate their containing criterion and remain visible.
+An exact unchanged criterion needs no metadata split. Splitting a container adds addresses
+and recovery guidance, so the reader keeps the complete presentation when the
+split would not save enough bytes. This tool-delivery assumption is separate
+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.
+
 ### 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

@@ -89,6 +89,13 @@ Full 表示完整保存视图可用，不要求判断任何问题前先遍历所
 消失，也不证明模型保留了它。不确定时取得必要当前事实。Default-off 与签名 envelope
 调用方继续遵守各自读取契约；这个渐进原生例子不放宽它们。
 
+原生会话持续期间，渐进首读可以引用此前完整读取实际送达的精确条件或要求正文。新增条件、
+修订评审或支持材料投影不必让每条未变正文重送；当前修订号、已改正文、评审状态和
+新控制条件使对应整条条件失效并重新显示；精确未变的整条条件无需拆开元数据。
+拆开容器会增加地址及恢复说明，节省不足时 reader 保留完整展现。
+这种工具投递假设与文件差分、provider 缓存分别成立，不证明模型保留或完整成本下降。
+未读完或连续性丢失不能取得该复用资格；显式同 capture 选择仍可恢复完整当前值。
+
 ### 读对对象，再恢复
 
 使用当前响应返回的路径、摘要与命令。原生响应的 `recovery` 指向当前完整必要视图；

loopx/control_plane/quota/progressive_context.ts

@@ -25,7 +25,7 @@ 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[]}[] {
+function reusableFirstReadFields(context: JsonObject): {path: Path; freshKeys: string[]; minimumBytes: number}[] {
   const base: Path = ["interaction_contract", "agent_channel", "work_context", "sources"];
   const sources = at(context, base);
   if (!Array.isArray(sources)) return [];
@@ -34,13 +34,34 @@ function reusableFirstReadFields(context: JsonObject): {path: Path; freshKeys: s
     if (!content) return [];
     const path = [...base, i, "content"];
     if (source?.kind === "agent_preferences")
-      return [{path, freshKeys: ["ok", "revision", "instruction"]}];
+      return [{path, freshKeys: ["ok", "revision", "instruction"], minimumBytes: MIN_REUSABLE_BODY_BYTES}];
     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"]}] : []) : [];
+    return Array.isArray(outcomes) ? outcomes.flatMap((value, index) => {
+      const outcome = jsonObject(value), path = [...outcomesPath, index];
+      if (typeof outcome?.id !== "string") return [];
+      const fields = [{path, freshKeys: ["id", "revision"], minimumBytes: MIN_REUSABLE_BODY_BYTES}];
+      // An appended criterion, revised review or supporting-material projection
+      // changes the outcome container, not every unchanged requirement body.
+      // Keep all other fields current, including unknown controls and order.
+      if (Array.isArray(outcome.requirements)) {
+        fields.push({path: [...path, "requirements"], freshKeys: [], minimumBytes: MIN_REUSABLE_BODY_BYTES});
+        outcome.requirements.forEach((value, i) => {
+          const requirement = jsonObject(value), body = [...path, "requirements", i, "description"];
+          if (typeof requirement?.id === "string" && typeof requirement.description === "string") {
+            const record = body.slice(0, -1);
+            // Prefer an exact complete criterion before splitting its metadata.
+            // Tiny bodies do not justify a digest and long-address journal row.
+            // Changed status, review or unknown controls invalidate the record;
+            // a large unchanged description can still be considered alone.
+            fields.push({path: record, freshKeys: ["id"], minimumBytes: MIN_REUSABLE_BODY_BYTES});
+            fields.push({path: body, freshKeys: [], minimumBytes: MIN_REUSABLE_BODY_BYTES});
+          }
+        });
+      }
+      return fields;
+    }) : [];
   });
 }
 
@@ -50,13 +71,36 @@ export function contextReadSectionDigests(request: JsonObject, paths: Path[]): J
   const context = requireJsonObject(request.context, "read section source");
   const identity = contextSourceIdentity(request, request.source_number_tokens);
   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 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);
+  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
+  // current literal selections. This is section coverage, never a full-input
+  // baseline. Without it, a partial read strands the next Turn on small fields
+  // even after every part of the revised outcome has actually been delivered.
+  if (inherited && Object.keys(inherited).length) {
+    for (const [address, expected] of Object.entries(inherited)) {
+      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)
+        coverage.add(address);
+    }
+    const complete = (path: Path): boolean => {
+      if (path.map((_, i) => pointer(path.slice(0, i + 1))).concat([""]).some(parent => coverage.has(parent))) return true;
+      const value = at(context, path);
+      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);
+  }
+  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) >= MIN_REUSABLE_BODY_BYTES
+    return selected.has(address) || Buffer.byteLength(content) >= minimumBytes
       ? [[address, createHash("sha256").update(content).digest("hex")]] : [];
   }));
 }
@@ -68,28 +112,43 @@ function observedFirstRead(request: JsonObject, pointers: string[]): JsonObject
   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 parents = new Set([...fields.values()].flatMap(({path}) => path.map((_, i) => pointer(path.slice(0, i)))));
   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 visit = (row: {pointer: string; path: Path}): typeof selections => {
     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;
+    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]) {
+        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]}));
+      }
+    }
+    if (!parents.has(address) || body === null || typeof body !== "object") return [row];
+    const before = Object.keys(reused).length;
+    const children = Object.keys(body).flatMap(key => {
+      const child = [...path, Array.isArray(body) ? Number(key) : key];
+      return visit({pointer: pointer(child), path: child});
+    });
+    return Object.keys(reused).length > before ? children : [row];
+  };
+  const current = selections.flatMap(visit);
+  if (!Object.keys(reused).length) return undefined;
+  const grouped = capturedContextSelection({context, initial_read: true}, current.map(row => row.pointer)).selections as typeof selections;
+  const original = capturedContextSelection({context, initial_read: true}, pointers).selections as typeof selections;
+  const bytes = (rows: typeof selections) => rows.reduce((total, row) => total + Buffer.byteLength(
+    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;
 }
 
 function inlineReadSections(request: JsonObject, required: JsonObject): JsonObject {

loopx/control_plane/turn_driver/native_context_selection.py

@@ -23,9 +23,11 @@ def verified_context_selection(state: dict[str, Any], page: dict[str, Any]) -> d
                   if isinstance(entry := state.get(key), dict)]
     reference_file = page.get("reference_file")
     selected = None
+    observation = None
     if isinstance(reference_file, str):
         _, metadata = read_context_read_reference(reference_file)
         selected = metadata.get("selection")
+        observation = metadata.get("observed_read")
         if not isinstance(selected, dict) or selected.get("list_fields") is not False:
             return None
         path, source_digest = metadata["path"], metadata["sha256"]
@@ -73,6 +75,15 @@ def verified_context_selection(state: dict[str, Any], page: dict[str, Any]) -> d
             or (page.get("eof") is False and (end >= total or page.get("next_offset") != end))
             or type(page.get("eof")) is not bool):
         return None
-    sections = _turn_envelope_result("context_read_sections", context=source, select=pointers)["sections"]
+    inherited = {}
+    if isinstance(observation, dict):
+        from .native_first_read import observation_current
+
+        # The hook already matched its event session/workspace to this journal.
+        # Native hook processes need not inherit the shell's CODEX_THREAD_ID.
+        if observation_current(observation, physical, source_digest, hook_session=state.get("session_id")):
+            inherited = observation["fields"]
+    sections = _turn_envelope_result("context_read_sections", context=source, select=pointers,
+        consumed_pointers=inherited)["sections"]
     return {**page, "schema_version": "context_capture_page_v1", "sha256": view_digest,
         "selected_pointers": pointers, "source_sha256": current["sha256"], "read_sections": sections}

loopx/control_plane/turn_driver/native_first_read.py

@@ -24,15 +24,16 @@ def observed_first_read_reference(directory: str, target: dict[str, Any], recove
         return None  # The ordinary first read remains available.
 
 
-def observation_current(observation: dict[str, Any], path: Path, digest: str) -> bool:
+def observation_current(observation: dict[str, Any], path: Path, digest: str, *,
+                        hook_session: str | None = None) -> 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 target["session_id"] == state["session_id"] == (hook_session or 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()))