Reuse unchanged Todo notes in native quota reads

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<ReusableReadField>((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<ReusableReadField>((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");
 }
 

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 {

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<JsonObject> {
   if (typeof todoId !== "string" || !todoId) return {error_code: "context_source_unavailable"};
   const {readCanonicalSnapshot, CANONICAL_SNAPSHOT_PAGE_REQUEST} =

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 = {"<validated_progress>": "fixture_note_reviewed",
+                        "<scale>": "single_surface", "<outcome>": "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