diff --git a/docs/guides/context-delivery.md b/docs/guides/context-delivery.md index d4af7f341a..5a3db43c4c 100644 --- a/docs/guides/context-delivery.md +++ b/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. diff --git a/docs/guides/context-delivery.zh-CN.md b/docs/guides/context-delivery.zh-CN.md index 3464dbe298..b5a98b6266 100644 --- a/docs/guides/context-delivery.zh-CN.md +++ b/docs/guides/context-delivery.zh-CN.md @@ -106,6 +106,10 @@ Full 表示完整保存视图可用,不要求判断任何问题前先遍历所 首读仍交付当前来源修订和 Turn 控制条件。正文或来源变化、读取不完整或发生压缩时, 读者恢复当前正文。 +同一原生上下文周期内,如果同一个已准入 Todo 的写回前置说明逐字相同,且先前已完整交付, +首读可以引用这段说明。当前结算身份、命令和其余步骤仍会交付。说明变化或上下文连续性丢失时, +首读恢复当前原文。显式读取同一 capture 始终能取得当前说明。 + 同一注册路由和 Todo 的工作已就绪时,原生首读可以引用会话此前完整收到的同一份写入 范围列表。当前边界的其他部分仍会显示,包括防护和停止条件。不确定某个路径是否在 范围内时,通过当前恢复读取命令选择 `/goal_boundary/write_scope`。列表或路由变化、 diff --git a/loopx/control_plane/quota/progressive_context.ts b/loopx/control_plane/quota/progressive_context.ts index c3e5cd60ce..fc9da0f06f 100644 --- a/loopx/control_plane/quota/progressive_context.ts +++ b/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((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((value, i) => { + return [...fields, ...sources.flatMap((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,10 +281,8 @@ 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)) @@ -229,6 +290,27 @@ function withLinkedObservedRemainder(observed: JsonObject | undefined, 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)); diff --git a/loopx/control_plane/turn_driver/native_first_read.py b/loopx/control_plane/turn_driver/native_first_read.py index 7618471fda..8999ce117c 100644 --- a/loopx/control_plane/turn_driver/native_first_read.py +++ b/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()} diff --git a/tests/control_plane/test_context_first_read.py b/tests/control_plane/test_context_first_read.py index b9db5ee9ba..33c1287eb3 100644 --- a/tests/control_plane/test_context_first_read.py +++ b/tests/control_plane/test_context_first_read.py @@ -1447,6 +1447,80 @@ def read(command): assert fallback_bytes - warm_bytes > 3000 +@pytest.mark.parametrize("version", ["quota_settlement_plan_v1", "quota_settlement_plan_v2"]) +@pytest.mark.parametrize("transition", ["same", "changed", "wrong_binding", "unknown_version", "rebound"]) +def test_native_first_read_reuses_only_observed_writeback_precondition( + tmp_path, monkeypatch, decision, version, transition, +): + from test_codex_context_lifecycle import deliver, setup + + event = setup(tmp_path, monkeypatch) + decision.update(agent_id="agent", should_run=True, normal_delivery_allowed=True, effective_action="normal_run") + decision.update(registry=str(tmp_path / "registry.json"), runtime_root=str(tmp_path / "runtime")) + contract = decision["interaction_contract"] + contract["agent_channel"]["delivery_allowed"] = True + source = contract["agent_channel"]["work_context"]["sources"][0] + narrative = source["content"]["text"] + source.update(source="goal_state", content={ + "state_file": "fixture-goal.md", "full_source": True, "text": narrative, + "revision": "sha256:" + hashlib.sha256(narrative.encode()).hexdigest(), + }) + goal_text = "/interaction_contract/agent_channel/work_context/sources/0/content/text" + cli = contract["cli_channel"] + cli["execution_preparation"] = {"state": "ready", "todo_id": "todo_current", "agent_id": "agent"} + plan = cli["settlement_plan"] + plan["schema_version"] = version + text = "After independent validation, write back under the original identity and respect the current stop. " * 38 + step = {"kind": "durable_writeback", "owner": "agent", "precondition": text} + plan["ordered_steps"] = [step] + root = "/interaction_contract/cli_channel/settlement_plan" + precondition = root + "/ordered_steps/0/precondition" + command = root + "/ordered_steps/0/command_template" + + def capture(turn): + identity = {"goal_id": "fixture", "agent_id": "agent", "todo_id": "todo_current", + "turn_instance_id": turn, "effect_id": f"fixture:agent:todo_current:{turn}"} + decision["heartbeat_receipt"] = {"settlement_identity": identity} + plan["identity"] = {**identity, "effect_id": "mismatch"} if turn == "next" and transition == "wrong_binding" else identity + step["command_template"] = f"loopx refresh-state --turn-instance-id {turn}" + directory = tmp_path / turn + directory.mkdir(mode=0o700) + prepared = codex_context.prepare_codex_context(decision, directory=str(tmp_path), capture_directory=directory, + goal_id="fixture", agent_id="agent", surface="quota_packet", capture_id=turn) + return model_context_reference(native_context_capture_view(prepared.delivery, directory, + context_directory=str(tmp_path), initial_read_plan=prepared.initial_read_plan)) + + first = capture("first") + page, facts = read_selection(first["first_read_command"]) + assert facts[precondition] == text and "previously_delivered_fields" not in page + deliver(tmp_path, event, [page]) + + if transition == "changed": + step["precondition"] += " Await the updated external approval." + elif transition == "unknown_version": + plan["schema_version"] = "unknown" + elif transition == "rebound": + decision["registry"] = str(tmp_path / "another-registry.json") + decision["goal_boundary"]["stop_condition"] = "Stop before publishing." + second = capture("next") + page, facts = read_selection(second["first_read_command"]) + assert facts[command] == "loopx refresh-state --turn-instance-id next" + assert facts["/goal_boundary"] == decision["goal_boundary"] + assert facts[root + "/identity/effect_id"] == plan["identity"]["effect_id"] + if transition != "same": + assert precondition not in page.get("previously_delivered_fields", []) + assert facts[precondition] == step["precondition"] + return + + assert set(page["previously_delivered_fields"]) == {goal_text, precondition} + assert goal_text not in facts and precondition not in facts + _, recovered = read_selection(second["recovery"]["command"] + " --select " + precondition) + assert recovered[precondition] == text + codex_context.codex_context_hook(str(tmp_path), {**event, "hook_event_name": "PostCompact"}) + page, facts = read_selection(second["first_read_command"]) + assert "previously_delivered_fields" not in page and facts[precondition] == text + assert facts[goal_text] == narrative + @pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_native_first_read_reuses_only_observed_write_scope(tmp_path, monkeypatch, provider): from test_codex_context_lifecycle import deliver, setup @@ -1574,8 +1648,14 @@ def read_initial(view): if key != "write_scope": assert warm_facts[f"/goal_boundary/{key}"] == value assert warm_facts["/interaction_contract/cli_channel/settlement_plan/identity/turn_instance_id"] == "scope-warm" - assert {key: value for key, value in restored_facts.items() - if key != "/goal_boundary" and not key.startswith(scope_path)} == warm_facts + restored_selection = {key: value for key, value in restored_facts.items() + if key != "/goal_boundary" and not key.startswith(scope_path)} + observed = warm[0].get("previously_delivered_fields", []) + assert restored_selection == { + **warm_facts, + **{key: value for key, value in restored_selection.items() + if any(key == pointer or key.startswith(pointer + "/") for pointer in observed)}, + } assert restored_bytes - warm_bytes > 10_000 @@ -1712,5 +1792,11 @@ def read(command): "warm_pages": len(warm), "recovery_pages": len(recovered)})) assert pointer in warm[0].get("previously_delivered_fields", []) and pointer not in facts assert facts[source_pointer + "/authority_read"] == current_authority - assert {key: value for key, value in restored.items() if key != pointer} == facts + restored_note = {key: value for key, value in restored.items() if key != pointer} + observed = warm[0].get("previously_delivered_fields", []) + assert restored_note == { + **facts, + **{key: value for key, value in restored_note.items() + if any(key == address or key.startswith(address + "/") for address in observed)}, + } assert fallback_bytes - warm_bytes > 10_000