fix(control-plane): bound large quota planning requests

docs/architecture/rfcs/typescript-control-plane-migration-v0.md

@@ -1450,6 +1450,16 @@ unknown-effect policy are unchanged; decoded expansion uses the existing
 64 MiB local-snapshot bound before copying records. This is internal transport,
 not a new capability, setting, authority source or installed-runtime claim.
 
+The resume-planning input uses a co-deployed v1 record table for repeated source
+lanes. It retains every v0 payload fact before typed decisions, rather than a
+consumer-specific field allowlist or a second Python restoration pass. Existing
+v0 callers remain valid; conflicting same-id facts are not merged. Both quota
+planning and its subsequent Vision-wait projection opt into the existing private,
+digest-verified local snapshot transport for unique histories beyond 2 MiB;
+the 64 MiB snapshot/expanded-response bounds and ordinary RPC limits remain.
+File/SQLite qualification must reach the public quota result, preserve the source
+and select tail work, not stop after the first oversized request is repaired.
+
 Current bounded delivery: shared-goal alignment and amendment admission use one
 `shared_goal_work_source.py` snapshot per decision, reusing the canonical Todo
 summary after promotion. The same provider read optionally supplies leases at

docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md

@@ -1045,6 +1045,14 @@ lane 正文超过 2 MiB 响应边界、从而阻断启动及 Goal 增长后的
 64 MiB 展开边界。这是内部传输，不新增 capability、设置、授权源，也不代表安装态
 已采用此修改。
 
+Resume-planning 输入采用协同发布的 v1 record table，供重复 source lane 引用。
+每条记录在 TS 决策前保留全部 v0 payload 事实，不使用面向某个消费者的字段白名单，
+也不再由 Python 二次恢复。已有 v0 调用者仍有效；相同 id 的冲突事实不会合并。
+Quota planning 及其后续 Vision-wait 投影均显式复用既有私有、摘要校验的本地
+snapshot 通道，承载超过 2 MiB 的独立历史；64 MiB snapshot／响应展开边界和
+普通 RPC 上限不变。File／SQLite 验证必须到达公开 quota 结果，保留来源并选中
+尾部工作，不能只以第一层超大请求得到修复作为完整交付证据。
+
 当前有边界交付：shared-goal alignment 与 amendment admission 每次决策共用一份
 `shared_goal_work_source.py` 快照，promotion 后复用 canonical Todo summary；同一次
 provider 读取可返回同 revision 的 lease。缺失／空／陈旧展示及旧 lease 文件不再是

loopx/control_plane/effect_runtime.py

@@ -43,6 +43,8 @@
 MAX_LOCAL_SNAPSHOT_BYTES = 64 * 1024 * 1024
 LOCAL_SNAPSHOT_METHODS = frozenset({
     "todo.read",
+    "todo.quota_planning.project",
+    "goal.vision_wait.project",
     "collaboration.inbox.read",
     "collaboration.delegation.inventory_page",
     "benchmark.trajectory.codex-usage",

loopx/control_plane/effect_runtime_server.ts

@@ -29,6 +29,8 @@ const MAX_INLINE_RESPONSE_BYTES = 2 * 1024 * 1024;
 // Explicit opt-in: ordinary effects retain the 2 MiB request/response wire.
 const LOCAL_SNAPSHOT_METHODS = new Set([
   "todo.read",
+  "todo.quota_planning.project",
+  "goal.vision_wait.project",
   "collaboration.inbox.read",
   "collaboration.delegation.inventory_page",
   "benchmark.trajectory.codex-usage",

loopx/control_plane/goals/goal_vision_wait.py

@@ -6,7 +6,7 @@
 
 from ..effect_runtime import effect_runtime_result
 from ..todos.contract import normalize_todo_id
-from ..todos.resume_planning import build_todo_resume_planning_request
+from ..todos.resume_planning import build_referenced_todo_resume_planning_request
 
 GOAL_VISION_WAIT_STATE_SCHEMA_VERSION = "goal_vision_wait_state_v0"
 
@@ -100,8 +100,8 @@ def build_goal_vision_wait_state(
         "acceptance_gaps": acceptance_gaps or [],
         "selectable_advancement_count": selectable_advancement_count,
         "lineage_source_items": _vision_wait_lineage_rows(lineage_items),
-        "resume_planning_request": build_todo_resume_planning_request(planning_input, agent_id=agent_id),
-    })
+        "resume_planning_request": build_referenced_todo_resume_planning_request(planning_input, agent_id=agent_id),
+    }, large_local_snapshot=True)
     if result is not None and (
         not isinstance(result, dict)
         or result.get("schema_version") != GOAL_VISION_WAIT_STATE_SCHEMA_VERSION

loopx/control_plane/todos/quota_selection.py

@@ -21,7 +21,7 @@
     todo_item_task_class, todo_projection_sort_key,
     todo_summary_monitor_writeback_supported,
 )
-from .resume_planning import build_todo_resume_planning_request
+from .resume_planning import build_referenced_todo_resume_planning_request
 from .handoff_note import todo_read_source
 from .user_gate import is_user_gate_todo_item
 from .route_continuation import build_todo_route_continuation_facts
@@ -151,6 +151,12 @@ def active(key: str) -> list[dict[str, Any]]:
         raw = value.get(key)
         return [encode(item) for item in raw if isinstance(item, dict)] if isinstance(raw, list) else []
 
+    resume_request = build_referenced_todo_resume_planning_request(
+        value,
+        agent_id=agent,
+        item_limit=8,
+        available_capabilities=(available_capabilities or []) if resolve_capacity else None,
+    )
     try:
         wire = effect_runtime_result("todo.quota_planning.project", {
             "schema_version": "todo_quota_planning_request_v4",
@@ -161,8 +167,7 @@ def active(key: str) -> list[dict[str, Any]]:
                 "excluded": normalize_todo_excluded_agents(gate.get("excluded_agents"))}
                 for gate in handoff_gates],
             "source_contract": _closure_source_facts(value),
-            "resume": build_todo_resume_planning_request(value, agent_id=agent, item_limit=8,
-                available_capabilities=(available_capabilities or []) if resolve_capacity else None),
+            "resume": resume_request,
             "selection": {
                 "observed_at": observed_at,
                 "available": normalize_required_capabilities(available_capabilities),
@@ -177,7 +182,7 @@ def active(key: str) -> list[dict[str, Any]]:
                 "frontier_revision_index": value.get("advancement_frontier_revision_index"),
                 "diagnostic_limit": 3, "backlog_limit": 8, "visibility_limit": 16,
             },
-        })
+        }, large_local_snapshot=True)
     except EffectRuntimeRejected as exc:
         raise ValueError(str(exc)) from None
     result = _decode_quota_planning_wire(wire)

loopx/control_plane/todos/resume_planning.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import json
 from typing import Any
 
 from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result
@@ -20,6 +21,8 @@
     "current_agent_claimed_monitor_items", "claimed_monitor_open_items",
 )
 
+_REFERENCE_REQUEST_SCHEMA = "todo_resume_planning_request_v1"
+
 
 def _planning_item(item: dict[str, Any]) -> dict[str, Any]:
     payload = compact_todo_projection_item(
@@ -83,6 +86,45 @@ def build_todo_resume_planning_request(
     }
 
 
+def build_referenced_todo_resume_planning_request(
+    value: Any, *, agent_id: str | None = None, item_limit: int = 5,
+    available_capabilities: Any = None,
+) -> dict[str, Any]:
+    """Encode one semantic snapshot without repeating full payloads per lane.
+
+    The ordinary v0 builder remains the compatibility contract for callers that
+    inspect or rewrite its source rows. The v1 wire changes only repeated lane
+    bodies: every record retains its complete v0 facts for all typed consumers.
+    """
+    request = build_todo_resume_planning_request(
+        value,
+        agent_id=agent_id,
+        item_limit=item_limit,
+        available_capabilities=available_capabilities,
+    )
+    records: list[dict[str, Any]] = []
+    references: dict[str, int] = {}
+    sources: dict[str, list[int]] = {}
+    for source_name, source_rows in request["sources"].items():
+        source_refs: list[int] = []
+        for row in source_rows:
+            # Equal normalized records are semantically interchangeable.  Use
+            # their exact JSON form instead of Todo id alone so conflicting
+            # historical facts remain separate and are still diagnosed.
+            identity = json.dumps(
+                row, ensure_ascii=False, sort_keys=True, separators=(",", ":")
+            )
+            reference = references.get(identity)
+            if reference is None:
+                reference = len(records)
+                references[identity] = reference
+                records.append(row)
+            source_refs.append(reference)
+        sources[source_name] = source_refs
+    return {**request, "schema_version": _REFERENCE_REQUEST_SCHEMA,
+            "sources": sources, "records": records}
+
+
 def project_todo_resume_planning(
     value: Any, *, agent_id: str | None = None, item_limit: int = 5,
     available_capabilities: Any = None,

loopx/control_plane/todos/resume_planning.ts

@@ -12,6 +12,7 @@ import {
 } from "./resume_condition.ts";
 
 export const RESUME_PLANNING_REQUEST = "todo_resume_planning_request_v0";
+export const REFERENCED_RESUME_PLANNING_REQUEST = "todo_resume_planning_request_v1";
 export const RESUME_PLANNING_RESULT = "todo_resume_planning_v0";
 const DEFERRED_POLICY = "quota may wake the current peer only for ready deferred todos " +
   "claimed by that agent or unclaimed; other-agent deferred todos remain " +
@@ -186,12 +187,18 @@ function resolveCapacity(items: readonly Item[], source: readonly Item[], capabi
   });
 }
 
-function decodeSources(value: unknown): Record<SourceKey, Item[]> {
+function decodeSources(value: unknown, records: Item[] | null = null): Record<SourceKey, Item[]> {
   const rawSources = requireJsonObject(value, "sources");
   const sources = {} as Record<SourceKey, Item[]>;
   for (const key of SOURCE_KEYS) {
     if (!Array.isArray(rawSources[key])) throw new EffectRuntimeRequestError(`${key} must be an array`);
-    sources[key] = rawSources[key].map(item);
+    sources[key] = records === null ? rawSources[key].map(item) : rawSources[key].map((value) => {
+      const reference = requireInteger(value, `${key} record reference`);
+      if (reference < 0 || reference >= records.length) {
+        throw new EffectRuntimeRequestError(`${key} record reference is out of range`);
+      }
+      return records[reference];
+    });
   }
   return sources;
 }
@@ -295,10 +302,19 @@ function blockedVisibility(resumeBlocked: Item[], monitorItems: Item[], options:
  * No mutation, claim, lease grant, monitor poll, or recovery effect is emitted. */
 export function projectTodoResumePlanning(value: unknown): JsonObject {
   const request = requireJsonObject(value, "resume planning request");
-  if (request.schema_version !== RESUME_PLANNING_REQUEST) {
+  if (request.schema_version !== RESUME_PLANNING_REQUEST &&
+      request.schema_version !== REFERENCED_RESUME_PLANNING_REQUEST) {
     throw new EffectRuntimeRequestError("resume planning schema mismatch");
   }
-  const sources = decodeSources(request.sources);
+  const records = request.schema_version === REFERENCED_RESUME_PLANNING_REQUEST
+    ? (() => {
+      if (!Array.isArray(request.records)) {
+        throw new EffectRuntimeRequestError("resume planning records must be an array");
+      }
+      return request.records.map(item);
+    })()
+    : null;
+  const sources = decodeSources(request.sources, records);
   diagnoseSources(sources);
   const display: DisplayOptions = {
     agent: optionalNonEmptyString(request.agent_id, "agent_id"),

tests/control_plane/test_quota_selection.py

@@ -1,6 +1,7 @@
 """Quota scope is distinct from execution ownership; all data are synthetic."""
 import pytest
 import json
+from copy import deepcopy
 from pathlib import Path
 
 from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime, empty_todo_display
@@ -58,6 +59,138 @@ def test_agent_execution_still_respects_claim_and_exclusion():
     assert result["claim_scope"]["other_agent_claimed_open_count"] == 1
 
 
+@pytest.mark.parametrize("identity_field", ["target_key", "action_kind", "title"])
+def test_quota_frontier_keeps_identity_without_a_typed_todo_id(identity_field):
+    row = item(None, task_class="advancement_task", resume_when="resume_at:2099-01-01T00:00:00Z",
+               resume_ready=False, next_due_at="2099-01-01T00:00:00Z",
+               **{identity_field: "synthetic-frontier-identity"})
+    result = summarize_user_todos_for_quota(summary([row]),
+        agent_identity={"agent_id": "agent-b"})
+    assert result["frontier_deadline"]["identity"] == "synthetic-frontier-identity"
+
+
+@pytest.mark.parametrize("entrypoint", ["summary", "planning"])
+def test_quota_returned_records_do_not_alias_source_or_other_lanes(entrypoint):
+    from loopx.control_plane.todos.quota_selection import project_quota_planning
+
+    row = item("todo_wait", task_class="advancement_task", required_capabilities=["shell"],
+        resume_when="todo_done:todo_dependency", resume_ready=False,
+        resume_condition={"kind": "todo_done", "satisfied": False,
+            "target_todo_id": "todo_dependency", "target_status": "open",
+            "target_task_class": "advancement_task"})
+    source = summary([row])
+    before = deepcopy(source)
+    if entrypoint == "summary":
+        result = summarize_user_todos_for_quota(source, agent_identity={"agent_id": "agent-b"})
+        output = result["resume_blocked_items"][0]
+        other = result["first_open_items"][0]
+    else:
+        result = project_quota_planning(source, all_open_items=[row], source_open_count=1,
+            agent_identity={"agent_id": "agent-b"}, filter_user_gate_blocks_agent=False,
+            available_capabilities=[], current_time="2026-10-01T00:00:00Z")
+        resume = result["resume_planning"]
+        output = resume["resume_blocked_lanes"]["resume_blocked_items"][0]
+        other = resume["blocked_successor_items"][0]
+    output["required_capabilities"].append("synthetic-change")
+    output["resume_condition"]["target_status"] = "done"
+    assert source == before
+    assert other["required_capabilities"] == ["shell"]
+    assert other["resume_condition"]["target_status"] == "open"
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_large_history_completes_public_quota_through_vision_wait(tmp_path, monkeypatch, provider):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    runtime, registry, state = tmp_path / "runtime", tmp_path / "registry.json", tmp_path / "state.md"
+    records = [item(f"todo_history_{index}", role="agent", source_section="Agent Todo",
+        schema_version="todo_item_v0", archive_state="active", done=True, status="done",
+        task_class="advancement_task", text=f"Completed synthetic work {index}. " * 60,
+        index=index + 1) for index in range(2500)]
+    records.append(item("todo_tail", role="agent", source_section="Agent Todo",
+        schema_version="todo_item_v0", archive_state="active", done=False,
+        task_class="advancement_task", claimed_by="agent-b", index=2501))
+    state.write_text("---\nstatus: active\n---\n# Goal\n\n## Objective\n"
+                     "Deliver a checked change.\n\n## Next Action\nComplete the remaining work.\n")
+    write_fixture_registry(project=tmp_path, runtime_root=runtime, registry_path=registry,
+        goal_id="goal-history", domain="quota-history", adapter_kind="generic_project_goal_v0",
+        state_file=str(state), registered_agents=["agent-b"], quota_allowed_slots=None)
+    projection = build_todo_runtime_shadow_projection(
+        goal_id="goal-history", todos=records, handoff_mode="soft_claim")
+    initialize_canonical_authority(runtime, "goal-history", projection, state_path=state, provider=provider)
+    before = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-history")
+    narrative = state.read_bytes()
+    code, packet = run_json_cli_result("quota", "should-run", "--goal-id", "goal-history",
+        "--agent-id", "agent-b", registry_path=registry, runtime_root=runtime)
+    assert code == 0, packet
+    assert packet["agent_todo_summary"]["total_count"] == 2501
+    assert packet["selected_todo"]["todo_id"] == "todo_tail"
+    assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-history") == before
+    assert state.read_bytes() == narrative
+
+
+@pytest.mark.parametrize("gate_scope", [False, True])
+def test_large_quota_history_uses_snapshot_without_losing_scope(
+    monkeypatch, gate_scope,
+):
+    from loopx.control_plane import effect_runtime
+
+    observed_sizes = []
+    original = effect_runtime._request_with_info
+
+    def observe(info, **kwargs):
+        if kwargs["method"] == "todo.quota_planning.project":
+            observed_sizes.append(len(json.dumps(
+                kwargs["params"], separators=(",", ":")
+            ).encode()))
+        return original(info, **kwargs)
+
+    monkeypatch.setattr(effect_runtime, "_request_with_info", observe)
+    history = [
+        item(
+            f"todo_history_{index}",
+            status="done",
+            task_class="advancement_task",
+            note="Synthetic historical evidence. " * 180,
+            index=index + 1,
+        )
+        for index in range(2500)
+    ]
+    tail = [
+        item("todo_tail_gate", blocks_agent="agent-b", index=2501),
+        item(
+            "todo_peer", task_class="advancement_task",
+            claimed_by="agent-a", index=2502,
+        ),
+        item(
+            "todo_owned", task_class="advancement_task",
+            claimed_by="agent-b", index=2503,
+        ),
+    ]
+    source = summary(history + tail)
+    source.update(
+        done_count=len(history), open_count=len(tail), first_open_items=tail,
+    )
+
+    result = summarize_user_todos_for_quota(
+        source,
+        agent_identity={"agent_id": "agent-b"},
+        filter_user_gate_blocks_agent=gate_scope,
+    )
+
+    # Unique historical payloads cannot be reduced below the inline wire limit
+    # without dropping facts. Use the existing digest-verified local snapshot
+    # transport rather than raising or truncating.
+    assert observed_sizes and max(observed_sizes) > effect_runtime.MAX_REQUEST_BYTES
+    assert [row["todo_id"] for row in result["gate_open_items"]] == [
+        "todo_tail_gate"
+    ]
+    assert [row["todo_id"] for row in result["first_executable_items"]] == (
+        [] if gate_scope else ["todo_owned"]
+    )
+    if not gate_scope:
+        assert result["claim_scope"]["other_agent_claimed_open_count"] == 1
+
+
 @pytest.mark.parametrize("agent", [None, "agent-b"])
 def test_route_replan_visibility_counts_before_budget_without_granting_execution(agent):
     own = [item(f"todo_route_{i}", task_class="advancement_task", claimed_by="agent-b",