perf(quota): reuse observed prefixes in appended Todo notes

docs/guides/context-delivery.md

@@ -81,10 +81,13 @@ canonical read must still establish the selected Todo and its source. The
 note binds to its Goal, Agent, Todo, reader and authority source; changing only
 evidence or other Todo metadata does not make the same literal note new. Current
 task text and controls remain fresh, and metadata keeps its current source and
-existing reader. Partial delivery, changed text, incompatible identity or
-compaction returns the full note; explicit recovery reads the same capture.
-This uses observed field coverage, not file existence as retention proof, and
-does not establish model comprehension or net workflow savings.
+existing reader. If the new note starts with the exact observed note, the reader
+can present only the added text. In an `append_to_observed_text` selection,
+`value` is the suffix after `prefix_bytes` UTF-8 bytes. The next read can use the
+longer note as its prefix only after the Host observes the complete suffix view.
+A partial prior read, rewrite, incompatible identity, or compaction calls for the
+full note. Explicit recovery reads the same capture. Observed delivery does not
+establish model comprehension or net workflow savings.
 
 Replan authoring uses the same-response form: CLI display can reference the
 complete `vision_authoring` in a durable-writeback step. The existing TypeScript

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

@@ -65,10 +65,12 @@ flowchart TD
 原生首读也可以复用当前 Todo 完全相同的备注，前提是同一兼容 Host session 和
 context epoch 中已经观察到整段正文交付；仍须新鲜 canonical 读取确认当前选择的 Todo
 及其来源。备注绑定 Goal、Agent、Todo、reader 和 authority 来源；仅更新证据或其他
-Todo 元数据，不会让相同备注变成新正文。当前任务文本与控制保持新鲜，元数据仍由
-当前来源和既有 reader 提供。只交付一部分、正文变化、身份不兼容或发生压缩时返回
-完整备注；显式恢复读取同一 capture。这依赖已观察的字段覆盖，不以文件存在证明
-保留，也不证明模型理解或完整流程净收益。
+Todo 元数据，不会让相同备注变成新正文。当前任务文本、控制和元数据仍由当前来源及
+既有 reader 提供。当前备注以已完整观察的旧备注开头时，reader 可只送新增文字。
+在 `append_to_observed_text` 选择项里，`value` 是从 `prefix_bytes` 个 UTF-8 字节
+之后开始的新增内容。Host 完整观察到这份增量后，后续读取才可把拼接后的备注作为
+前缀。此前只读到一部分、正文被改写、身份不兼容或发生压缩时，reader 返回完整备注。
+显式恢复仍读同一 capture。观察到交付本身无法证明模型理解或完整流程净收益。
 
 重规划 authoring 使用同响应引用：CLI 展示可指向 durable-writeback 步骤中完整的
 `vision_authoring`。现有 TypeScript Turn owner 选择精确一致的正文，并为普通及捕获

loopx/cli_commands/quota_context_reader.py

@@ -105,9 +105,13 @@ def bind_capture_selection_transport(payload: dict[str, Any], args: argparse.Nam
 
 
 
-def _observed_read_presentation(fields: dict[str, Any]) -> dict[str, Any]:
-    return {"previously_delivered_fields": list(fields),
-        "field_recovery": "These unchanged fields still apply. If needed or uncertain, append --select "
+def _observed_read_presentation(fields: dict[str, Any], text_append: dict[str, Any] | None = None) -> dict[str, Any]:
+    appended = text_append["pointer"] if text_append is not None else None
+    instruction = "These unchanged fields still apply. "
+    if appended is not None:
+        instruction += "For append_to_observed_text, append value to the previously delivered text. "
+    return {"previously_delivered_fields": [pointer for pointer in fields if pointer != appended],
+        "field_recovery": instruction + "If needed or uncertain, append --select "
             "and the field pointer to recovery.command for its complete current body."}
 
 
@@ -282,6 +286,7 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
         selectors = getattr(args, "select", None)
         initial_read = getattr(args, "initial_read", False)
         selection_digest = getattr(args, "selection_sha256", None)
+        text_append = None
         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")
@@ -311,8 +316,11 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
                             reference_file = write_context_read_reference(path, digest)
                         except OSError:
                             reference_file = None
-                elif offset == 0:
-                    presentation = {**(presentation or {}), **_observed_read_presentation(observation["fields"])}
+                else:
+                    text_append = observation.get("text_append")
+                    if offset == 0:
+                        presentation = {**(presentation or {}),
+                            **_observed_read_presentation(observation["fields"], text_append)}
         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"]
@@ -339,7 +347,8 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
                     expected_sha256=selection_digest, preserve_selection_order=pinned_explicit,
                     present_goal_narrative=(pinned_selection.get("present_goal_narrative") is True
                         if pinned_selection is not None else None),
-                    selection_paths=reference.get("selection_paths") if pinned_selection is not None else None)
+                    selection_paths=reference.get("selection_paths") if pinned_selection is not None else None,
+                    text_append=text_append)
             except ContextSelectionDigestChanged as exc:
                 # A verified file digest only binds bytes. Preserve an explicit
                 # selection or listing, including a receipt-only diagnostic.
@@ -634,7 +643,8 @@ def budgeted_ready_opening(navigation: dict[str, Any], plan: dict[str, Any],
         try:
             page = _capture_page(path, content, digest, offset=0, size=page_bytes,
                 output_limit=output_bytes, instruction=False, select=[], reference_file=Path(reference),
-                presentation=_observed_read_presentation(observation["reused"]) if observation else None,
+                presentation=_observed_read_presentation(observation["reused"], observation.get("text_append"))
+                    if observation else None,
                 reader_argv=reader_argv)
         except ValueError:
             pass  # The ordinary reader has no room for its page framing.

loopx/control_plane/quota/progressive_context.ts

@@ -20,7 +20,9 @@ import {createHash} from "node:crypto";
 type Path = (string | number)[];
 export type PreparationScopeDetail = {path: Path; marker_path?: Path; instruction: string};
 type ReadSelection = {pointer: string; path: Path};
-type ReusableReadField = {path: Path; freshKeys: string[]; minimumBytes: number; sourceIdentity?: string | string[]};
+type ReusableReadField = {
+  path: Path; freshKeys: string[]; minimumBytes: number; sourceIdentity?: string | string[]; appendText?: true;
+};
 export type ReusableExactBody = {path: Path; sourceIdentity: string};
 const MIN_REUSABLE_BODY_BYTES = 1024;
 const SETTLEMENT_PLAN_PATH = ["interaction_contract", "cli_channel", "settlement_plan"];
@@ -112,7 +114,7 @@ function reusableFirstReadFields(context: JsonObject, exactBody?: ReusableExactB
         known(content, ["todo", "authority_read"]) && (todo.todo_id === undefined || todo.todo_id === selectedId);
       if (!full && !compact) return [];
       return [{path: [...path, "todo", "note"], freshKeys: [], minimumBytes: MIN_REUSABLE_BODY_BYTES,
-        sourceIdentity: [...route, authority.source_authority, selectedId]}];
+        sourceIdentity: [...route, authority.source_authority, selectedId], appendText: true}];
     }
     if (source?.kind === "agent_preferences")
       return [{path, freshKeys: ["ok", "revision", "instruction"], minimumBytes: MIN_REUSABLE_BODY_BYTES}];
@@ -190,6 +192,32 @@ function observedSectionDigest(body: string, field?: ReusableReadField): string
   return createHash("sha256").update(value).digest("hex");
 }
 
+/** A length-bearing proof lets the current captured note verify a previous
+ * prefix without keeping another copy of that note in the native journal. */
+function observedSectionProof(body: string, value: unknown, field?: ReusableReadField): string {
+  const digest = observedSectionDigest(body, field);
+  return field?.appendText && typeof value === "string" ? `${Buffer.byteLength(value)}:${digest}` : digest;
+}
+
+function matchesObservedSection(expected: unknown, body: string, value: unknown, field?: ReusableReadField): boolean {
+  const proof = observedSectionProof(body, value, field);
+  return expected === proof || (field?.appendText === true && expected === proof.slice(proof.indexOf(":") + 1));
+}
+
+function observedTextAppend(value: unknown, proof: unknown, identity: ReturnType<typeof contextSourceIdentity>,
+    field?: ReusableReadField):
+    {prefixBytes: number; suffix: string} | undefined {
+  if (!field?.appendText || typeof value !== "string" || typeof proof !== "string") return undefined;
+  const parts = /^(\d+):([a-f0-9]{64})$/.exec(proof);
+  if (!parts) return undefined; // A legacy digest can still prove exact equality.
+  const prefixBytes = Number(parts[1]), bytes = Buffer.from(value);
+  if (!Number.isSafeInteger(prefixBytes) || prefixBytes < field.minimumBytes || prefixBytes >= bytes.length
+      || (bytes[prefixBytes]! & 0xc0) === 0x80) return undefined;
+  const prefix = bytes.subarray(0, prefixBytes).toString("utf8");
+  if (observedSectionDigest(identity(prefix, ["context", ...field.path]), field) !== parts[2]) return undefined;
+  return {prefixBytes, suffix: bytes.subarray(prefixBytes).toString("utf8")};
+}
+
 /** 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[], exactBody?: ReusableExactBody): JsonObject {
@@ -213,7 +241,8 @@ export function contextReadSectionDigests(request: JsonObject, paths: Path[], ex
       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 (observedSectionDigest(identity(at(context, path), ["context", ...path]), fields.get(address)) === expected)
+      const value = at(context, path);
+      if (matchesObservedSection(expected, identity(value, ["context", ...path]), value, fields.get(address)))
         coverage.add(address);
     }
     const complete = (path: Path): boolean => {
@@ -224,12 +253,22 @@ export function contextReadSectionDigests(request: JsonObject, paths: Path[], ex
     };
     for (const [address, field] of fields)
       if (complete(field.path)) covered.set(address, field);
+    // The native reader only presents this after authenticating the retained
+    // proof against the current epoch and checking the rendered page bytes.
+    const presented = jsonObject(request.observed_text_append);
+    const address = presented?.pointer;
+    const field = typeof address === "string" ? fields.get(address) : undefined;
+    if (typeof address === "string" && field) {
+      const append = observedTextAppend(at(context, field.path), inherited[address], identity, field);
+      if (append && append.prefixBytes === presented?.prefix_bytes) covered.set(address, field);
+    }
   }
   return Object.fromEntries([...covered].flatMap(([address, {path, minimumBytes}]) => {
-    const content = identity(at(context, path), ["context", ...path]);
+    const value = at(context, path);
+    const content = identity(value, ["context", ...path]);
     // Parent delivery must not multiply tiny bodies into a larger journal.
     return selected.has(address) || Buffer.byteLength(content) >= minimumBytes
-      ? [[address, observedSectionDigest(content, fields.get(address))]] : [];
+      ? [[address, observedSectionProof(content, value, fields.get(address))]] : [];
   }));
 }
 
@@ -268,7 +307,7 @@ function observedFirstRead(request: JsonObject, pointers: string[], exactBody?:
     const body = at(context, path);
     if (typeof seen[address] === "string" && field) {
       const content = identity(body, ["context", ...path]);
-      if (Buffer.byteLength(content) >= field.minimumBytes && observedSectionDigest(content, field) === seen[address]) {
+      if (Buffer.byteLength(content) >= field.minimumBytes && matchesObservedSection(seen[address], content, body, field)) {
         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]}));
@@ -283,16 +322,27 @@ function observedFirstRead(request: JsonObject, pointers: string[], exactBody?:
     return Object.keys(reused).length > before ? children : [row];
   };
   const current = selections.flatMap(visit);
-  if (!Object.keys(reused).length) return authoringOnly();
+  const note = current.find(row => fields.get(row.pointer)?.appendText);
+  const value = note && at(context, note.path);
+  const append = note && observedTextAppend(value, seen[note.pointer], identity, fields.get(note.pointer));
+  if (!append && !Object.keys(reused).length) return authoringOnly();
   const grouped = capturedContextSelection({context, initial_read: true}, current.map(row => row.pointer)).selections as typeof selections;
+  const textAppend = note && append && grouped.some(row => row.pointer === note.pointer)
+    ? {pointer: note.pointer, prefix_bytes: append.prefixBytes} : undefined;
+  if (textAppend) reused[textAppend.pointer] = seen[textAppend.pointer];
+  if (!Object.keys(reused).length) return authoringOnly();
   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.
-  if (bytes(grouped) + Buffer.byteLength(JSON.stringify(Object.keys({...reused, ...retainedAuthoring})))
+  // Reserve space for the appended selection's display marker and prefix length.
+  const appendSavings = textAppend && append
+    ? Buffer.byteLength(JSON.stringify(value)) - Buffer.byteLength(JSON.stringify(append.suffix)) - 128 : 0;
+  if (bytes(grouped) - appendSavings + Buffer.byteLength(JSON.stringify(Object.keys({...reused, ...retainedAuthoring})))
       + MIN_REUSABLE_BODY_BYTES >= bytes(original)) return authoringOnly();
   Object.assign(reused, retainedAuthoring);
+  const observation = {reused, ...(textAppend ? {text_append: textAppend} : {})};
   if (request.include_initial_read_plan === true) {
     // The IO adapter already has the ordinary plan. Keep unchanged runs even
     // when reused sections are far apart in the ordered selection.
@@ -311,9 +361,9 @@ function observedFirstRead(request: JsonObject, pointers: string[], exactBody?:
       splices.push({start, end: original.length, selections: grouped.slice(next)});
     const edits = splices.length === 1 ? {selection_splice: splices[0]} : {selection_splices: splices};
     if (splices.length && Buffer.byteLength(JSON.stringify(edits)) < Buffer.byteLength(JSON.stringify({selections: grouped})))
-      return {...edits, reused};
+      return {...edits, ...observation};
   }
-  return {selections: grouped, reused};
+  return {selections: grouped, ...observation};
 }
 
 function compactObservedSelection(ordinary: ReadSelection[], selected: ReadSelection[]): JsonObject {
@@ -344,7 +394,9 @@ function withLinkedObservedRemainder(observed: JsonObject | undefined,
   const remaining = initial.remaining.selections as ReadSelection[];
   const scoped = remaining.flatMap(row => addresses.has(row.pointer) || selected.some(item => pathStartsWith(row.path, item.path))
     ? [row] : selected.filter(item => pathStartsWith(item.path, row.path)));
-  return {...observed, linked_remaining: compactObservedSelection(remaining, scoped)};
+  const append = jsonObject(observed.text_append);
+  const text = append && scoped.some(row => row.pointer === append.pointer) ? {text_append: append} : {};
+  return {...observed, linked_remaining: {...compactObservedSelection(remaining, scoped), ...text}};
 }
 
 function observedSelection(ordinary: ReadSelection[], observed: JsonObject): ReadSelection[] | undefined {