From 2f0a5c311130fbc017ee96ef0cbf78503c0f4b54 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 11 Nov 2026 03:50:48 +0800 Subject: [PATCH] perf(quota): reuse observed prefixes in appended Todo notes Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Signed-off-by: lusendong.6789 Signed-off-by: lusendong.6789 Co-authored-by: TRAE CLI Signed-off-by: huangruiteng --- docs/guides/context-delivery.md | 11 ++- docs/guides/context-delivery.zh-CN.md | 10 ++- loopx/cli_commands/quota_context_reader.py | 24 ++++-- .../quota/progressive_context.ts | 74 ++++++++++++++++--- .../turn_driver/decision_capture.py | 33 +++++++-- .../turn_driver/native_context_selection.py | 50 ++++++++++--- .../turn_driver/native_first_read.py | 15 +++- .../control_plane/test_context_first_read.py | 61 +++++++++++++-- .../progressive_context.test.ts | 45 ++++++++++- 9 files changed, 272 insertions(+), 51 deletions(-) diff --git a/docs/guides/context-delivery.md b/docs/guides/context-delivery.md index 27f8e75d31..9994534972 100644 --- a/docs/guides/context-delivery.md +++ b/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 diff --git a/docs/guides/context-delivery.zh-CN.md b/docs/guides/context-delivery.zh-CN.md index 09df05d068..edbd782222 100644 --- a/docs/guides/context-delivery.zh-CN.md +++ b/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 选择精确一致的正文，并为普通及捕获 diff --git a/loopx/cli_commands/quota_context_reader.py b/loopx/cli_commands/quota_context_reader.py index 2dbc48889f..290a16d2c1 100644 --- a/loopx/cli_commands/quota_context_reader.py +++ b/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. diff --git a/loopx/control_plane/quota/progressive_context.ts b/loopx/control_plane/quota/progressive_context.ts index d0729e176c..82e9801714 100644 --- a/loopx/control_plane/quota/progressive_context.ts +++ b/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, + 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 { diff --git a/loopx/control_plane/turn_driver/decision_capture.py b/loopx/control_plane/turn_driver/decision_capture.py index 7fa9d0a615..3a5ad206e3 100644 --- a/loopx/control_plane/turn_driver/decision_capture.py +++ b/loopx/control_plane/turn_driver/decision_capture.py @@ -703,10 +703,14 @@ def _first_read_goal_text(payload: dict[str, Any], path: list[Any], value: Any) def _context_selection_bytes(payload: dict[str, Any], selections: list[dict[str, Any]], *, list_fields: bool, list_depth: int | None, - present_goal_narrative: bool) -> tuple[bytes, list[str]]: + present_goal_narrative: bool, + text_append: dict[str, Any] | None = None) -> tuple[bytes, list[str]]: """Render TS-verified addresses from the original, digest-checked values.""" + if text_append is not None and (list_fields or not isinstance(text_append, dict)): + raise ValueError("observed text append requires a selected value") views: list[dict[str, Any]] = [] presented: list[str] = [] + append_rendered = False for item in selections: selected: Any = payload path, pointer = item["path"], item["pointer"] @@ -717,7 +721,16 @@ def _context_selection_bytes(payload: dict[str, Any], selections: list[dict[str, selected = selected[key] else: raise ValueError("selected address does not match the original capture") - if list_fields: + if text_append is not None and pointer == text_append.get("pointer"): + prefix = text_append.get("prefix_bytes") + raw_text = selected.encode("utf-8") if isinstance(selected, str) else b"" + if type(prefix) is not int or not 0 < prefix < len(raw_text): + raise ValueError("observed text append is outside the selected value") + views.append({"pointer": pointer, "value": raw_text[prefix:].decode("utf-8"), + "presentation": "append_to_observed_text", "prefix_bytes": prefix}) + presented.append(pointer) + append_rendered = True + elif list_fields: views.append({"pointer": pointer, "fields": context_index_fields(payload, item["children"])}) else: text = _first_read_goal_text(payload, path, selected) if present_goal_narrative else None @@ -726,6 +739,8 @@ def _context_selection_bytes(payload: dict[str, Any], selections: list[dict[str, else: presented.append(pointer) views.append({"pointer": pointer, "value": text, "presentation": _GOAL_NARRATIVE_PRESENTATION}) + if text_append is not None and not append_rendered: + raise ValueError("observed text append does not match a selected value") view_options = {"list_depth": list_depth} if list_depth is not None and list_depth != 1 else {} raw = json.dumps({"schema_version": "context_capture_index_v1" if list_fields else "context_capture_selection_v1", "selections": views, **view_options}, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") + b"\n" @@ -738,17 +753,18 @@ def context_selection_binding(select: list[str], *, list_fields: bool, list_dept "sha256": sha256, **({"present_goal_narrative": True} if presented else {})} -def initial_context_selection_bytes(payload: dict[str, Any], - selections: list[dict[str, Any]]) -> tuple[bytes, list[str]]: +def initial_context_selection_bytes(payload: dict[str, Any], selections: list[dict[str, Any]], *, + text_append: dict[str, Any] | None = None) -> tuple[bytes, list[str]]: return _context_selection_bytes(payload, selections, list_fields=False, list_depth=None, - present_goal_narrative=True) + present_goal_narrative=True, text_append=text_append) def write_initial_context_read_reference(path: Path, digest: str, payload: dict[str, Any], selections: list[dict[str, Any]], *, observed_read: dict[str, Any] | None = None) -> tuple[Path, bytes]: """Pin the shared rendered selection and return its exact rendered bytes.""" - raw, presented = initial_context_selection_bytes(payload, selections) + text_append = observed_read.get("text_append") if observed_read else None + raw, presented = initial_context_selection_bytes(payload, selections, text_append=text_append) reference = write_context_read_reference(path, digest, selection=context_selection_binding([item["pointer"] for item in selections], list_fields=False, list_depth=None, sha256=hashlib.sha256(raw).hexdigest(), presented=bool(presented)), @@ -781,8 +797,11 @@ def context_selection_view(payload: dict[str, Any], *, select: list[str] | None, expected_sha256: str | None, preserve_selection_order: bool = False, present_goal_narrative: bool | None = None, selection_paths: list[list[str | int]] | None = None, + text_append: dict[str, Any] | None = None, ) -> tuple[dict[str, Any], bytes | None, str | None]: """Render the digest-pinned view from a new or previously resolved plan.""" + if text_append is not None and (initial_read or list_fields or expected_sha256 is None): + raise ValueError("observed text append requires a pinned value selection") if selection_paths is not None: if initial_read or list_fields or list_depth is not None or not preserve_selection_order or expected_sha256 is None: raise ValueError("reader selection paths require a pinned value selection") @@ -803,7 +822,7 @@ def context_selection_view(payload: dict[str, Any], *, select: list[str] | None, def render(plan: dict[str, Any], present: bool) -> tuple[bytes, str, list[str]]: raw, displayed = _context_selection_bytes(payload, plan["selections"], list_fields=list_fields, - list_depth=list_depth, present_goal_narrative=present) + list_depth=list_depth, present_goal_narrative=present, text_append=text_append) return raw, hashlib.sha256(raw).hexdigest(), displayed raw, digest, displayed = render(plan, presentation) diff --git a/loopx/control_plane/turn_driver/native_context_selection.py b/loopx/control_plane/turn_driver/native_context_selection.py index 3ffde1ec39..7e63120714 100644 --- a/loopx/control_plane/turn_driver/native_context_selection.py +++ b/loopx/control_plane/turn_driver/native_context_selection.py @@ -22,16 +22,9 @@ def _terminal_range_seen(state: dict[str, Any], digest: str, total: int) -> bool ) -def verified_context_selection( - state: dict[str, Any], page: dict[str, Any], *, defer_section_digests: bool = False, +def _context_selection_target( + page: dict[str, Any], references: list[dict[str, Any]], ) -> dict[str, Any] | None: - from ..quota.turn_envelope import _turn_envelope_result - - current = state.get("source_recovery") - if not isinstance(current, dict) or page.get("ok") is not True: - return None - references = [entry for key in ("source_recovery", "recovery") - if isinstance(entry := state.get(key), dict)] reference_file = page.get("reference_file") selected = None observation = None @@ -62,6 +55,39 @@ def verified_context_selection( # A whole original-source read covers sections without pretending # it was the separate native required-view delivery. pointers, view_digest = [""], source_digest + return { + "initial_read": initial_read, + "observation": observation, + "path": path, + "pointers": pointers, + "selected": selected, + "selection_paths": selection_paths, + "source_digest": source_digest, + "view_digest": view_digest, + } + + +def verified_context_selection( + state: dict[str, Any], page: dict[str, Any], *, defer_section_digests: bool = False, +) -> dict[str, Any] | None: + from ..quota.turn_envelope import _turn_envelope_result + + current = state.get("source_recovery") + if not isinstance(current, dict) or page.get("ok") is not True: + return None + references = [entry for key in ("source_recovery", "recovery") + if isinstance(entry := state.get(key), dict)] + target = _context_selection_target(page, references) + if target is None: + return None + path = target["path"] + source_digest = target["source_digest"] + pointers = target["pointers"] + view_digest = target["view_digest"] + selected = target["selected"] + observation = target["observation"] + selection_paths = target["selection_paths"] + initial_read = target["initial_read"] physical = Path(path).expanduser().absolute() physical = physical.parent.resolve(strict=True) / physical.name trusted = next((entry for entry in references if entry.get("sha256") == source_digest @@ -79,13 +105,14 @@ def verified_context_selection( _, raw, source = read_context_capture_file(str(physical), source_digest, private=True) actual_digest = source_digest presentation_pointers: set[str] = set() + text_append = observation.get("text_append") if isinstance(observation, dict) else None if initial_read or selected is not None or "selected_pointers" in page: selection, selected_raw, actual_digest = context_selection_view(source, select=pointers, initial_read=initial_read, list_fields=False, list_depth=selected.get("list_depth") if selected else None, expected_sha256=view_digest, preserve_selection_order=selected is not None and not initial_read, present_goal_narrative=selected.get("present_goal_narrative") is True if selected else None, - selection_paths=selection_paths) + selection_paths=selection_paths, text_append=text_append) if selected_raw is None: return None raw = selected_raw @@ -115,6 +142,7 @@ def verified_context_selection( 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=literal_pointers, - consumed_pointers=inherited)["sections"] + consumed_pointers=inherited, + **({"observed_text_append": text_append} if inherited and text_append is not None else {}))["sections"] return {**page, "schema_version": "context_capture_page_v1", "sha256": view_digest, "selected_pointers": pointers, "source_sha256": current["sha256"], "read_sections": sections} diff --git a/loopx/control_plane/turn_driver/native_first_read.py b/loopx/control_plane/turn_driver/native_first_read.py index 8999ce117c..fecd43424a 100644 --- a/loopx/control_plane/turn_driver/native_first_read.py +++ b/loopx/control_plane/turn_driver/native_first_read.py @@ -24,6 +24,8 @@ def observed_first_read_reference(directory: str, target: dict[str, Any], recove opening = ({"path": reference["path"], "content": content, "reused": plan["reused"], "remaining": plan["linked_remaining"]} if "linked_remaining" in plan and "remaining" in initial else None) + if opening is not None and "text_append" in plan: + opening["text_append"] = plan["text_append"] return reference, opening @@ -31,8 +33,14 @@ def observed_remaining_read_reference(directory: str, target: dict[str, Any], re context: dict[str, Any], opening: dict[str, Any], ordinary_remaining: dict[str, Any]) -> Path | None: """Bind the smaller remainder to the original observation and fallback.""" + remaining, reused = opening["remaining"], opening["reused"] + if "text_append" in opening and "text_append" not in remaining: + reused = {pointer: proof for pointer, proof in reused.items() + if pointer != opening["text_append"]["pointer"]} + if not reused: + return None pinned = _pin_observed_selection(directory, target, recovery, context, - {**opening["remaining"], "reused": opening["reused"]}, ordinary_remaining) + {**remaining, "reused": reused}, ordinary_remaining) return Path(pinned[0]["path"]) if pinned is not None else None @@ -42,6 +50,11 @@ def _pin_observed_selection(directory: str, target: dict[str, Any], recovery: di try: path, digest = Path(recovery["path"]), recovery["sha256"] observation = {"directory": directory, "target": target, "fields": plan["reused"]} + if "text_append" in plan: + append = plan["text_append"] + if append["pointer"] not in plan["reused"] or ordinary_plan is None: + return None + observation["text_append"] = append selections = plan.get("selections") if ordinary_plan is not None: ordinary = ordinary_plan["selections"] diff --git a/tests/control_plane/test_context_first_read.py b/tests/control_plane/test_context_first_read.py index 0f0d985f91..71bf46f180 100644 --- a/tests/control_plane/test_context_first_read.py +++ b/tests/control_plane/test_context_first_read.py @@ -2170,6 +2170,10 @@ def read(command): # The later task switch retains the same independent acceptance path. update(ALTERNATIVE_TODO_ID, "Prepare an unrelated result.", "seed") + warm_note = note.rstrip() + if not complete: + warm_note += " Appended after an incomplete observation." + update(TODO_ID, warm_note, "append-after-partial") second = capture("note-warm") _, original = read_context_read_reference(second["recovery"]["reference_file"]) source_path = Path(original["path"]) @@ -2183,7 +2187,7 @@ def read(command): cold_authority = cold_facts[source_pointer + "/authority_read"] assert current_authority["provider_revision"] != cold_authority["provider_revision"] warm, facts, warm_bytes = read(second["first_read_command"]) - assert current_source[source_pointer]["todo"]["note"] == note.rstrip() + assert current_source[source_pointer]["todo"]["note"] == warm_note if revision == "selected_evidence": current_todo = current_source[source_pointer]["todo"] assert current_todo["evidence"] == "result:independent-verification" @@ -2191,22 +2195,69 @@ def read(command): codex_context.codex_context_hook(str(journal), {**event, "hook_event_name": "PostCompact"}) recovered, restored, fallback_bytes = read(second["first_read_command"]) assert pointer not in recovered[0].get("previously_delivered_fields", []) - assert restored[pointer] == note.rstrip() and source_path.read_bytes() == source + assert restored[pointer] == warm_note and source_path.read_bytes() == source assert _spend_run_count(runtime) == 1 deliver(journal, event, recovered) settle("note-warm") - revised = note + "Use the revised acceptance check before writing." + revised = "Revised acceptance check. " + note.rstrip() update(TODO_ID, revised, "revise") update(ALTERNATIVE_TODO_ID, revised, "revise") corrected, current, _ = read(capture("note-corrected")["first_read_command"]) assert current[pointer] == revised and pointer not in corrected[0].get("previously_delivered_fields", []) deliver(journal, event, corrected) settle("note-corrected") + + expected_spends = 3 + if revision == "selected_evidence": + def appended_turn(turn, prefix, suffix): + update(TODO_ID, prefix + suffix, turn) + guard = capture(turn) + pages, _, wire_bytes = read(guard["first_read_command"]) + body = json.loads("".join(page["content"] for page in pages)) + appended = next(row for row in body["selections"] if row["pointer"] == pointer) + assert appended == {"pointer": pointer, "value": suffix, + "presentation": "append_to_observed_text", + "prefix_bytes": len(prefix.encode("utf-8"))} + assert pointer not in pages[0].get("previously_delivered_fields", []) + _, exact, _ = read(guard["recovery"]["command"] + " --select " + pointer) + assert exact[pointer] == prefix + suffix + reference = Path(pages[0]["reference_file"]) + _, metadata = read_context_read_reference(str(reference)) + fallback_name = metadata["observed_read"]["fallback_reference"]["filename"] + fallback_command = guard["first_read_command"].replace( + shlex.quote(str(reference)), shlex.quote(str(reference.with_name(fallback_name)))) + _, _, full_bytes = read(fallback_command) + assert full_bytes - wire_bytes > 10_000 + print(json.dumps({"provider": provider, "prior_read": observation, "turn": turn, + "full_bytes": full_bytes, "append_bytes": wire_bytes})) + return guard, pages + + _, appended = appended_turn("note-append", revised, " 新增决策 🧪，沿用前面的验收。") + note_proof = codex_context._state(journal)["consumed_pointers"][pointer] + assert len(appended) > 1 + deliver(journal, event, appended[-1:]) + assert codex_context._state(journal)["consumed_pointers"][pointer] == note_proof + deliver(journal, event, appended[:-1]) + assert codex_context._state(journal)["consumed_pointers"][pointer] != note_proof + settle("note-append") + + revised += " 新增决策 🧪，沿用前面的验收。" + second_suffix = " Second addition with new evidence." + chained, _ = appended_turn("note-append-next", revised, second_suffix) + revised += second_suffix + codex_context.codex_context_hook(str(journal), {**event, "hook_event_name": "PostCompact"}) + fallback, exact, _ = read(chained["first_read_command"]) + assert exact[pointer] == revised + assert pointer not in fallback[0].get("previously_delivered_fields", []) + settle("note-append-next") + update(ALTERNATIVE_TODO_ID, revised, "copy-final-note") + expected_spends += 2 + alternate, other, _ = read(capture("note-alternate", ALTERNATIVE_TODO_ID)["first_read_command"]) assert other[pointer] == revised and other["/selected_todo"]["todo_id"] == ALTERNATIVE_TODO_ID assert pointer not in alternate[0].get("previously_delivered_fields", []) - assert _heartbeat_receipt_count(runtime, "note-warm") == 1 and _spend_run_count(runtime) == 3 + assert _heartbeat_receipt_count(runtime, "note-warm") == 1 and _spend_run_count(runtime) == expected_spends print(json.dumps({"provider": provider, "revision": revision, "observation": observation, "warm_bytes": warm_bytes, "recovery_bytes": fallback_bytes, @@ -2224,4 +2275,4 @@ def read(command): assert fallback_bytes - warm_bytes > 10_000 else: assert pointer not in warm[0].get("previously_delivered_fields", []) - assert facts[pointer] == restored[pointer] == note.rstrip() + assert facts[pointer] == restored[pointer] == warm_note diff --git a/tests/control_plane_ts/progressive_context.test.ts b/tests/control_plane_ts/progressive_context.test.ts index cb63c9d2aa..d37bdfe40a 100644 --- a/tests/control_plane_ts/progressive_context.test.ts +++ b/tests/control_plane_ts/progressive_context.test.ts @@ -1,7 +1,8 @@ import assert from "node:assert/strict"; import test from "node:test"; import {visionAuthoringContract} from "../../loopx/control_plane/goals/vision_checkpoint.ts"; -import {initialContextReadPointers, progressiveContextPlan} from "../../loopx/control_plane/quota/progressive_context.ts"; +import {contextReadSectionDigests, initialContextReadPointers, + progressiveContextPlan} from "../../loopx/control_plane/quota/progressive_context.ts"; import {FULL_OUTCOME_READ_GUIDANCE, PROGRESSIVE_OUTCOME_READ_GUIDANCE} from "../../loopx/control_plane/work_items/interaction_contract.ts"; const outcome = {id: "o", objective: "Keep all requirements.", revision: 1, owner_agent_id: "a", todo_ids: ["t"], @@ -97,6 +98,48 @@ test("unknown and failed source semantics remain whole rather than becoming gues assert.deepEqual(plan.deferred, []); }); +test("selected Todo append proofs require an exact current-source prefix", () => { + const note = "Keep the complete acceptance evidence. ".repeat(40); + const context = (value: string, source = "provider") => ({ + goal_id: "goal", registry: "registry", runtime_root: "runtime", + selected_todo: {todo_id: "todo"}, + interaction_contract: {agent_channel: {work_context: { + selected_todo_text_verified: true, + sources: [{source: "selected_todo", content: {ok: true, matched: true, ambiguous: false, + source, todo: {todo_id: "todo", note: value}, + authority_read: {decision_read_from_provider: true, legacy_fallback_used: false, + source_authority: "provider"}}}]}}}, + }); + const path = ["interaction_contract", "agent_channel", "work_context", "sources", 0, + "content", "todo", "note"]; + const address = "/interaction_contract/agent_channel/work_context/sources/0/content/todo/note"; + const section = (value: string, consumed?: string, append?: {pointer: string; prefix_bytes: number}, + source = "provider") => contextReadSectionDigests({context: context(value, source), + ...(consumed === undefined ? {} : {consumed_pointers: {[address]: consumed}}), + ...(append === undefined ? {} : {observed_text_append: append})}, + consumed === undefined ? [path] : [])[address]; + + const original = section(note)!; + assert.match(original, /^\d+:[a-f0-9]{64}$/); + assert.equal(original.split(":")[0], String(Buffer.byteLength(note))); + const suffix = "Appended requirement."; + const appended = note + suffix; + const current = section(appended, original, + {pointer: address, prefix_bytes: Buffer.byteLength(note)}); + assert.match(current!, /^\d+:[a-f0-9]{64}$/); + assert.deepEqual(section(appended, original, + {pointer: address, prefix_bytes: Buffer.byteLength(note) - 1}), undefined); + assert.equal(section(`Replacement. ${note}`, original, + {pointer: address, prefix_bytes: Buffer.byteLength(note)}), undefined); + assert.equal(section(appended, original, + {pointer: address, prefix_bytes: Buffer.byteLength(note)}, "other-provider"), undefined); + + const legacy = original.slice(original.indexOf(":") + 1); + assert.match(section(note, legacy)!, /^\d+:[a-f0-9]{64}$/); + assert.equal(section(appended, legacy, + {pointer: address, prefix_bytes: Buffer.byteLength(note)}), undefined); +}); + test("exact repeated Todo rows preserve membership without collapsing distinct evidence or revisions", () => { const row = {schema_version: "todo_item_v0", todo_id: "same-title", text: "Shared title.", note: "Keep this complete final constraint. ".repeat(20), content_revision: "revision:one",