diff --git a/docs/guides/context-delivery.md b/docs/guides/context-delivery.md index 6b09d061f5..93934c0632 100644 --- a/docs/guides/context-delivery.md +++ b/docs/guides/context-delivery.md @@ -129,6 +129,11 @@ from file deltas and provider caching; it does not prove model retention or lower total cost. Partial reads and lost continuity cannot qualify this reuse. Explicit same-capture selections always recover the complete current value. +An unchanged Goal narrative can also be referenced after an exact read of the +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. + ### Read and recover the right object Use the path, digest and command returned by the current response. A native diff --git a/docs/guides/context-delivery.zh-CN.md b/docs/guides/context-delivery.zh-CN.md index fcfabbe2b7..757807833c 100644 --- a/docs/guides/context-delivery.zh-CN.md +++ b/docs/guides/context-delivery.zh-CN.md @@ -102,6 +102,10 @@ Full 表示完整保存视图可用，不要求判断任何问题前先遍历所 这种工具投递假设与文件差分、provider 缓存分别成立，不证明模型保留或完整成本下降。 未读完或连续性丢失不能取得该复用资格；显式同 capture 选择仍可恢复完整当前值。 +当前原生会话在同一上下文周期内完整读取同一注册来源后，也可引用未变的 Goal 正文。 +首读仍交付当前来源修订和 Turn 控制条件。正文或来源变化、读取不完整或发生压缩时， +读者恢复当前正文。 + ### 读对对象，再恢复 使用当前响应返回的路径、摘要与命令。原生响应的 `recovery` 指向当前完整必要视图； diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py index 4bf1e7449a..635197f099 100644 --- a/loopx/cli_commands/quota_capture.py +++ b/loopx/cli_commands/quota_capture.py @@ -143,6 +143,17 @@ def _validate_context_page_window(offset: int, size: int, output_limit: int | No "larger pages require a sufficient --max-page-output-bytes within the capture limit.") +def _validate_pinned_selection(selection: dict[str, Any]) -> None: + if (set(selection) - {"present_goal_narrative"} != {"select", "initial_read", "list_fields", "list_depth", "sha256"} + or type(selection["initial_read"]) is not bool or type(selection["list_fields"]) is not bool + or ("present_goal_narrative" in selection and ( + selection["present_goal_narrative"] is not True + or selection["list_fields"] or selection["initial_read"])) + or not isinstance(selection["sha256"], str) + or BARE_SHA256_PATTERN.fullmatch(selection["sha256"]) is None): + raise ValueError("reader selection reference is incomplete or unsupported") + + def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, Any] | None = None, next_page_bytes: int | None = None, next_page_output_bytes: int | None = None, @@ -204,30 +215,31 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An 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") - if (set(pinned_selection) - {"present_goal_narrative"} != - {"select", "initial_read", "list_fields", "list_depth", "sha256"} - or type(pinned_selection["initial_read"]) is not bool or type(pinned_selection["list_fields"]) is not bool - or ("present_goal_narrative" in pinned_selection and ( - pinned_selection["present_goal_narrative"] is not True - or pinned_selection["list_fields"] or pinned_selection["initial_read"])) - or not isinstance(pinned_selection["sha256"], str) - or BARE_SHA256_PATTERN.fullmatch(pinned_selection["sha256"]) is None): - raise ValueError("reader selection reference is incomplete or unsupported") + _validate_pinned_selection(pinned_selection) if observation := reference.get("observed_read"): - from ..control_plane.turn_driver.native_first_read import observation_current + from ..control_plane.turn_driver.native_first_read import observation_current, observed_first_read_fallback if not observation_current(observation, path, digest): + fallback = observed_first_read_fallback(observation, path, digest) + if fallback is not None: + try: + _validate_pinned_selection(fallback[1]) + except ValueError: + fallback = None if offset: return {"ok": False, "reason": "Observed first-read scope changed during paging.", - "recovery_command": context_capture_read_command(path, digest, initial_read=True, + "recovery_command": context_capture_read_command(path, digest, initial_read=fallback is None, + reference_file=fallback[0] if fallback is not None else None, page_bytes=size, max_page_output_bytes=output_limit, reader_argv=reader_argv)} # Restart in the ordinary complete view of this same capture. - # Subsequent pages must pin that view even if the journal changes. - pinned_selection, initial_read = None, True - try: - reference_file = write_context_read_reference(path, digest) - except OSError: - reference_file = None + if fallback is not None: + reference_file, pinned_selection = fallback + else: + pinned_selection, initial_read = None, True + try: + reference_file = write_context_read_reference(path, digest) + except OSError: + reference_file = None elif offset == 0: presentation = {**(presentation or {}), "previously_delivered_fields": list(observation["fields"]), "field_recovery": "These unchanged fields still apply. If needed or uncertain, append --select " diff --git a/loopx/control_plane/quota/progressive_context.ts b/loopx/control_plane/quota/progressive_context.ts index c0479ab174..943cbd4aee 100644 --- a/loopx/control_plane/quota/progressive_context.ts +++ b/loopx/control_plane/quota/progressive_context.ts @@ -3,6 +3,7 @@ import {EffectiveAction} from "./effective_action.generated.ts"; * addressed in the immutable source. This plan changes no canonical state. */ import {isDeepStrictEqual} from "node:util"; import {visionAuthoringContract} from "../goals/vision_checkpoint.ts"; +import {completeGoalNarrative} from "../work_items/goal_context_source.ts"; import type {JsonObject} from "../effect_program.ts"; import {jsonObject, requireBoolean, requireJsonObject} from "../runtime_decode.ts"; import {GOAL_OUTCOMES_SCHEMA, OUTCOME_CONTEXT_NAVIGATION_SCHEMA} from "../goals/outcome_requirements.ts"; @@ -17,6 +18,7 @@ import {createHash} from "node:crypto"; type Path = (string | number)[]; +type ReusableReadField = {path: Path; freshKeys: string[]; minimumBytes: number; sourceFile?: string}; const MIN_REUSABLE_BODY_BYTES = 1024; const pointer = (path: Path): string => path.length ? "/" + path.map(key => String(key).replaceAll("~", "~0").replaceAll("/", "~1")).join("/") : ""; @@ -26,14 +28,21 @@ const known = (row: JsonObject, fields: string[]): boolean => Object.keys(row).e /** 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): {path: Path; freshKeys: string[]; minimumBytes: number}[] { +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((value, i) => { + return sources.flatMap((value, i) => { const source = jsonObject(value), content = jsonObject(source?.content); if (!content) return []; const path = [...base, i, "content"]; + // The current source and full-file revision still reach the first read. + // Only the literal narrative may carry forward across generated Todo edits. + if (source?.source === "goal_state" && content.ok !== false && + (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}]; if (source?.kind === "agent_preferences") return [{path, freshKeys: ["ok", "revision", "instruction"], minimumBytes: MIN_REUSABLE_BODY_BYTES}]; if (source?.source !== "goal_acceptance") return []; @@ -66,15 +75,22 @@ function reusableFirstReadFields(context: JsonObject): {path: Path; freshKeys: s }); } +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]); + return createHash("sha256").update(value).digest("hex"); +} + /** 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[]): JsonObject { const context = requireJsonObject(request.context, "read section source"); const identity = contextSourceIdentity(request, request.source_number_tokens); const selected = new Set(paths.map(pointer)); + const fields = new Map(reusableFirstReadFields(context).map(field => [pointer(field.path), field])); const covered = new Map(paths.map(path => [pointer(path), {path, minimumBytes: MIN_REUSABLE_BODY_BYTES}])); - for (const field of reusableFirstReadFields(context)) - if (paths.some(parent => pathStartsWith(field.path, parent))) covered.set(pointer(field.path), field); + for (const [address, field] of fields) + if (paths.some(parent => pathStartsWith(field.path, parent))) covered.set(address, field); const inherited = jsonObject(request.consumed_pointers), coverage = new Set(paths.map(pointer)); // The IO owner authenticates the reader's native observation. Recheck each // inherited value against this exact source before combining it with the @@ -86,7 +102,7 @@ export function contextReadSectionDigests(request: JsonObject, paths: Path[]): J 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 (createHash("sha256").update(identity(at(context, path), ["context", ...path])).digest("hex") === expected) + if (observedSectionDigest(identity(at(context, path), ["context", ...path]), fields.get(address)) === expected) coverage.add(address); } const complete = (path: Path): boolean => { @@ -95,18 +111,18 @@ export function contextReadSectionDigests(request: JsonObject, paths: Path[]): J return value !== null && typeof value === "object" && Object.keys(value).length > 0 && Object.keys(value).every(key => complete([...path, Array.isArray(value) ? Number(key) : key])); }; - for (const field of reusableFirstReadFields(context)) - if (complete(field.path)) covered.set(pointer(field.path), field); + for (const [address, field] of fields) + if (complete(field.path)) covered.set(address, field); } return Object.fromEntries([...covered].flatMap(([address, {path, minimumBytes}]) => { const content = identity(at(context, path), ["context", ...path]); // Parent delivery must not multiply tiny bodies into a larger journal. return selected.has(address) || Buffer.byteLength(content) >= minimumBytes - ? [[address, createHash("sha256").update(content).digest("hex")]] : []; + ? [[address, observedSectionDigest(content, fields.get(address))]] : []; })); } -/** Only previously delivered supplementary bodies may use a reference. Current +/** Only registered, previously delivered bodies may use a reference. Current * decisions, stops, selection, settlement and unknown source shapes stay inline. */ function observedFirstRead(request: JsonObject, pointers: string[]): JsonObject | undefined { const context = requireJsonObject(request.context, "current first read"); @@ -126,7 +142,7 @@ function observedFirstRead(request: JsonObject, pointers: string[]): JsonObject const body = at(context, path); if (typeof seen[address] === "string" && field) { const content = identity(body, ["context", ...path]); - if (Buffer.byteLength(content) >= field.minimumBytes && createHash("sha256").update(content).digest("hex") === seen[address]) { + if (Buffer.byteLength(content) >= field.minimumBytes && observedSectionDigest(content, field) === seen[address]) { 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]})); @@ -148,8 +164,22 @@ function observedFirstRead(request: JsonObject, pointers: string[]): JsonObject 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. - return bytes(grouped) + Buffer.byteLength(JSON.stringify(Object.keys(reused))) + MIN_REUSABLE_BODY_BYTES < bytes(original) - ? {selections: grouped, reused} : undefined; + 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--; + } + 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}; + } + return {selections: grouped, reused}; } function inlineReadSections(request: JsonObject, required: JsonObject): JsonObject { diff --git a/loopx/control_plane/turn_driver/codex_context.py b/loopx/control_plane/turn_driver/codex_context.py index 72abed39c0..9d1796d926 100644 --- a/loopx/control_plane/turn_driver/codex_context.py +++ b/loopx/control_plane/turn_driver/codex_context.py @@ -785,7 +785,7 @@ def publish_recovery(reference: dict[str, Any]) -> None: from .native_first_read import observed_first_read_reference reference = observed_first_read_reference(directory, result["target"], recovery, - transaction["context"], plan) + transaction["context"], plan, transaction.get("initial_read_plan")) if reference is not None: state["pending"]["observed_first_read"] = reference _save(root, state) diff --git a/loopx/control_plane/turn_driver/native_first_read.py b/loopx/control_plane/turn_driver/native_first_read.py index 9e805f6f70..1f0c646a5e 100644 --- a/loopx/control_plane/turn_driver/native_first_read.py +++ b/loopx/control_plane/turn_driver/native_first_read.py @@ -1,25 +1,63 @@ """Pin a smaller initial selection to previously observed native tool output.""" from __future__ import annotations +import hashlib import os from pathlib import Path from typing import Any -from .decision_capture import write_initial_context_read_reference +from .decision_capture import read_context_read_reference, write_initial_context_read_reference def observed_first_read_reference(directory: str, target: dict[str, Any], recovery: dict[str, Any], - context: dict[str, Any], plan: dict[str, Any]) -> dict[str, Any] | None: - """Write the typed delivery plan once. The host renders it from the immutable source.""" + context: dict[str, Any], plan: dict[str, Any], + initial_read_plan: dict[str, Any] | None = None) -> dict[str, Any] | None: + """Pin the reduced view and its ordinary fallback from the same capture.""" try: - reference, _ = write_initial_context_read_reference(Path(recovery["path"]), recovery["sha256"], - context, plan["selections"], - observed_read={"directory": directory, "target": target, "fields": plan["reused"]}) + path, digest = Path(recovery["path"]), recovery["sha256"] + observation = {"directory": directory, "target": target, "fields": plan["reused"]} + 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): + return None + selections = [*ordinary[:start], *splice["selections"], *ordinary[end:]] + fallback, body = write_initial_context_read_reference(path, digest, context, ordinary) + observation["fallback_reference"] = {"filename": fallback.name, + "selection_sha256": hashlib.sha256(body).hexdigest()} + if selections is None: + return None + reference, _ = write_initial_context_read_reference(path, digest, context, selections, + observed_read=observation) return {"path": str(reference), "pointer_bytes": sum(len(key.encode()) + 4 for key in plan["reused"])} - except (OSError, ValueError): + except (OSError, ValueError, KeyError, TypeError): return None # The ordinary first read remains available. +def observed_first_read_fallback(observation: dict[str, Any], path: Path, digest: str) -> tuple[Path, dict[str, Any]] | None: + """Locate the ordinary pinned selection of this exact capture.""" + fallback = observation.get("fallback_reference") if isinstance(observation, dict) else None + if not isinstance(fallback, dict): + return None + filename = fallback.get("filename") + if not isinstance(filename, str) or not filename.startswith(path.stem + ".") or not filename.endswith(".reader.json"): + return None + try: + reference, metadata = read_context_read_reference(str(path.with_name(filename))) + selection = metadata.get("selection") + source = Path(metadata["path"]).expanduser().absolute() + if (metadata["sha256"] == digest and source.parent.resolve(strict=True) / source.name == path + and "observed_read" not in metadata and isinstance(selection, dict) + and isinstance(fallback.get("selection_sha256"), str) + and selection.get("sha256") == fallback["selection_sha256"]): + return reference, selection + except (OSError, ValueError): + pass + return None + + def observation_current(observation: dict[str, Any], path: Path, digest: str, *, hook_session: str | None = None) -> bool: from .codex_context import _private_root, _state diff --git a/tests/control_plane/test_context_first_read.py b/tests/control_plane/test_context_first_read.py index 79c617830e..4987601630 100644 --- a/tests/control_plane/test_context_first_read.py +++ b/tests/control_plane/test_context_first_read.py @@ -1049,3 +1049,87 @@ def facts(pages): assert facts(restarted)[prefix + "/0"] == outcomes[0] recovered = pages(second["first_read_command"]) assert facts(recovered)[prefix + "/0"] == outcomes[0] + + +@pytest.mark.parametrize("transition", ["unchanged", "corrected", "different_source", "incomplete"]) +def test_native_first_read_goal_narrative_reuse(tmp_path, monkeypatch, decision, transition): + from test_codex_context_lifecycle import deliver, setup + + event = setup(tmp_path, monkeypatch) + narrative = "保留完整目标、独立验收、权限边界和停止条件。\n" * 140 + source = decision["interaction_contract"]["agent_channel"]["work_context"]["sources"][0] + plan = decision["interaction_contract"]["cli_channel"]["settlement_plan"] + prefix = "/interaction_contract/agent_channel/work_context/sources/0/content" + body_pointer = prefix + "/text" + source.update(source="goal_state", ordering="before_work") + + def goal_source(path, text, generated): + def revision(value): + return "sha256:" + hashlib.sha256(value.encode()).hexdigest() + + line = len(text.splitlines()) + 1 + return {"state_file": path, "text": text, "revision": revision(text + generated), + "excluded_todo_regions": [{"role": "agent", "start_line": line, "end_line": line, + "revision": revision(generated)}]} + + def capture(name, content): + source.update(content=content, command=f"cat -- {content['state_file']}") + plan["identity"]["turn_instance_id"] = name + directory = tmp_path / name + 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=name) + return model_context_reference(native_context_capture_view(prepared.delivery, directory, + context_directory=str(tmp_path), initial_read_plan=prepared.initial_read_plan)) + + def read(command): + pages, size, offset = [], 0, 0 + while command: + words = shlex.split(command) + stdout = subprocess.check_output([sys.executable, "-m", "loopx.entrypoint", "--format", "json", + *words[words.index("quota"):]]) + page = json.loads(stdout) + assert page["ok"] and page["offset"] == offset, page + pages.append(page) + size += len(stdout) + offset += len(page["content"].encode()) + command = page.get("next_command") + assert pages[-1]["eof"] and offset == pages[-1]["total_bytes"] + return pages, selected_context_values("".join(page["content"] for page in pages)), size + + first = capture("first", goal_source("goal.md", narrative, "old generated todo\n")) + pages, facts, _ = read(first["first_read_command"]) + assert facts[body_pointer] == narrative and "previously_delivered_fields" not in pages[0] + deliver(tmp_path, event, pages) + assert body_pointer in codex_context._state(tmp_path)["consumed_pointers"] + + text = narrative + "Use the revised owner decision.\n" if transition == "corrected" else narrative + current = goal_source("alternate.md" if transition == "different_source" else "goal.md", + text, "new generated todo\n") + if transition == "incomplete": + current.update(read_complete=False, recovery="Restore the complete Goal source.") + decision["goal_boundary"]["stop_condition"] = "Stop before an external write." + second = capture("second", current) + pages, facts, warm_bytes = read(second["first_read_command"]) + assert facts[prefix + "/revision"] == current["revision"] + assert facts["/goal_boundary"] == decision["goal_boundary"] + assert facts["/interaction_contract/cli_channel/settlement_plan/identity/turn_instance_id"] == "second" + if transition != "unchanged": + assert facts[body_pointer] == text and "previously_delivered_fields" not in pages[0] + if transition == "incomplete": + assert facts[prefix + "/recovery"] == current["recovery"] + return + + assert pages[0]["previously_delivered_fields"] == [body_pointer] and body_pointer not in facts + observed = json.loads(Path(pages[0]["reference_file"]).read_text())["observed_read"] + ordinary = Path(pages[0]["reference_file"]).with_name(observed["fallback_reference"]["filename"]) + recovery = second["recovery"]["command"] + _, restored, _ = read(recovery + " --select " + body_pointer) + assert restored[body_pointer] == narrative + codex_context.codex_context_hook(str(tmp_path), {**event, "hook_event_name": "PostCompact"}) + fallback, after, fallback_bytes = read(second["first_read_command"]) + assert "previously_delivered_fields" not in fallback[0] and after[body_pointer] == narrative + 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