Reuse observed writeback guidance in native context

docs/guides/context-delivery.md

@@ -134,6 +134,12 @@ 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.
 
+For the same ready Todo in a native context epoch, the first read can reference
+an exactly matching writeback precondition already delivered in full. It still
+delivers the current settlement identity, commands and other steps. Changed
+guidance and lost continuity restore the literal text. An explicit same-capture
+read always returns the current precondition.
+
 For ready work on the same registered route and Todo, the native first read can
 reference an exact write-scope list that the session already received in full.
 The rest of the current boundary, including guards and stops, still appears.

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

@@ -106,6 +106,10 @@ Full 表示完整保存视图可用，不要求判断任何问题前先遍历所
 首读仍交付当前来源修订和 Turn 控制条件。正文或来源变化、读取不完整或发生压缩时，
 读者恢复当前正文。
 
+同一原生上下文周期内，如果同一个已准入 Todo 的写回前置说明逐字相同，且先前已完整交付，
+首读可以引用这段说明。当前结算身份、命令和其余步骤仍会交付。说明变化或上下文连续性丢失时，
+首读恢复当前原文。显式读取同一 capture 始终能取得当前说明。
+
 同一注册路由和 Todo 的工作已就绪时，原生首读可以引用会话此前完整收到的同一份写入
 范围列表。当前边界的其他部分仍会显示，包括防护和停止条件。不确定某个路径是否在
 范围内时，通过当前恢复读取命令选择 `/goal_boundary/write_scope`。列表或路由变化、

loopx/control_plane/quota/progressive_context.ts

@@ -22,25 +22,69 @@ type ReadSelection = {pointer: string; path: Path};
 type ReusableReadField = {path: Path; freshKeys: string[]; minimumBytes: number; sourceIdentity?: string | string[]};
 export type ReusableExactBody = {path: Path; sourceIdentity: string};
 const MIN_REUSABLE_BODY_BYTES = 1024;
+const SETTLEMENT_PLAN_PATH = ["interaction_contract", "cli_channel", "settlement_plan"];
 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));
 
+function orderedSettlementSteps(plan: JsonObject | null | undefined): unknown[] | undefined {
+  return (plan?.schema_version === "quota_settlement_plan_v1" || plan?.schema_version === "quota_settlement_plan_v2")
+    && Array.isArray(plan.ordered_steps) ? plan.ordered_steps : undefined;
+}
+
+function readySettlementIdentity(context: JsonObject, interaction: JsonObject,
+    plan: JsonObject | null | undefined): JsonObject | undefined {
+  const cli = jsonObject(interaction.cli_channel), prepared = jsonObject(cli?.execution_preparation);
+  const identity = jsonObject(plan?.identity);
+  return context.effective_action === EffectiveAction.NORMAL_RUN && context.normal_delivery_allowed === true
+    && jsonObject(interaction.agent_channel)?.delivery_allowed === true && cli?.selection_required !== true
+    && prepared?.state === "ready" && typeof identity?.effect_id === "string" && identity.effect_id.length > 0
+    && typeof identity.todo_id === "string" && prepared.todo_id === identity.todo_id
+    && jsonObject(context.selected_todo)?.todo_id === identity.todo_id
+    && isDeepStrictEqual(identity, jsonObject(jsonObject(context.heartbeat_receipt)?.settlement_identity))
+    ? identity : undefined;
+}
+
+function reusableWritebackPrecondition(context: JsonObject): ReusableReadField | undefined {
+  const interaction = jsonObject(context.interaction_contract);
+  if (interaction?.schema_version !== "loopx_interaction_contract_v0") return undefined;
+  const plan = jsonObject(at(context, SETTLEMENT_PLAN_PATH)), steps = orderedSettlementSteps(plan);
+  const identity = readySettlementIdentity(context, interaction, plan);
+  const identityAgent = jsonObject(context.agent_identity)?.agent_id;
+  const agent = context.agent_id ?? identityAgent;
+  if (!plan || !steps || !identity || typeof agent !== "string" || agent !== identity.agent_id
+      || (identityAgent !== undefined && identityAgent !== agent)
+      || typeof context.goal_id !== "string" || context.goal_id !== identity.goal_id
+      || typeof context.registry !== "string" || !context.registry
+      || typeof context.runtime_root !== "string" || !context.runtime_root) return undefined;
+  const writebacks = steps.map((value, index) => ({step: jsonObject(value), index}))
+    .filter(({step}) => step?.kind === "durable_writeback");
+  if (writebacks.length !== 1) return undefined;
+  const {step, index} = writebacks[0]!;
+  if (step?.owner !== "agent" || typeof step.precondition !== "string") return undefined;
+  return {path: [...SETTLEMENT_PLAN_PATH, "ordered_steps", index, "precondition"],
+    freshKeys: [], minimumBytes: MIN_REUSABLE_BODY_BYTES,
+    sourceIdentity: JSON.stringify([context.registry, context.runtime_root, context.goal_ref ?? null,
+      plan.schema_version, identity.goal_id, agent, identity.todo_id, step.kind, step.owner])};
+}
+
 /** 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, exactBody?: ReusableExactBody): ReusableReadField[] {
   const exactFields = exactBody && at(context, exactBody.path) !== undefined
     ? [{...exactBody, freshKeys: [], minimumBytes: MIN_REUSABLE_BODY_BYTES}] : [];
+  const writeback = reusableWritebackPrecondition(context);
+  const fields = [...exactFields, ...(writeback ? [writeback] : [])];
   const workPath: Path = ["interaction_contract", "agent_channel", "work_context"];
   const work = jsonObject(at(context, workPath));
   const base = [...workPath, "sources"];
   const sources = work?.sources;
-  if (!Array.isArray(sources)) return exactFields;
+  if (!Array.isArray(sources)) return fields;
   const selectedId = jsonObject(context.selected_todo)?.todo_id;
   const selectedSources = sources.filter(value => jsonObject(value)?.source === "selected_todo").length;
-  return [...exactFields, ...sources.flatMap<ReusableReadField>((value, i) => {
+  return [...fields, ...sources.flatMap<ReusableReadField>((value, i) => {
     const source = jsonObject(value), content = jsonObject(source?.content);
     if (!content) return [];
     const path = [...base, i, "content"];
@@ -192,8 +236,27 @@ function observedFirstRead(request: JsonObject, pointers: string[], exactBody?:
   // retain a margin for the existing recovery guidance, or deliver it whole.
   if (bytes(grouped) + Buffer.byteLength(JSON.stringify(Object.keys(reused))) + MIN_REUSABLE_BODY_BYTES >= bytes(original))
     return undefined;
-  return {...(request.include_initial_read_plan === true
-    ? compactObservedSelection(original, grouped) : {selections: grouped}), reused};
+  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.
+    const positions = new Map(grouped.map((row, index) => [row.pointer, index]));
+    const splices: {start: number; end: number; selections: typeof selections}[] = [];
+    let start = 0, next = 0;
+    for (let end = 0; end < original.length; end++) {
+      const position = positions.get(original[end].pointer);
+      if (position === undefined || position < next) continue;
+      if (start < end || next < position)
+        splices.push({start, end, selections: grouped.slice(next, position)});
+      start = end + 1;
+      next = position + 1;
+    }
+    if (start < original.length || next < grouped.length)
+      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 {selections: grouped, reused};
 }
 
 function compactObservedSelection(ordinary: ReadSelection[], selected: ReadSelection[]): JsonObject {
@@ -218,17 +281,36 @@ function withLinkedObservedRemainder(observed: JsonObject | undefined,
   if (!observed || !initial?.linked_opening_detail || !initial.remaining
       || Object.hasOwn(observed.reused as JsonObject, initial.linked_opening_detail)) return observed;
   const ordinary = initial.plan.selections as ReadSelection[];
-  const splice = observed.selection_splice as {start: number; end: number; selections: ReadSelection[]} | undefined;
-  const selected = splice
-    ? [...ordinary.slice(0, splice.start), ...splice.selections, ...ordinary.slice(splice.end)]
-    : observed.selections as ReadSelection[];
+  const selected = observedSelection(ordinary, observed);
+  if (selected === undefined) return observed;
   const addresses = new Set(selected.map(row => row.pointer));
   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)};
 }
 
+function observedSelection(ordinary: ReadSelection[], observed: JsonObject): ReadSelection[] | undefined {
+  const candidate = observed.selection_splices ?? (observed.selection_splice === undefined
+    ? undefined : [observed.selection_splice]);
+  if (candidate === undefined)
+    return Array.isArray(observed.selections) ? observed.selections as ReadSelection[] : undefined;
+  if (!Array.isArray(candidate) || !candidate.length) return undefined;
+  const selected: ReadSelection[] = [];
+  let cursor = 0;
+  for (const value of candidate) {
+    const splice = jsonObject(value);
+    const start = splice?.start, end = splice?.end, replacement = splice?.selections;
+    if (typeof start !== "number" || !Number.isSafeInteger(start) || typeof end !== "number"
+        || !Number.isSafeInteger(end) || start < cursor || end < start || end > ordinary.length
+        || !Array.isArray(replacement)) return undefined;
+    selected.push(...ordinary.slice(cursor, start), ...replacement as ReadSelection[]);
+    cursor = end;
+  }
+  selected.push(...ordinary.slice(cursor));
+  return selected;
+}
+
 function inlineReadSections(request: JsonObject, required: JsonObject, exactBody?: ReusableExactBody): JsonObject {
   const original = requireJsonObject(request.context, "inline source"), paths: Path[] = [];
   const visit = (source: unknown, shown: unknown, path: Path) => {
@@ -792,23 +874,14 @@ function initialContextReadGroups(context: JsonObject, decisionFields: readonly
     include(...source, "content", "harness", "orchestration_gate");
   });
   const cli = object(interaction.cli_channel);
-  const planPath = ["interaction_contract", "cli_channel", "settlement_plan"];
+  const planPath = SETTLEMENT_PLAN_PATH;
   const plan = object(cli?.settlement_plan);
-  const steps = (plan?.schema_version === "quota_settlement_plan_v1" || plan?.schema_version === "quota_settlement_plan_v2")
-    && Array.isArray(plan.ordered_steps)
-    ? plan.ordered_steps : undefined;
-  const prepared = object(cli?.execution_preparation);
-  const identity = object(plan?.identity);
+  const steps = orderedSettlementSteps(plan);
   const settlementSteps = steps?.map(object).filter(step => step?.kind === "durable_writeback" || step?.kind === "quota_spend");
   // A ready Turn already shows its ordered writeback and spend with their
   // conditions. Hide only the exact duplicate fallback in this first read;
   // recovery, legacy readers and the original capture keep both forms.
-  const repeatsSettlement = context.effective_action === EffectiveAction.NORMAL_RUN && context.normal_delivery_allowed === true
-    && channel?.delivery_allowed === true && cli?.selection_required !== true && prepared?.state === "ready"
-    && typeof identity?.effect_id === "string" && identity.effect_id.length > 0
-    && typeof identity.todo_id === "string" && prepared.todo_id === identity.todo_id
-    && object(context.selected_todo)?.todo_id === identity.todo_id
-    && isDeepStrictEqual(identity, object(object(context.heartbeat_receipt)?.settlement_identity))
+  const repeatsSettlement = readySettlementIdentity(context, interaction, plan) !== undefined
     && settlementSteps?.length === 2 && settlementSteps[0]?.kind === "durable_writeback" && settlementSteps[1]?.kind === "quota_spend"
     && settlementSteps.every(step => typeof step?.command_template === "string" && step.command_template.length > 0)
     && isDeepStrictEqual(cli?.next_cli_actions, settlementSteps.map(step => step?.command_template));

loopx/control_plane/turn_driver/native_first_read.py

@@ -45,11 +45,22 @@ def _pin_observed_selection(directory: str, target: dict[str, Any], recovery: di
         selections = plan.get("selections")
         if ordinary_plan is not None:
             ordinary = ordinary_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):
+            splices = plan.get("selection_splices")
+            if splices is None and (splice := plan.get("selection_splice")) is not None:
+                splices = [splice]
+            if splices is not None:
+                if not isinstance(splices, list) or not splices:
                     return None
-                selections = [*ordinary[:start], *splice["selections"], *ordinary[end:]]
+                selections, cursor = [], 0
+                for splice in splices:
+                    start, end, replacement = splice["start"], splice["end"], splice["selections"]
+                    if (type(start) is not int or type(end) is not int or not cursor <= start <= end <= len(ordinary)
+                            or not isinstance(replacement, list)):
+                        return None
+                    selections.extend(ordinary[cursor:start])
+                    selections.extend(replacement)
+                    cursor = end
+                selections.extend(ordinary[cursor:])
             fallback, body = write_initial_context_read_reference(path, digest, context, ordinary)
             observation["fallback_reference"] = {"filename": fallback.name,
                 "selection_sha256": hashlib.sha256(body).hexdigest()}