diff --git a/loopx/control_plane/quota/progressive_context.ts b/loopx/control_plane/quota/progressive_context.ts index 1593589897..68df57c7a6 100644 --- a/loopx/control_plane/quota/progressive_context.ts +++ b/loopx/control_plane/quota/progressive_context.ts @@ -7,7 +7,7 @@ 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"; -import {FULL_OUTCOME_READ_GUIDANCE, PROGRESSIVE_OUTCOME_READ_GUIDANCE} +import {FULL_OUTCOME_READ_GUIDANCE, PROGRESSIVE_OUTCOME_READ_GUIDANCE, verifiedSelectedTodoSource} from "../work_items/interaction_contract.ts"; import {quotaSummaryReferences, contextSourceIdentity, contextProjectionChanges, type ProjectionSourceValues} from "./diagnostic_projection.ts"; @@ -18,7 +18,7 @@ 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 pointer = (path: Path): string => path.length ? "/" + path.map(key => String(key).replaceAll("~", "~0").replaceAll("/", "~1")).join("/") : ""; @@ -29,10 +29,11 @@ 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): 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) => { + const workPath: Path = ["interaction_contract", "agent_channel", "work_context"]; + const work = jsonObject(at(context, workPath)); + if (!work || !Array.isArray(work.sources)) return []; + const base = [...workPath, "sources"]; + const fields = work.sources.flatMap((value, i) => { const source = jsonObject(value), content = jsonObject(source?.content); if (!content) return []; const path = [...base, i, "content"]; @@ -42,7 +43,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 []; @@ -73,11 +74,36 @@ function reusableFirstReadFields(context: JsonObject): ReusableReadField[] { return fields; }) : []; }); + // Only the note may carry forward, after a fresh canonical read of the ready + // Todo. Keep the current task, ownership, constraints and settlement visible. + const todoId = jsonObject(context.selected_todo)?.todo_id; + const selected = verifiedSelectedTodoSource(work, todoId); + const authority = jsonObject(selected?.content.authority_read); + const preparation = jsonObject(at(context, ["interaction_contract", "cli_channel", "execution_preparation"])); + const identity = jsonObject(context.agent_identity); + if (context.effective_action === EffectiveAction.NORMAL_RUN && context.normal_delivery_allowed === true + && at(context, ["interaction_contract", "agent_channel", "delivery_allowed"]) === true + && preparation?.state === "ready" && preparation.delivery_allowed === true && preparation.todo_id === todoId + && typeof preparation.agent_id === "string" && preparation.agent_id === (context.agent_id ?? identity?.agent_id) + && (identity?.agent_id === undefined || identity.agent_id === preparation.agent_id) + && typeof context.goal_id === "string" && selected?.content.ambiguous === false + && typeof selected.source.command === "string" && selected.source.command.length > 0 + && typeof selected.todo.note === "string" && typeof selected.todo.updated_at === "string" + && authority?.decision_read_from_provider === true && authority.legacy_fallback_used === false + && typeof authority.provider_revision === "string" && typeof authority.source_authority === "string" + && authority.source_authority === selected.content.source) { + fields.push({path: [...base, selected.index, "content", "todo", "note"], freshKeys: [], + minimumBytes: MIN_REUSABLE_BODY_BYTES, sourceIdentity: JSON.stringify([ + context.goal_id, preparation.agent_id, todoId, selected.source.command, authority.source_authority, + selected.todo.updated_at, + ])}); + } + 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]); + // A repeated body at the same position can belong to a different source. + const value = field?.sourceIdentity === undefined ? body : JSON.stringify([field.sourceIdentity, body]); return createHash("sha256").update(value).digest("hex"); } diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts index 649ffaad59..655cd2f8ed 100644 --- a/loopx/control_plane/quota/turn_envelope.ts +++ b/loopx/control_plane/quota/turn_envelope.ts @@ -26,7 +26,7 @@ import { projectTurnStartUnavailableContext } from "../capability_hooks.ts"; import { jsonObject, requireBoolean, requireJsonObject, requireNonEmptyString, requireStringLiteral, sourceNumberTokens } from "../runtime_decode.ts"; import { projectPendingCapabilityIntent } from "../work_items/pending_capability_intent.ts"; -import { INTERACTION_CONTRACT_SCHEMA_VERSION, assertGoalContextDelivered, assertInteractionRequiredReads, capturedTodoTextSha256, decodeInteractionContract, projectInteractionRequiredReads, selectedWorkTodo, selectedWorkTodoId, USER_CHANNEL_NOTIFICATION_POLICIES } from "../work_items/interaction_contract.ts"; +import { INTERACTION_CONTRACT_SCHEMA_VERSION, assertGoalContextDelivered, assertInteractionRequiredReads, capturedTodoTextSha256, decodeInteractionContract, projectInteractionRequiredReads, selectedWorkTodo, selectedWorkTodoId, USER_CHANNEL_NOTIFICATION_POLICIES, verifiedSelectedTodoSource } from "../work_items/interaction_contract.ts"; import { measureTurnEnvelope, TURN_ENVELOPE_SECTION_TARGETS } from "./turn_envelope_budget.ts"; export { TURN_ENVELOPE_BUDGET_BYTES } from "./turn_envelope_budget.ts"; @@ -598,17 +598,8 @@ function referenceVisionContinuation(payload: JsonObject, function currentTaskTitleOwner(payload: JsonObject): {index: number; todo: JsonObject} | null { const lane = object(payload.agent_lane_next_action), selected = object(payload.selected_todo); const context = object(object(object(payload.interaction_contract).agent_channel).work_context); - if (context.complete !== true || context.selected_todo_text_verified !== true || - typeof lane.todo_id !== "string" || !lane.todo_id || selected.todo_id !== lane.todo_id || - !Array.isArray(context.sources)) return null; - const matches = context.sources.map((value, index) => ({source: object(value), index})) - .filter(({source}) => source.source === "selected_todo"); - if (matches.length !== 1) return null; - const {source, index} = matches[0], content = object(source.content), todo = object(content.todo); - if (content.ok !== true || content.matched !== true || content.ambiguous === true || - content.read_complete === false || typeof todo.text !== "string" || - todo.archive_state === "archive" || todo.status === "done" || todo.todo_id !== lane.todo_id) return null; - return {index, todo}; + if (selected.todo_id !== lane.todo_id) return null; + return verifiedSelectedTodoSource(context, lane.todo_id); } function requireCurrentTaskTitleReference(payload: JsonObject): void { diff --git a/loopx/control_plane/work_items/interaction_contract.ts b/loopx/control_plane/work_items/interaction_contract.ts index f7f4dba175..5d8bd3ccb8 100644 --- a/loopx/control_plane/work_items/interaction_contract.ts +++ b/loopx/control_plane/work_items/interaction_contract.ts @@ -881,6 +881,24 @@ export function assertGoalContextDelivered(channel: JsonObject): void { } } +/** The complete selected Todo is the source for current task text and notes. */ +export function verifiedSelectedTodoSource(context: JsonObject, todoId: unknown): + {index: number; source: JsonObject; content: JsonObject; todo: JsonObject} | null { + if (context.complete !== true || context.selected_todo_text_verified !== true || + typeof todoId !== "string" || !todoId || !Array.isArray(context.sources)) return null; + const matches = context.sources.flatMap((value, index) => { + const source = jsonObject(value); + return source?.source === "selected_todo" ? [{source, index}] : []; + }); + if (matches.length !== 1) return null; + const {source, index} = matches[0]; + const content = jsonObject(source.content), todo = jsonObject(content?.todo); + if (content?.ok !== true || content.matched !== true || content.ambiguous === true || + content.read_complete === false || typeof todo?.text !== "string" || + todo.archive_state === "archive" || todo.status === "done" || todo.todo_id !== todoId) return null; + return {index, source, content, todo}; +} + async function readSelectedTodoContext(source: JsonObject, todoId: unknown): Promise { if (typeof todoId !== "string" || !todoId) return {error_code: "context_source_unavailable"}; const {readCanonicalSnapshot, CANONICAL_SNAPSHOT_PAGE_REQUEST} = diff --git a/tests/control_plane/test_context_first_read.py b/tests/control_plane/test_context_first_read.py index 915e4ceaef..c76f7931c2 100644 --- a/tests/control_plane/test_context_first_read.py +++ b/tests/control_plane/test_context_first_read.py @@ -1185,3 +1185,119 @@ 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("provider", ["file", "sqlite"]) +def test_ready_native_guard_reuses_only_observed_selected_todo_note(tmp_path, monkeypatch, provider): + from test_codex_context_lifecycle import deliver, setup + from test_quota_settlement_cli import ( + AGENT_ID, GOAL_ID, TODO_ID, _configure_read_only_todo, _heartbeat_receipt_count, + _initialize_fixture_authority, _run_cli, _spend_run_count, _write_fixture, + ) + from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime + from loopx.control_plane.coordination.local_authority import read_canonical_todos_if_promoted + + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_fixture(tmp_path) + state = _configure_read_only_todo(project) + task = "Validate and settle the selected delivery. " + task += "Keep the independent verification and its evidence available. " * 32 + state.write_text(state.read_text().replace("Validate and settle the selected delivery.", task.strip())) + registry_body = json.loads(registry.read_text()) + registry_body["goals"][0]["quota"]["allowed_slots"] = 3 + registry.write_text(json.dumps(registry_body)) + _initialize_fixture_authority(project, runtime, provider=provider, handoff_mode="hard_lease") + + journal = tmp_path / "journal" + journal.mkdir(mode=0o700) + captures = tmp_path / "captures" + captures.mkdir(mode=0o700) + event = setup(journal, monkeypatch) + scope = ["--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--todo-id", TODO_ID] + + def run(*args): + code, reply = _run_cli(registry, runtime, *args, cwd=project) + assert code == 0 and reply["ok"], {key: reply.get(key) for key in ("error", "error_code", "reason", "state")} + return reply + + claim = run("todo", "claim", *scope, "--claimed-by", AGENT_ID) + note = ("复核原始证据并保留任务的完整验证条件。 " * 360).strip() + + def update_note(text, operation): + current = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=GOAL_ID) + assert current is not None + run("todo", "update", *scope, "--update-operation-id", operation, "--note", text, + "--update-expected-provider-revision", current["provider_revision"], + "--claim-operation-id", claim["claim_operation_id"]) + + def capture(turn): + return run("quota", "should-run", "--codex-app", "--context-delivery", "on", + "--codex-context-dir", str(journal), "--decision-output-root", str(captures), + "--turn-instance-id", turn, "--available-capability", "shell", + "--available-capability", "filesystem_read", *scope) + + def read(command): + pages, offset = [], 0 + while command: + words = shlex.split(command) + page = run(*words[words.index("quota"):]) + assert page["offset"] == offset + pages.append(page) + 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)) + + def settle(view): + plan_path = "/interaction_contract/cli_channel/settlement_plan" + plan = read(view["recovery"]["command"] + " --select " + plan_path)[1][plan_path] + steps = {step["kind"]: step for step in plan["ordered_steps"]} + command = shlex.split(steps["durable_writeback"]["command_template"]) + replacements = {"": "fixture_note_reviewed", + "": "single_surface", "": "outcome_progress"} + argv = [replacements.get(word, word) for word in command[command.index("refresh-state"):]] + written = run(*argv, "--delivery-boundary", "in_flight_continuation", + "--no-global-sync", "--suppress-external-sinks") + if written["settlement_progress"]["state"] == "spend_required": + command = shlex.split(steps["quota_spend"]["command_template"]) + written = run(*command[command.index("quota"):]) + assert written["settlement_progress"]["state"] == "settled" + assert written["settlement_identity"] == plan["identity"] + + update_note(note, "fixture-note-original") + cold = capture("note-cold") + pages, facts = read(cold["first_read_command"]) + note_pointer = next(path for path in facts if path.endswith("/content/todo/note")) + assert facts[note_pointer] == note + task_pointer = note_pointer.removesuffix("/note") + "/text" + current_task = facts[task_pointer] + assert current_task.endswith(task.strip()) + assert "previously_delivered_fields" not in pages[0] + deliver(journal, event, pages) + settle(cold) + + warm = capture("note-warm") + pages, facts = read(warm["first_read_command"]) + assert note_pointer in pages[0]["previously_delivered_fields"] + assert note_pointer not in facts + assert facts[task_pointer] == current_task + assert facts["/heartbeat_receipt"]["turn_instance_id"] == "note-warm" + assert "/goal_boundary" in facts + + recovery = warm["recovery"]["command"] + " --select " + shlex.quote(note_pointer) + assert read(recovery)[1][note_pointer] == note + codex_context.codex_context_hook(str(journal), {**event, "hook_event_name": "PostCompact"}) + pages, facts = read(warm["first_read_command"]) + assert "previously_delivered_fields" not in pages[0] and facts[note_pointer] == note + deliver(journal, event, pages) + settle(warm) + + correction = note + " 采用新的独立验收条件。" + update_note(correction, "fixture-note-corrected") + corrected = capture("note-corrected") + pages, facts = read(corrected["first_read_command"]) + assert note_pointer not in pages[0].get("previously_delivered_fields", []) + assert facts[note_pointer] == correction + assert all(_heartbeat_receipt_count(runtime, turn) == 1 + for turn in ("note-cold", "note-warm", "note-corrected")) + assert _spend_run_count(runtime) == 2