fix(context): reuse retained candidate authoring in quota deltas

docs/reference/required-work-context.md

@@ -284,6 +284,14 @@ Unknown, failed, incompatible or ambiguous hook views stay complete. A changed
 scope or lost baseline still selects full delivery; reuse never makes a retired
 or expired preference actionable or grants authority.
 
+Quota capability candidates use the same exact codec for registered Todo text,
+title and note strings. Every candidate remains in the complete cold input;
+warm input still carries current membership/order, identity, revision, ownership,
+eligibility, handoff and unknown fields. Unknown schemas, duplicate identities or
+structured values in those text fields stay literal. This removes repeated
+authoring only when a complete baseline is usable; it neither narrows candidate
+selection nor implements the proposed decision-stage read contract.
+
 Object edits may group complete current fields in `values`, alongside recursive
 `changed` edits and `removed` keys. Array edits use exact indices in `items` and
 `values`, with `length` for resizing. These literals are present in the current
@@ -608,6 +616,11 @@ TurnEnvelope。在基线连续性成立的口径内，两入口仍逐轮发送 `
 读取不完整时保留完整投递。
 仅因移除可复用字段而变空的容器，不会把这些字段重新带回差分；原本为空的未知对象和
 非字符串来源引用仍逐轮发送。复用保留定位符的精确值，不重读其指向的材料，也不刷新权限。
+quota 能力候选沿用同一精确 codec 复用已登记 Todo 的正文、标题与笔记字符串。
+冷启动仍完整交付全部候选；warm 仍逐轮携带当前成员与顺序、身份、修订、归属、
+准入能力、交接及未知字段。未知 schema、重复身份或文本字段中的结构化值保持原文。
+这只在完整基线可用时减少重复正文，不缩小候选选择范围，也不等于按决策阶段读取的
+替代契约已经实现。
 对象编辑可以在 `values` 中合并携带本轮完整字段，同时保留递归 `changed` 与删除
 `removed`；数组的 `items` 与 `values` 使用精确索引，`length` 调整长度。
 字段仍在本轮输入中，只减少重复 wrapper；codec 比较普通编辑、合并字段

loopx/control_plane/quota/context_delivery.ts

@@ -242,6 +242,32 @@ function freshPaths(context: JsonObject, surface: Surface, legacyQuotaFields: re
       });
     }
   };
+  const candidates = (value: unknown, path: Path) => {
+    const gate = jsonObject(value), ids = new Set<string>();
+    const lanes = ["runnable_candidates", "blocked_candidates"];
+    if (gate?.schema_version !== "capability_gate_v0"
+        || lanes.some(lane => !Array.isArray(gate[lane]) || (gate[lane] as unknown[]).some(value => {
+          const row = jsonObject(value), id = row?.todo_id;
+          if (typeof id !== "string" || !id || ids.has(id)) return true;
+          ids.add(id); return false;
+        }))) { paths.push(path); return; }
+    for (const [field, item] of Object.entries(gate)) {
+      if (!lanes.includes(field)) { paths.push([...path, field]); continue; }
+      if (!(item as unknown[]).length) { paths.push([...path, field]); continue; }
+      (item as JsonObject[]).forEach((row, i) => {
+        const prefix = [...path, field, i];
+        if (row.schema_version !== COORDINATION_STATE_CONTRACT.todo_read_record.item_schema_version
+            || ["text", "title", "note"].some(key => Object.hasOwn(row, key) && typeof row[key] !== "string")) {
+          paths.push(prefix); return;
+        }
+        // Reuse only registered authoring strings through the exact codec.
+        // Selection, order, revisions, ownership, capability facts, handoffs and
+        // all unknown controls remain current literals, even when unchanged.
+        for (const key of Object.keys(row))
+          if (!["text", "title", "note"].includes(key)) paths.push([...prefix, key]);
+      });
+    }
+  };
   const horizon = (value: unknown, path: Path) => {
     const object = jsonObject(value);
     if (!object) { paths.push(path); return; }
@@ -278,6 +304,7 @@ function freshPaths(context: JsonObject, surface: Surface, legacyQuotaFields: re
   } else {
     for (const [key, value] of Object.entries(context)) {
       if (key === "planning_horizon") { horizon(value, [key]); continue; }
+      if (key === "capability_gate") { candidates(value, [key]); continue; }
       if (key !== "interaction_contract") {
         if (!REUSABLE_PACKET_FIELDS.has(key) && !legacyQuotaFields.includes(key)) paths.push([key]);
         continue;

tests/control_plane/test_context_delivery_cli.py

@@ -180,7 +180,7 @@ def test_cli_capture_exact_delta_and_same_capture_recovery(tmp_path: Path, monke
     body = "Preserve every source row and cancellation condition. " * 500
     body += "FINAL CONDITION: stop before publication."
     todos = read_fixture_todos(state)
-    todos[0].update(text=body, title=body)
+    todos[0].update(text=body, title=body, required_capabilities=["shell"])
     initialize_canonical_authority(runtime, GOAL_ID,
         build_todo_runtime_shadow_projection(goal_id=GOAL_ID, todos=todos, handoff_mode="soft_claim"),
         state_path=state, provider=provider)
@@ -227,6 +227,13 @@ def test_cli_capture_exact_delta_and_same_capture_recovery(tmp_path: Path, monke
         delta = prepare_context_delivery(projection, scope=scope, capture_id="whole-b", surface=surface,
             baseline=first, retained={"complete": True, "target": first["target"]})
         assert delta["mode"] == "delta"
+        if surface == "quota_packet":
+            candidates = projection["capability_gate"]["runnable_candidates"]
+            assert any(row["todo_id"] == TODO_ID and row["title"] == body for row in candidates)
+            # The real producer retains every candidate in the cold input. Its
+            # unchanged full body must not recur in the warm model input.
+            assert first["context"]["capability_gate"] == projection["capability_gate"]
+            assert body not in json.dumps(delta["changes"], ensure_ascii=False)
         restored = restore_context_delivery(delta, baseline=first, current=next_capture)
         assert restored["mode"] == "delta" and restored["context"] == projection
         missing = restore_context_delivery(delta, baseline=None, current=next_capture)

tests/control_plane/test_progressive_native_context.py

@@ -36,6 +36,10 @@ def outcome(document):
 def test_cold_warm_and_repeated_compaction_keep_requirements_without_replaying_material(tmp_path, monkeypatch):
     event = setup(tmp_path, monkeypatch)
     original = payload()
+    candidate_text = "Preserve this candidate's complete final constraint. " * 120
+    original["capability_gate"] = {"schema_version": "capability_gate_v0", "action": "run",
+        "runnable_candidates": [{"schema_version": "todo_item_v0", "todo_id": "todo_work",
+            "text": candidate_text, "status": "open", "claimed_by": "agent"}], "blocked_candidates": []}
     _, first = project(tmp_path, payload=original)
     view = first["context"]
     assert first["mode"] == "full" and view["stop"] == original["stop"]
@@ -61,6 +65,7 @@ def test_cold_warm_and_repeated_compaction_keep_requirements_without_replaying_m
     assert context._state(tmp_path)["candidate"]["capture_file"] == first["recovery"]["path"]
 
     _, warm = project(tmp_path, "warm", payload=original)
+    assert warm["mode"] == "delta" and candidate_text not in json.dumps(warm["changes"])
     assert warm["target"]["context_epoch"] == first["target"]["context_epoch"]
     deliver(tmp_path, event, read_pages(tmp_path, warm))
     before = context._state(tmp_path)
@@ -72,6 +77,7 @@ def test_cold_warm_and_repeated_compaction_keep_requirements_without_replaying_m
         assert "Previous failure" not in json.dumps(result)
     _, fresh = project(tmp_path, "fresh", payload={**original, "stop": "Stop all delivery."})
     assert fresh["mode"] == "full" and fresh["context"]["stop"] == "Stop all delivery."
+    assert fresh["context"]["capability_gate"] == original["capability_gate"]
     assert json.loads(Path(archive["path"]).read_bytes()) == original
 
 

tests/control_plane_ts/context_delivery.test.ts

@@ -74,6 +74,70 @@ function warm(value: JsonObject, baseline: JsonObject): JsonObject {
 const recover = (delivery: JsonObject, baseline: JsonObject, value: JsonObject): JsonObject =>
   restoreContextDelivery({delivery, baseline, current: full(value, "capture-2")});
 
+test("candidate requirements reuse exact retained text while eligibility remains current", () => {
+  const row = (id: string) => ({schema_version: "todo_item_v0", todo_id: id,
+    text: `Complete ${id}; preserve the tail condition. `.repeat(120),
+    title: `Full objective for ${id}. `.repeat(80), note: "Evidence is not acceptance. ".repeat(80),
+    content_revision: "revision-a", status: "open", claimed_by: "agent",
+    required_capabilities: ["shell"], missing_capabilities: [], capability_action: "run",
+    unknown_control: {stop: false}, handoff_note: {intent: "implement", evidence_refs: ["evidence:a"]}});
+  const original = {capability_gate: {schema_version: "capability_gate_v0", action: "run",
+    runnable_candidates: [row("one"), row("two")], blocked_candidates: [row("blocked")],
+    resolution_bindings: [{capability: "shell", owner: "user", blocked_todo_ids: ["blocked"]}]}};
+  const saved = structuredClone(original);
+  const args = {scope, surface: "quota_packet", capture_id: "current"};
+  const baseline = prepareContextDelivery({...args, context: original, capture_id: "base"});
+  assert.deepEqual(baseline.context, original); // Cold still contains every candidate and full body.
+  const project = (context: JsonObject, extra = {}) => prepareContextDelivery({...args, context,
+    baseline, retained: {complete: true, target: baseline.target}, ...extra});
+  const delta = project(original);
+  assert.equal(delta.mode, "delta");
+  for (const lane of ["runnable_candidates", "blocked_candidates"] as const) {
+    for (const [i, candidate] of original.capability_gate[lane].entries()) {
+      for (const [key, value] of Object.entries(candidate)) {
+        const visible = inline(delta, ["capability_gate", lane, i, key]);
+        if (["text", "title", "note"].includes(key)) assert.equal(visible, undefined);
+        else assert.deepEqual(visible, value);
+      }
+    }
+  }
+  assert.deepEqual(inline(delta, ["capability_gate", "resolution_bindings"]), original.capability_gate.resolution_bindings);
+  assert.ok(JSON.stringify(delta).length < JSON.stringify(baseline).length / 2);
+  const changed = structuredClone(original);
+  changed.capability_gate.action = "ask_owner";
+  changed.capability_gate.runnable_candidates[0].text += "New final condition: do not execute.";
+  changed.capability_gate.runnable_candidates[0].claimed_by = "peer";
+  changed.capability_gate.runnable_candidates[0].unknown_control.stop = true;
+  const moved = structuredClone(original);
+  moved.capability_gate.blocked_candidates.push(moved.capability_gate.runnable_candidates.shift()!);
+  moved.capability_gate.runnable_candidates.reverse();
+  const removed = structuredClone(original); removed.capability_gate.runnable_candidates = [];
+  for (const current of [original, changed, moved, removed]) {
+    const delivery = project(current), full = prepareContextDelivery({...args, context: current});
+    assert.deepEqual(restoreContextDelivery({delivery, baseline, current: full}).context, current);
+  }
+  assert.equal(inline(project(changed), ["capability_gate", "action"]), "ask_owner");
+  assert.equal(inline(project(changed), ["capability_gate", "runnable_candidates", 0, "text"]),
+    changed.capability_gate.runnable_candidates[0].text);
+  for (const retention of [undefined, {complete: false, target: baseline.target}])
+    assert.deepEqual(project(original, {retained: retention}).context, original);
+  assert.equal(project(original, {scope: {...scope, context_epoch: "after-compaction"}}).mode, "full");
+  // Unknown or ambiguous identity cannot authorize reuse of a candidate body.
+  const unknownGate = structuredClone(original); unknownGate.capability_gate.schema_version = "future";
+  const unknownRow = structuredClone(original); unknownRow.capability_gate.runnable_candidates[0].schema_version = "future";
+  const duplicate = structuredClone(original); duplicate.capability_gate.runnable_candidates[1].todo_id = "one";
+  const malformed = structuredClone(original) as JsonObject;
+  ((malformed.capability_gate as JsonObject).runnable_candidates as JsonObject[])[0].text = {stop: true};
+  for (const current of [unknownGate, unknownRow, duplicate, malformed]) {
+    const gate = current.capability_gate as JsonObject, first = (gate.runnable_candidates as JsonObject[])[0];
+    const delivery = project(current);
+    const visible = delivery.mode === "full" ? (delivery.context as JsonObject).capability_gate
+      : {runnable_candidates: [{text: inline(delivery, ["capability_gate", "runnable_candidates", 0, "text"])}]};
+    assert.deepEqual(((visible as JsonObject).runnable_candidates as JsonObject[])[0].text, first.text);
+  }
+  assert.deepEqual(original, saved);
+});
+
 test("scoped preferences reuse complete prose while every current lifecycle fact stays inline", () => {
   const preferences = {ok: true, status: "read", current: {schema_version: "agent_preferences_v1",
     revision: "revision-a", authority: "advisory_only",