Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions docs/guides/context-delivery.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.

### Read and recover the right object

Use the path, digest and command returned by the current response. A native
Expand Down
4 changes: 4 additions & 0 deletions docs/guides/context-delivery.zh-CN.md
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,10 @@ Full 表示完整保存视图可用,不要求判断任何问题前先遍历所
首读仍交付当前来源修订和 Turn 控制条件。正文或来源变化、读取不完整或发生压缩时,
读者恢复当前正文。

同一原生上下文周期内,如果同一个已准入 Todo 的写回前置说明逐字相同,且先前已完整交付,
首读可以引用这段说明。当前结算身份、命令和其余步骤仍会交付。说明变化或上下文连续性丢失时,
首读恢复当前原文。显式读取同一 capture 始终能取得当前说明。

### 读对对象,再恢复

使用当前响应返回的路径、摘要与命令。原生响应的 `recovery` 指向当前完整必要视图;
Expand Down
102 changes: 72 additions & 30 deletions loopx/control_plane/quota/progressive_context.ts
Original file line number Diff line number Diff line change
Expand Up @@ -18,21 +18,65 @@ import {createHash} from "node:crypto";


type Path = (string | number)[];
type ReusableReadField = {path: Path; freshKeys: string[]; minimumBytes: number; sourceFile?: string};
type ReusableReadField = {path: Path; freshKeys: string[]; minimumBytes: number; 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): ReusableReadField[] {
const base: Path = ["interaction_contract", "agent_channel", "work_context", "sources"];
const sources = at(context, base);
if (!Array.isArray(sources)) return [];
return sources.flatMap<ReusableReadField>((value, i) => {
const writeback = reusableWritebackPrecondition(context);
const fields = writeback ? [writeback] : [];
if (!Array.isArray(sources)) return fields;
return [...fields, ...sources.flatMap<ReusableReadField>((value, i) => {
const source = jsonObject(value), content = jsonObject(source?.content);
if (!content) return [];
const path = [...base, i, "content"];
Expand All @@ -42,7 +86,7 @@ function reusableFirstReadFields(context: JsonObject): ReusableReadField[] {
(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}];
sourceIdentity: 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 [];
Expand Down Expand Up @@ -72,12 +116,12 @@ function reusableFirstReadFields(context: JsonObject): ReusableReadField[] {
}
return fields;
}) : [];
});
})];
}

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]);
// Matching bytes at an address do not establish a match after a source rebind.
const value = field?.sourceIdentity === undefined ? body : JSON.stringify([field.sourceIdentity, body]);
return createHash("sha256").update(value).digest("hex");
}

Expand Down Expand Up @@ -123,7 +167,7 @@ export function contextReadSectionDigests(request: JsonObject, paths: Path[]): J
}

/** Only registered, previously delivered bodies may use a reference. Current
* decisions, stops, selection, settlement and unknown source shapes stay inline. */
* decisions, stops, settlement commands and identities stay inline. */
function observedFirstRead(request: JsonObject, pointers: string[]): JsonObject | undefined {
const context = requireJsonObject(request.context, "current first read");
const seen = requireJsonObject(request.consumed_pointers, "observed sections");
Expand Down Expand Up @@ -167,17 +211,24 @@ function observedFirstRead(request: JsonObject, pointers: string[]): JsonObject
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--;
// 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;
}
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};
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};
}
Expand Down Expand Up @@ -684,23 +735,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));
Expand Down
19 changes: 15 additions & 4 deletions loopx/control_plane/turn_driver/native_first_read.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,11 +19,22 @@ def observed_first_read_reference(directory: str, target: dict[str, Any], recove
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):
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()}
Expand Down
75 changes: 75 additions & 0 deletions tests/control_plane/test_context_first_read.py
Original file line number Diff line number Diff line change
Expand Up @@ -1185,3 +1185,78 @@ def read(command):
assert Path(fallback[0]["reference_file"]) == ordinary
assert {key: value for key, value in after.items() if key != body_pointer} == facts
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
Loading