fix(context): reuse exact Todo rows in native required views

docs/reference/required-work-context.md

@@ -109,6 +109,21 @@ Historical captures retain their original obligations. This acceptance-material
 slice does not yet stage candidate selection, preferences or Explore history,
 and a smaller body alone does not establish lower total model cost.
 
+Native required views also replace exact repeated registered Todo rows in
+summary lanes with `{todo_id, context_ref}`. The JSON pointer addresses the
+complete first occurrence in this **same required view**, never an old baseline
+or an unexpanded detail. Every lane retains its membership and order; all unique
+fields, revisions, ownership, dependency state, contrary evidence and final
+constraints remain inline. Referenced rows carry exactly those same obligations.
+Unknown row fields/schemas, unknown lanes and unsafe numeric values stay literal.
+The reader must recover a missing target from this same capture before dependent
+work; it does not need a new detail read when the complete target is present.
+This is exact reuse, not candidate relevance or selection. It changes no
+canonical record or authority, and does not permit skipping other candidates.
+Signed envelopes, disabled native delivery and historical captures keep their
+existing behavior. The existing size comparison retains a full view when the
+reference guidance would cost more than the duplicate rows.
+
 A context epoch is not a work Turn. Compaction or a failed native hook invalidates
 delta reuse; it does not require another admission or unconditional replay of the
 old quota packet. Within the existing Turn, continue from the working brief with

loopx/control_plane/quota/diagnostic_projection.ts

@@ -7,6 +7,7 @@
 import type {JsonObject} from "../effect_program.ts";
 import {jsonObject} from "../runtime_decode.ts";
 import {TODO_CANONICAL_READ_RECORD_FIELDS} from "../coordination/coordination_state_contract.ts";
+import {canonicalAuthorityJson} from "../coordination/authority_store_codec.ts";
 
 type ProjectionPath = (string | number)[];
 type ProjectionChange = {path: ProjectionPath; value: unknown} |
@@ -74,6 +75,38 @@ const DIAGNOSTIC_LANES = new Set([
 // The read codec adds these display fields to canonical records. Retain them
 // on visible items; they do not make a known diagnostic list a future schema.
 const ITEM_FIELDS = new Set<string>([...TODO_CANONICAL_READ_RECORD_FIELDS, "index", "content_revision"]);
+
+/** Exact repeated records remain fully available in this model view. Unlike
+ * diagnostic omission, this preserves every lane's membership and ordering,
+ * including structured dependencies and handoffs. No relevance is inferred. */
+export function quotaSummaryReferences(payload: JsonObject): JsonObject[] {
+  const references: JsonObject[] = [];
+  const safeNumbers = (value: unknown): boolean => typeof value === "number"
+    ? Number.isSafeInteger(value)
+    : value === null || typeof value !== "object" || Object.values(value).every(safeNumbers);
+  for (const root of ["agent_todo_summary", "user_todo_summary"]) {
+    const summary = jsonObject(payload[root]);
+    if (summary?.schema_version !== "todo_summary_v0") continue;
+    const seen = new Map<string, ProjectionPath>();
+    for (const [lane, rows] of Object.entries(summary)) {
+      if (!DIAGNOSTIC_LANES.has(lane) || !Array.isArray(rows)) continue;
+      rows.forEach((value, index) => {
+        const row = jsonObject(value);
+        if (!row || row.schema_version !== "todo_item_v0" || typeof row.todo_id !== "string"
+            || !row.todo_id || !Object.keys(row).every(key => ITEM_FIELDS.has(key)) || !safeNumbers(row)) return;
+        const key = JSON.stringify(canonicalAuthorityJson(row)), path = [root, lane, index];
+        const from = seen.get(key);
+        if (!from) { seen.set(key, path); return; }
+        // Keep the original literal as the target; references never form chains.
+        const reference = {todo_id: row.todo_id, context_ref: "/" + from.join("/")};
+        if (Buffer.byteLength(JSON.stringify(reference)) < Buffer.byteLength(JSON.stringify(row)))
+          references.push({path, from, value: reference});
+      });
+    }
+  }
+  return references;
+}
+
 function knownItem(value: unknown): boolean {
   const scalar = (value: unknown) => value === null || ["string", "number", "boolean"].includes(typeof value);
   const item = jsonObject(value);

loopx/control_plane/quota/progressive_context.ts

@@ -4,6 +4,7 @@ import type {JsonObject} from "../effect_program.ts";
 import {jsonObject, requireJsonObject} from "../runtime_decode.ts";
 import {GOAL_OUTCOMES_SCHEMA} from "../goals/outcome_requirements.ts";
 import {FULL_OUTCOME_READ_GUIDANCE, PROGRESSIVE_OUTCOME_READ_GUIDANCE} from "../work_items/interaction_contract.ts";
+import {quotaSummaryReferences} from "./diagnostic_projection.ts";
 
 type Path = (string | number)[];
 const at = (root: unknown, path: Path): unknown => path.reduce<unknown>((row, key) =>
@@ -19,7 +20,9 @@ export function progressiveContextPlan(request: JsonObject): JsonObject {
   // Signed envelopes need their own presentation contract. Unknown shapes and
   // already projected packets retain their entire original content.
   if (request.surface !== "quota_packet" || payload.ok !== true || payload.schema_version !== undefined
-      || payload.contract_binding !== undefined || payload.context_details !== undefined) return unchanged;
+      || payload.contract_binding !== undefined || payload.context_details !== undefined
+      || payload.context_reference_instruction !== undefined) return unchanged;
+  const references = quotaSummaryReferences(payload);
   const carriers: Path[] = [
     ["planning_horizon", "core_goal", "acceptance_contract"],
     ["autonomous_replan_obligation", "replan_context", "core_goal", "acceptance_contract"],
@@ -67,10 +70,10 @@ export function progressiveContextPlan(request: JsonObject): JsonObject {
         deferred.push({path: [...parent, "requirements", j, "agent_review", field], parent, index_key});
     });
   }
-  if (!deferred.length) return unchanged;
+  if (!deferred.length && !references.length) return unchanged;
   // Replace only the exact source-owned obligation. Other merged obligations,
   // objective adoption, unknown provider prose and pending reads remain intact.
-  for (const listPath of [sourcePath, [...agent, "required_reads"]]) {
+  for (const listPath of deferred.length ? [sourcePath, [...agent, "required_reads"]] : []) {
     const rows = at(payload, listPath);
     if (!Array.isArray(rows)) continue;
     rows.forEach((value, i) => {
@@ -81,5 +84,9 @@ export function progressiveContextPlan(request: JsonObject): JsonObject {
           row.reason.replace(FULL_OUTCOME_READ_GUIDANCE, PROGRESSIVE_OUTCOME_READ_GUIDANCE)});
     });
   }
-  return {deferred, instructions};
+  if (references.length) instructions.push({path: ["context_reference_instruction"], value:
+    "A Todo row with context_ref is exactly the complete row at that JSON pointer in THIS required view. "
+    + "Use that row for every referenced lane; all constraints, evidence, ownership and order still apply. "
+    + "No detail read or older baseline is needed. Missing target content requires recovery of this same capture before dependent work."});
+  return {deferred, instructions, ...(references.length ? {references} : {})};
 }

loopx/control_plane/turn_driver/codex_context.py

@@ -478,7 +478,7 @@ def _capture_required_context(payload: dict[str, Any], *, directory: Path,
         plan = plan_progressive_context(payload, surface=surface)
     except (EffectRuntimeRemoteError, EffectRuntimeStartupError):
         return payload, archive
-    if not plan["deferred"]:
+    if not plan["deferred"] and not plan.get("references"):
         return payload, archive
     projected = copy.deepcopy(payload)
 
@@ -488,6 +488,13 @@ def at(path):
             row = row[key]
         return row
 
+    for edit in plan.get("references", []):
+        # Compare original Python values as well: the typed bridge must never
+        # equate distinct integers or change the source record's numeric types.
+        value, target = at(edit["path"]), at(edit["from"])
+        if json.dumps(value, sort_keys=True) != json.dumps(target, sort_keys=True):
+            return payload, archive
+        at(edit["path"][:-1])[edit["path"][-1]] = edit["value"]
     for edit in plan["deferred"]:
         path, parent = edit["path"], edit["parent"]
         del at(path[:-1])[path[-1]]
@@ -496,7 +503,7 @@ def at(path):
     for edit in plan["instructions"]:
         at(edit["path"][:-1])[edit["path"][-1]] = edit["value"]
     page_bytes, output_bytes = native_context_page_budget()
-    projected["context_details"] = {
+    details = {
         "path": archive["path"], "sha256": archive["sha256"],
         "command_template": context_capture_read_command(Path(archive["path"]), archive["sha256"],
             page_bytes=page_bytes, max_page_output_bytes=output_bytes) + " --select '<supporting_material_pointer>/sources'",
@@ -509,6 +516,8 @@ def at(path):
             "Detail reads never establish a full baseline. "
             "Uncertain or missing necessary facts must be recovered before dependent action; Agent review is not independent acceptance.",
     }
+    if plan["deferred"]:
+        projected["context_details"] = details
     # These are transport bytes, not a semantic relevance heuristic or a model
     # token estimate. Small material is cheaper inline than an index plus IO.
     required_bytes = len((json.dumps(projected, ensure_ascii=False, separators=(",", ":"), allow_nan=False) + "\n").encode("utf-8"))

tests/control_plane/test_progressive_native_context.py

@@ -194,6 +194,71 @@ def test_projection_preserves_numbers_and_original_values(tmp_path, monkeypatch)
     assert original == before and required["opaque_counter"] == before["opaque_counter"]
 
 
+def test_saved_cli_references_keep_tail_constraints_and_recover_without_detail_reads(tmp_path, monkeypatch):
+    event = setup(tmp_path, monkeypatch)
+    row = {"schema_version": "todo_item_v0", "todo_id": "task", "text": "Identical title",
+        "note": "Complete work and verify evidence. " * 150 + "Stop before publication.",
+        "content_revision": "revision:one", "claimed_by": "peer",
+        "resume_condition": {"schema_version": "todo_resume_condition_v0", "satisfied": False}}
+    revised = {**row, "content_revision": "revision:two", "note": "New contrary evidence; reconsider."}
+    original = {"ok": True, "goal_id": "fixture", "agent_identity": {"agent_id": "agent"},
+        "heartbeat_receipt": {"turn_instance_id": "original-turn"},
+        "agent_todo_summary": {"schema_version": "todo_summary_v0", "first_open_items": [row, revised],
+            "backlog_items": [copy.deepcopy(row), copy.deepcopy(revised)],
+            "claimed_open_items": [copy.deepcopy(row)]}}
+    path = tmp_path / "original.json"
+    raw = json.dumps(original).encode()
+    path.write_bytes(raw)
+    assert context.project_codex_context(original, directory=None, capture_directory=tmp_path,
+        goal_id="fixture", agent_id="agent", surface="quota_packet", capture_id="off") is original
+    digest = hashlib.sha256(raw).hexdigest()
+    registry, runtime = tmp_path / "no-registry.json", tmp_path / "no-runtime"
+    command = ["quota", "context-project", "--capture-file", str(path), "--capture-sha256", digest,
+        "--goal-id", "fixture", "--agent-id", "agent", "--turn-instance-id", "original-turn",
+        "--context-delivery", "on", "--codex-app", "--codex-context-dir", str(tmp_path),
+        "--max-page-output-bytes", "4000"]
+    for cycle in range(3):
+        if cycle == 2:
+            context.codex_context_hook(str(tmp_path), {**event, "source": "compact"})
+        code, view = _run_cli(registry, runtime, *command, cwd=tmp_path)
+        assert code == 0 and view["ok"], view
+        assert view["mode"] == ("delta" if cycle == 1 else "full")
+        argv = shlex.split(view["recovery"]["command"])
+        pages, contents = [], b""
+        while True:
+            code, page = _run_cli(registry, runtime, *argv[argv.index("quota"):], cwd=tmp_path)
+            assert code == 0 and page["offset"] == len(contents), page
+            assert len((json.dumps(page, ensure_ascii=False, separators=(",", ":")) + "\n").encode()) <= 4000
+            contents += page["content"].encode()
+            pages.append(page)
+            if page["eof"]:
+                break
+            argv = shlex.split(page["next_command"])
+        required = json.loads(contents)
+        assert "context_details" not in required, "in-view reuse must not create extra evidence reads"
+        shown = required["agent_todo_summary"]
+        for lane in ("first_open_items", "backlog_items", "claimed_open_items"):
+            resolved = []
+            for item in shown[lane]:
+                if "context_ref" in item:
+                    value = required
+                    for part in item["context_ref"].split("/")[1:]:
+                        value = value[int(part)] if isinstance(value, list) else value[part]
+                    assert "context_ref" not in value
+                    resolved.append(value)
+                else:
+                    resolved.append(item)
+            assert resolved == original["agent_todo_summary"][lane]
+        assert shown["first_open_items"][0]["note"].endswith("Stop before publication.")
+        assert shown["first_open_items"][1]["note"] == revised["note"]
+        if cycle == 0:
+            assert len(pages) > 1
+            deliver(tmp_path, event, pages[:1])
+            assert "candidate" not in context._state(tmp_path)
+        deliver(tmp_path, event, pages)
+    assert path.read_bytes() == raw and not registry.exists() and not runtime.exists()
+
+
 def test_unavailable_typed_plan_keeps_original_capture(tmp_path, monkeypatch):
     from loopx.control_plane.quota import turn_envelope
     from loopx.control_plane.effect_runtime import EffectRuntimeStartupError
@@ -221,6 +286,15 @@ def test_small_material_keeps_full_capture_when_the_detail_index_costs_more(tmp_
     assert not (tmp_path / "required-context.json").exists()
 
 
+def test_small_duplicate_rows_do_not_gain_reference_overhead(tmp_path):
+    row = {"schema_version": "todo_item_v0", "todo_id": "task", "text": "Keep this constraint.", "claimed_by": "peer"}
+    original = {"ok": True, "agent_todo_summary": {"schema_version": "todo_summary_v0",
+        "first_open_items": [row], "backlog_items": [dict(row)]}}
+    required, recovery = context._capture_required_context(original, directory=tmp_path, surface="quota_packet")
+    assert required is original
+    assert Path(recovery["path"]).name == "projection.json"
+
+
 def test_public_quota_cli_delivers_required_view_and_original_evidence(acceptance_goal, tmp_path, monkeypatch):
     workspace, document, cli, run = acceptance_goal
     original = {"outcome_id": "export", "owner_agent_id": "agent-a", "todo_ids": ["todo_export"],

tests/control_plane_ts/progressive_context.test.ts

@@ -40,3 +40,36 @@ test("unknown and failed source semantics remain whole rather than becoming gues
   assert.deepEqual(plan.instructions, []);
   assert.deepEqual(plan.deferred, []);
 });
+
+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",
+    claimed_by: "peer", resume_condition: {schema_version: "todo_resume_condition_v0", satisfied: false}};
+  const revised = {...row, content_revision: "revision:two", note: "Contrary evidence requires reconsideration."};
+  const summary = {schema_version: "todo_summary_v0", first_open_items: [row, revised],
+    backlog_items: [row, revised, {...row, claimed_by: "other"}]};
+  const payload = {ok: true, agent_todo_summary: summary};
+  const before = structuredClone(payload);
+  const plan = progressiveContextPlan({surface: "quota_packet", payload});
+  assert.deepEqual(plan.references, [
+    {path: ["agent_todo_summary", "backlog_items", 0], from: ["agent_todo_summary", "first_open_items", 0],
+      value: {todo_id: row.todo_id, context_ref: "/agent_todo_summary/first_open_items/0"}},
+    {path: ["agent_todo_summary", "backlog_items", 1], from: ["agent_todo_summary", "first_open_items", 1],
+      value: {todo_id: row.todo_id, context_ref: "/agent_todo_summary/first_open_items/1"}},
+  ]);
+  assert.deepEqual(plan.deferred, []);
+  assert.deepEqual(payload, before);
+  const pendingSource = {...source, content: {...source.content, read_complete: false}};
+  const pendingPlan = progressiveContextPlan({surface: "quota_packet", payload: {...payload,
+    interaction_contract: {agent_channel: {work_context: {sources: [pendingSource]},
+      required_reads: [{source: "goal_acceptance", reason: FULL_OUTCOME_READ_GUIDANCE}]}}}});
+  assert.deepEqual(pendingPlan.deferred, []);
+  assert.deepEqual(pendingPlan.instructions.map(row => row.path), [["context_reference_instruction"]]);
+  for (const unknown of [{...row, new_obligation: "Keep inline."}, {...row, schema_version: "future"},
+    {...row, resume_monitor_generation: 2 ** 60}, {...row, resume_monitor_generation: 1.25}]) {
+    const view = {ok: true, agent_todo_summary: {...summary, first_open_items: [unknown], backlog_items: [unknown]}};
+    assert.equal(progressiveContextPlan({surface: "quota_packet", payload: view}).references, undefined);
+  }
+  assert.equal(progressiveContextPlan({surface: "quota_packet", payload: {...payload,
+    contract_binding: {signature: "signed"}}}).references, undefined);
+});