diff --git a/docs/guides/context-delivery.md b/docs/guides/context-delivery.md index b7fc90c7ec..8dc92eaf1d 100644 --- a/docs/guides/context-delivery.md +++ b/docs/guides/context-delivery.md @@ -109,6 +109,18 @@ the model has retained it. Retrieve necessary current facts when uncertain. Default-off and signed-envelope callers retain their own read contracts; the progressive native example does not relax them. +In a continuing native session, a progressive first read can reference exact +criteria or requirement descriptions delivered by an earlier complete read. An appended +criterion, revised review or supporting-material projection need not replay +every unchanged description. Current revisions, changed descriptions, review +states and new controls invalidate their containing criterion and remain visible. +An exact unchanged criterion needs no metadata split. Splitting a container adds addresses +and recovery guidance, so the reader keeps the complete presentation when the +split would not save enough bytes. This tool-delivery assumption is separate +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. + ### 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 7982e5c7bf..ce392026af 100644 --- a/docs/guides/context-delivery.zh-CN.md +++ b/docs/guides/context-delivery.zh-CN.md @@ -89,6 +89,13 @@ Full 表示完整保存视图可用，不要求判断任何问题前先遍历所 消失，也不证明模型保留了它。不确定时取得必要当前事实。Default-off 与签名 envelope 调用方继续遵守各自读取契约；这个渐进原生例子不放宽它们。 +原生会话持续期间，渐进首读可以引用此前完整读取实际送达的精确条件或要求正文。新增条件、 +修订评审或支持材料投影不必让每条未变正文重送；当前修订号、已改正文、评审状态和 +新控制条件使对应整条条件失效并重新显示；精确未变的整条条件无需拆开元数据。 +拆开容器会增加地址及恢复说明，节省不足时 reader 保留完整展现。 +这种工具投递假设与文件差分、provider 缓存分别成立，不证明模型保留或完整成本下降。 +未读完或连续性丢失不能取得该复用资格；显式同 capture 选择仍可恢复完整当前值。 + ### 读对对象，再恢复 使用当前响应返回的路径、摘要与命令。原生响应的 `recovery` 指向当前完整必要视图； diff --git a/loopx/control_plane/quota/progressive_context.ts b/loopx/control_plane/quota/progressive_context.ts index 6b6e521dd3..410efa97cb 100644 --- a/loopx/control_plane/quota/progressive_context.ts +++ b/loopx/control_plane/quota/progressive_context.ts @@ -25,7 +25,7 @@ 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[]}[] { +function reusableFirstReadFields(context: JsonObject): {path: Path; freshKeys: string[]; minimumBytes: number}[] { const base: Path = ["interaction_contract", "agent_channel", "work_context", "sources"]; const sources = at(context, base); if (!Array.isArray(sources)) return []; @@ -34,13 +34,34 @@ function reusableFirstReadFields(context: JsonObject): {path: Path; freshKeys: s if (!content) return []; const path = [...base, i, "content"]; if (source?.kind === "agent_preferences") - return [{path, freshKeys: ["ok", "revision", "instruction"]}]; + return [{path, freshKeys: ["ok", "revision", "instruction"], minimumBytes: MIN_REUSABLE_BODY_BYTES}]; if (source?.source !== "goal_acceptance") return []; const outcomesPath = [...path, "goal_acceptance_contract", "outcome_requirements", "outcomes"]; const outcomes = at(context, outcomesPath); - return Array.isArray(outcomes) ? outcomes.flatMap((outcome, index) => - typeof jsonObject(outcome)?.id === "string" - ? [{path: [...outcomesPath, index], freshKeys: ["id", "revision"]}] : []) : []; + return Array.isArray(outcomes) ? outcomes.flatMap((value, index) => { + const outcome = jsonObject(value), path = [...outcomesPath, index]; + if (typeof outcome?.id !== "string") return []; + const fields = [{path, freshKeys: ["id", "revision"], minimumBytes: MIN_REUSABLE_BODY_BYTES}]; + // An appended criterion, revised review or supporting-material projection + // changes the outcome container, not every unchanged requirement body. + // Keep all other fields current, including unknown controls and order. + if (Array.isArray(outcome.requirements)) { + fields.push({path: [...path, "requirements"], freshKeys: [], minimumBytes: MIN_REUSABLE_BODY_BYTES}); + outcome.requirements.forEach((value, i) => { + const requirement = jsonObject(value), body = [...path, "requirements", i, "description"]; + if (typeof requirement?.id === "string" && typeof requirement.description === "string") { + const record = body.slice(0, -1); + // Prefer an exact complete criterion before splitting its metadata. + // Tiny bodies do not justify a digest and long-address journal row. + // Changed status, review or unknown controls invalidate the record; + // a large unchanged description can still be considered alone. + fields.push({path: record, freshKeys: ["id"], minimumBytes: MIN_REUSABLE_BODY_BYTES}); + fields.push({path: body, freshKeys: [], minimumBytes: MIN_REUSABLE_BODY_BYTES}); + } + }); + } + return fields; + }) : []; }); } @@ -50,13 +71,36 @@ export function contextReadSectionDigests(request: JsonObject, paths: Path[]): J const context = requireJsonObject(request.context, "read section source"); const identity = contextSourceIdentity(request, request.source_number_tokens); const selected = new Set(paths.map(pointer)); - const covered = new Map(paths.map(path => [pointer(path), path])); - for (const {path} of reusableFirstReadFields(context)) - if (paths.some(parent => pathStartsWith(path, parent))) covered.set(pointer(path), path); - return Object.fromEntries([...covered].flatMap(([address, path]) => { + 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); + 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 + // current literal selections. This is section coverage, never a full-input + // baseline. Without it, a partial read strands the next Turn on small fields + // even after every part of the revised outcome has actually been delivered. + if (inherited && Object.keys(inherited).length) { + for (const [address, expected] of Object.entries(inherited)) { + 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) + coverage.add(address); + } + const complete = (path: Path): boolean => { + if (path.map((_, i) => pointer(path.slice(0, i + 1))).concat([""]).some(parent => coverage.has(parent))) return true; + const value = at(context, path); + 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); + } + 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) >= MIN_REUSABLE_BODY_BYTES + return selected.has(address) || Buffer.byteLength(content) >= minimumBytes ? [[address, createHash("sha256").update(content).digest("hex")]] : []; })); } @@ -68,28 +112,43 @@ function observedFirstRead(request: JsonObject, pointers: string[]): JsonObject const seen = requireJsonObject(request.consumed_pointers, "observed sections"); if (typeof jsonObject(context.selected_todo)?.todo_id !== "string") return undefined; const fields = new Map(reusableFirstReadFields(context).map(field => [pointer(field.path), field])); + const parents = new Set([...fields.values()].flatMap(({path}) => path.map((_, i) => pointer(path.slice(0, i))))); const identity = contextSourceIdentity(request, request.source_number_tokens); const reused: JsonObject = {}; // Match semantic fields before the reader groups complete parents. Otherwise // selecting every outcome hides the individual unchanged bodies from reuse. const selections = capturedContextSelection({context, select: pointers, preserve_selection_order: true}).selections as {pointer: string; path: Path}[]; - const current = selections.flatMap(row => { + const visit = (row: {pointer: string; path: Path}): typeof selections => { const {path, pointer: address} = row; const field = fields.get(address); - if (typeof seen[address] !== "string" || !field) return [row]; - const body = jsonObject(at(context, path)); - if (!body) return [row]; - const content = identity(body, ["context", ...path]); - // A reference and its recovery instructions should cost less than the field. - if (Buffer.byteLength(content) < MIN_REUSABLE_BODY_BYTES || createHash("sha256").update(content).digest("hex") !== seen[address]) - return [row]; - reused[address] = seen[address]; - return field.freshKeys.filter(key => - Object.hasOwn(body, key)).map(key => ({pointer: address + "/" + key, path: [...path, key]})); - }); - return Object.keys(reused).length ? {selections: capturedContextSelection({context, initial_read: true}, - current.map(row => row.pointer)).selections, reused} : undefined; + 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]) { + 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]})); + } + } + if (!parents.has(address) || body === null || typeof body !== "object") return [row]; + const before = Object.keys(reused).length; + const children = Object.keys(body).flatMap(key => { + const child = [...path, Array.isArray(body) ? Number(key) : key]; + return visit({pointer: pointer(child), path: child}); + }); + return Object.keys(reused).length > before ? children : [row]; + }; + const current = selections.flatMap(visit); + if (!Object.keys(reused).length) return undefined; + const grouped = capturedContextSelection({context, initial_read: true}, current.map(row => row.pointer)).selections as typeof selections; + 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. + return bytes(grouped) + Buffer.byteLength(JSON.stringify(Object.keys(reused))) + MIN_REUSABLE_BODY_BYTES < bytes(original) + ? {selections: grouped, reused} : undefined; } function inlineReadSections(request: JsonObject, required: JsonObject): JsonObject { diff --git a/loopx/control_plane/turn_driver/native_context_selection.py b/loopx/control_plane/turn_driver/native_context_selection.py index 4291766770..1201648968 100644 --- a/loopx/control_plane/turn_driver/native_context_selection.py +++ b/loopx/control_plane/turn_driver/native_context_selection.py @@ -23,9 +23,11 @@ def verified_context_selection(state: dict[str, Any], page: dict[str, Any]) -> d if isinstance(entry := state.get(key), dict)] reference_file = page.get("reference_file") selected = None + observation = None if isinstance(reference_file, str): _, metadata = read_context_read_reference(reference_file) selected = metadata.get("selection") + observation = metadata.get("observed_read") if not isinstance(selected, dict) or selected.get("list_fields") is not False: return None path, source_digest = metadata["path"], metadata["sha256"] @@ -73,6 +75,15 @@ def verified_context_selection(state: dict[str, Any], page: dict[str, Any]) -> d or (page.get("eof") is False and (end >= total or page.get("next_offset") != end)) or type(page.get("eof")) is not bool): return None - sections = _turn_envelope_result("context_read_sections", context=source, select=pointers)["sections"] + inherited = {} + if isinstance(observation, dict): + from .native_first_read import observation_current + + # The hook already matched its event session/workspace to this journal. + # Native hook processes need not inherit the shell's CODEX_THREAD_ID. + 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=pointers, + consumed_pointers=inherited)["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 a24b2421eb..12fa1d2a10 100644 --- a/loopx/control_plane/turn_driver/native_first_read.py +++ b/loopx/control_plane/turn_driver/native_first_read.py @@ -24,7 +24,8 @@ def observed_first_read_reference(directory: str, target: dict[str, Any], recove return None # The ordinary first read remains available. -def observation_current(observation: dict[str, Any], path: Path, digest: str) -> bool: +def observation_current(observation: dict[str, Any], path: Path, digest: str, *, + hook_session: str | None = None) -> bool: from .codex_context import _private_root, _state try: @@ -32,7 +33,7 @@ def observation_current(observation: dict[str, Any], path: Path, digest: str) -> recovery = state["recovery"] target = observation["target"] return (target == recovery.get("target") and target["context_epoch"] == state["context_epoch"] - and target["session_id"] == state["session_id"] == os.environ.get("CODEX_THREAD_ID") + and target["session_id"] == state["session_id"] == (hook_session or os.environ.get("CODEX_THREAD_ID")) and recovery["sha256"] == digest and Path(recovery["path"]).resolve() == path.resolve() and bool(observation["fields"]) and all(state.get("consumed_pointers", {}).get(pointer) == expected for pointer, expected in observation["fields"].items())) diff --git a/tests/control_plane/test_context_first_read.py b/tests/control_plane/test_context_first_read.py index 3c3d76a1b0..0530ec8909 100644 --- a/tests/control_plane/test_context_first_read.py +++ b/tests/control_plane/test_context_first_read.py @@ -815,6 +815,107 @@ def test_warm_delta_initial_read_uses_current_action_without_the_old_base(tmp_pa assert not registry.exists() and not runtime.exists() +@pytest.mark.parametrize("invalidate,retained", [("none", 7), ("compact", 7), ("partial", 7), ("off", 7), ("none", 1)]) +def test_native_first_read_reuses_requirement_bodies_after_canonical_correction( + tmp_path, monkeypatch, navigated_decision, invalidate, retained): + from test_codex_context_lifecycle import setup, deliver + + event = setup(tmp_path, monkeypatch) + source = navigated_decision["interaction_contract"]["agent_channel"]["work_context"]["sources"][0] + content = source["content"] + outcome = content["goal_acceptance_contract"]["outcome_requirements"]["outcomes"][0] + outcome.update(objective="Preserve cumulative acceptance.", sources=[], requirements=[ + {"id": f"criterion-{i}", "description": (f"Criterion {i}: preserve original requirements, cancellation, source and independent validation. " * 12), + "status": "unverified", "planning_satisfied": False} for i in range(8)]) + prefix = "/interaction_contract/agent_channel/work_context/sources/0/content/goal_acceptance_contract/outcome_requirements/outcomes/0" + + def navigation(name, enabled=True): + directory = tmp_path / name + directory.mkdir(mode=0o700) + prepared = codex_context.prepare_codex_context(navigated_decision, directory=str(tmp_path), + capture_directory=directory, goal_id="fixture", agent_id="agent", surface="quota_packet", + capture_id=name, progressive_enabled=enabled) + return model_context_reference(native_context_capture_view(prepared.delivery, directory, + context_directory=str(tmp_path), progressive_enabled=enabled, + initial_read_plan=prepared.initial_read_plan, max_page_output_bytes=6000)) + + def pages(command): + result = [] + while command: + words = shlex.split(command) + page = json.loads(subprocess.check_output([sys.executable, "-m", "loopx.entrypoint", "--format", "json", + *words[words.index("quota"):]])) + assert page["ok"] + result.append(page) + command = page.get("next_command") + return result + + first = pages(navigation("before")["first_read_command"]) + assert len(first) > 1 + deliver(tmp_path, event, first[1:] if invalidate == "partial" else first) + if invalidate == "compact": + codex_context.codex_context_hook(str(tmp_path), {**event, "hook_event_name": "PostCompact"}) + # A real canonical correction also changes metadata and may switch from a + # complete source to a smaller supporting-material projection. Neither is a + # change to the seven retained descriptions. Reviews and new controls are. + outcome["revision"] += 1 + outcome["sources"] = [{"ref": "fixture:correction", "summary": "Supporting correction evidence. " * 600}] + outcome["requirements"][0]["description"] = "Revised cancellation: never publish the result." + outcome["requirements"][1]["status"] = "accepted" + outcome["requirements"][2]["new_control"] = "Read the current authorization before acting." + outcome["requirements"].append({"id": "new", "description": "A newly required independent check."}) + if retained == 1: + for item in outcome["requirements"][2:8]: + item["description"] = "Revised requirement: " + item["description"] + current = navigation("after", invalidate != "off") + read = pages(current["first_read_command"] if invalidate != "off" else current["recovery"]["command"]) + if invalidate == "off": + assert "first_read_command" not in current + assert json.loads("".join(page["content"] for page in read)) == navigated_decision + assert all("previously_delivered_fields" not in page for page in read) + return + body = "".join(page["content"] for page in read) + facts = selected_context_values(body) + assert facts[prefix + "/revision"] == 2 + assert facts[prefix + "/requirements/0/description"] == outcome["requirements"][0]["description"] + assert facts[prefix + "/requirements/1/status"] == "accepted" + assert facts[prefix + "/requirements/2/new_control"] == outcome["requirements"][2]["new_control"] + assert facts[prefix + "/requirements/8"] == outcome["requirements"][8] + assert facts["/goal_boundary"] == navigated_decision["goal_boundary"] + if invalidate != "none" or retained == 1: + assert "previously_delivered_fields" not in read[0] + assert facts[prefix + "/requirements/1/description"] == outcome["requirements"][1]["description"] + return + expected = {prefix + f"/requirements/{i}/description" for i in (1, 2)} | { + prefix + f"/requirements/{i}" for i in range(3, 8)} + assert set(read[0]["previously_delivered_fields"]) >= expected + assert prefix + "/requirements/1/description" not in facts + cold = pages(current["recovery"]["command"] + " --initial-read") + assert len(body.encode()) < sum(len(page["content"].encode()) for page in cold) + recovered = pages(current["recovery"]["command"] + " --select " + prefix + "/requirements/1/description") + assert selected_context_values("".join(page["content"] for page in recovered))[ + prefix + "/requirements/1/description"] == outcome["requirements"][1]["description"] + before = codex_context._state(tmp_path)["consumed_pointers"][prefix] + metadata = json.loads(Path(read[0]["reference_file"]).read_text()) + metadata["observed_read"]["fields"][next(iter(metadata["observed_read"]["fields"]))] = "f" * 64 + forged = tmp_path / "forged-reference.json" + forged.write_text(json.dumps(metadata)) + forged.chmod(0o600) + deliver(tmp_path, event, [{**page, "reference_file": str(forged)} for page in read]) + assert codex_context._state(tmp_path)["consumed_pointers"][prefix] == before + # Only complete current pages plus authenticated, exact inherited fields + # can qualify the revised parent. No full-input baseline is manufactured. + with monkeypatch.context() as hook_env: + hook_env.delenv("CODEX_THREAD_ID") + deliver(tmp_path, event, read[:-1]) + assert codex_context._state(tmp_path)["consumed_pointers"][prefix] == before + deliver(tmp_path, event, read[-1:]) + assert codex_context._state(tmp_path)["consumed_pointers"][prefix] != before + next_read = pages(navigation("continued")["first_read_command"]) + assert prefix in next_read[0]["previously_delivered_fields"] + assert selected_context_values("".join(page["content"] for page in next_read))[prefix + "/revision"] == 2 + + @pytest.mark.parametrize("invalidate", ["compact", "resume", "partial", "task_switch", "off"]) @pytest.mark.parametrize("coalesced", [False, True]) def test_native_first_read_reuses_only_observed_current_fields(tmp_path, monkeypatch, navigated_decision, invalidate, coalesced):