refactor(quota): compose quota summary in one TypeScript planning crossing

docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-11-02-quota-summary-transaction.md

@@ -0,0 +1,22 @@
+# Quota summary transaction cutover (2026-11-02)
+
+The existing `todo.quota_planning.project` request now returns the complete
+user or project-asset quota summary. TypeScript owns source selection, lane
+composition, counts, limits, warnings and display projection in process.
+Python gathers compatibility facts and validates and dereferences the response.
+Source counts, metadata, selected Vision proofs, closure overlays and warning
+fields retain their original Python values through owner-selected references;
+planning and display rows retain their existing JSON normalization.
+
+The pre-cutover complete-output fixtures remain the parity guard. For both
+summary variants, every matrix path previously using 1–5 crossings now uses
+one; non-dict input stays at zero and owner refusal stays at one without fallback.
+No handler, authority or persisted format is added. The planning-only Python
+codec remains for direct planning consumers; remove it only after their last
+caller migrates. Payload/status compaction and source helpers retain other
+callers. No facade exit or latency improvement is claimed.
+
+The earlier Phase 1 transport-loss finding remains valid for plain JSON;
+source references resolve that blocker without changing the pinned outputs.
+This source-checkout slice does not qualify installed or long-horizon benefit.
+Rollback is the code diff; no state migration is needed.

docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-11-02-quota-summary-transaction.zh-CN.md

@@ -0,0 +1,17 @@
+# 配额摘要事务切换（2026-11-02）
+
+现有 `todo.quota_planning.project` 请求现在返回完整的用户或项目资产配额摘要。
+TypeScript 在进程内负责来源选择、泳道组合、计数、限制、警告和显示投影。
+Python 收集兼容性事实，并校验响应、解析引用。
+来源计数、元数据、选中的 Vision 证明、闭合意图叠加及警告字段通过由所有者
+选择的引用保留原始 Python 值；规划和显示行保留现有的 JSON 归一化行为。
+
+切换前的完整输出夹具继续作为等价性守卫。两种摘要的矩阵路径中，原先使用
+1–5 次跨运行时调用的路径现在均使用一次；非字典输入保持零次，所有者拒绝
+保持一次且不回退。不增加处理器、权限或持久化格式。仅供规划使用的 Python
+编解码器仍有直接调用者；只有最后一个调用者迁移后才能删除。载荷和状态压缩
+以及来源辅助函数仍有其他调用者。不宣称外观层退出或延迟改善。
+
+先前第一阶段关于纯 JSON 传输损失的发现仍然成立；来源引用在不改变固定输出
+的前提下解决了该阻塞。本源码检出切片不证明已安装行为或长程收益。
+回滚只需撤销代码差异，无需状态迁移。

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

@@ -2835,3 +2835,4 @@ The native query now owns recovery lookup and completion capability evidence,
 with shared write/read status-phase checks. See the [2026-10-02 checkpoint](ledger/typescript-control-plane-migration-v0/2026-10-02-turn-journal-readback.md) for compatibility, measured cost, validation boundaries and facade exit.
 
 - Summary-display final-head qualification and preserved tail cohorts: [2026-10-27 follow-up](ledger/typescript-control-plane-migration-v0/2026-10-27-todo-summary-qualification.md).
+- Quota summary composition in one `todo.quota_planning.project` crossing with source references: [2026-11-02 cutover](ledger/typescript-control-plane-migration-v0/2026-11-02-quota-summary-transaction.md).

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

@@ -2100,3 +2100,4 @@ SQLite 中位数略升，收益不能外推为全面加速。这只资格化本
 原生查询统一拥有恢复定位和 completion capability 证据，并共享写入／回读的状态阶段约束。兼容变更、实测代价、验证范围与 facade 退出条件见 [2026-10-02 检查点](ledger/typescript-control-plane-migration-v0/2026-10-02-turn-journal-readback.zh-CN.md)。
 
 - 摘要展示最终 head 验收与历史尾延迟批次：[2026-10-27 后续记录](ledger/typescript-control-plane-migration-v0/2026-10-27-todo-summary-qualification.zh-CN.md)。
+- 配额摘要经来源引用在一次 `todo.quota_planning.project` 跨运行时调用内完成组合：[2026-11-02 切换记录](ledger/typescript-control-plane-migration-v0/2026-11-02-quota-summary-transaction.zh-CN.md)。

loopx/control_plane/todos/context_projection.ts

@@ -238,7 +238,7 @@ function compactSummaryRecord(item: JsonObject, body: unknown): JsonObject {
 }
 
 /** Standalone compatibility display, also composed inside quota planning. */
-export function projectTodoSummaryDisplay(input: JsonObject): JsonObject[] {
+export function projectTodoSummaryDisplay(input: JsonObject, onSource?: (key: string, index: number) => void): JsonObject[] {
   const planning = input.include_terminal !== undefined, terminal = input.include_terminal === true;
   const source = input.source === undefined ? null : decodeTodoReadSource(requireJsonObject(input.source, "summary source"));
   const ready = new Set((Array.isArray(input.handoff_gates) ? input.handoff_gates : []).flatMap(value => {
@@ -248,7 +248,7 @@ export function projectTodoSummaryDisplay(input: JsonObject): JsonObject[] {
   const selected: {item: JsonObject; text: unknown}[] = [], seen = new Set<string>();
   if (source) {
     for (const key of planning ? PLANNING_SOURCES : SUMMARY_SOURCES) {
-      for (const value of Array.isArray(source[key]) ? source[key] as unknown[] : []) {
+      for (const [index, value] of (Array.isArray(source[key]) ? source[key] as unknown[] : []).entries()) {
         if (!value || typeof value !== "object" || Array.isArray(value)) continue;
         const item = value as JsonObject, body = stripPythonWhitespace(pythonFieldText(item, "text"));
         let todo = id(item.todo_id);
@@ -257,6 +257,7 @@ export function projectTodoSummaryDisplay(input: JsonObject): JsonObject[] {
         if (!planning && selected.some(({item: prior, text}) => item.todo_id && prior.todo_id ? item.todo_id === prior.todo_id : prior.index === item.index && text === body)) continue;
         if (todo) seen.add(todo);
         selected.push({item, text: body});
+        onSource?.(key, index);
       }
     }
   } else {

loopx/control_plane/todos/quota_selection.py

@@ -19,10 +19,11 @@
     todo_item_has_removed_continuation_policy, todo_item_is_actionable_open,
     todo_item_next_due_at, todo_item_expires_at, todo_item_is_watch_only_monitor,
     todo_item_task_class, todo_projection_sort_key,
-    todo_summary_monitor_writeback_supported,
+    todo_summary_monitor_writeback_supported, todo_presentation_sort_key,
 )
 from .resume_planning import build_referenced_todo_resume_planning_request
 from .handoff_note import todo_read_source
+from .summary_item import TODO_SUMMARY_SOURCE_KEYS
 from .user_gate import is_user_gate_todo_item
 from .route_continuation import build_todo_route_continuation_facts
 from .handoff_gate import todo_summary_handoff_gates
@@ -112,6 +113,7 @@ def project_quota_planning(
     source_open_count: Any, agent_identity: dict[str, Any] | None,
     filter_user_gate_blocks_agent: bool, available_capabilities: Any,
     resolve_capacity: bool = False, current_time: str | None = None,
+    summary_kind: str | None = None,
 ) -> dict[str, Any]:
     identity = agent_identity if isinstance(agent_identity, dict) else {}
     profile = identity.get("agent_profile")
@@ -123,11 +125,12 @@ def project_quota_planning(
     observed_at = observed_now.timestamp()
     handoff_gates = todo_summary_handoff_gates(value)
 
-    def encode(item: dict[str, Any]) -> dict[str, Any]:
+    def encode(item: dict[str, Any], *, include_payload: bool = True) -> dict[str, Any]:
         priority, index = todo_projection_sort_key(item)
         due, expires = todo_item_next_due_at(item), todo_item_expires_at(item)
         return {
-            "payload": item, "display_source": todo_read_source(item),
+            **({"payload": item, "display_source": todo_read_source(item)} if include_payload else {}),
+            "presentation": list(todo_presentation_sort_key(item)),
             "claim": normalize_todo_claimed_by(item.get("claimed_by")),
             "bound": normalize_todo_bound_agent(item.get("bound_agent")),
             "blocks": normalize_todo_blocks_agent(item.get("blocks_agent")),
@@ -151,6 +154,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 []
 
+    def integer_text(raw: Any) -> str | None:
+        try:
+            return str(int(raw)) if raw is not None else None
+        except (TypeError, ValueError):
+            return None
+
     resume_request = build_referenced_todo_resume_planning_request(
         value,
         agent_id=agent,
@@ -161,6 +170,19 @@ def active(key: str) -> list[dict[str, Any]]:
         wire = effect_runtime_result("todo.quota_planning.project", {
             "schema_version": "todo_quota_planning_request_v4",
             "current_time": observed_now.isoformat(),
+            **({"summary_request": {
+                "kind": summary_kind, "source": todo_read_source(value),
+                "integer_counts": {"source": integer_text(value.get("completed_without_successor_count")),
+                    "warning": integer_text((value.get("todo_succession_warning") or {}).get("count"))
+                    if isinstance(value.get("todo_succession_warning"), dict) else None},
+                "next_fact": encode({"index": value.get("next_index", 1),
+                    "text": str(value.get("next") or "").strip(),
+                    **({"claimed_by": str(value.get("next_claimed_by") or "").strip()}
+                       if str(value.get("next_claimed_by") or "").strip() else {})}),
+                "source_rows": {key: [encode(row, include_payload=False) if isinstance(row, dict) else None
+                                      for row in value.get(key, [])]
+                                for key in TODO_SUMMARY_SOURCE_KEYS if isinstance(value.get(key), list)},
+            }} if summary_kind else {}),
             "frontier_deadline": value.get("frontier_deadline"),
             "route_items": build_todo_route_continuation_facts(value, handoff_gates=handoff_gates),
             "handoff_items": [{"display": gate,
@@ -185,6 +207,8 @@ def active(key: str) -> list[dict[str, Any]]:
         }, large_local_snapshot=True)
     except EffectRuntimeRejected as exc:
         raise ValueError(str(exc)) from None
+    if summary_kind:
+        return _decode_quota_summary_wire(wire, value)
     result = _decode_quota_planning_wire(wire)
     if not isinstance(result, dict) or result.get("schema_version") != "todo_quota_planning_v0":
         raise RuntimeError("TypeScript Todo quota planning shape mismatch")
@@ -197,3 +221,67 @@ def active(key: str) -> list[dict[str, Any]]:
     elif result["closure_intent"] is not None:
         raise RuntimeError("TypeScript Todo quota planning closure intent shape mismatch")
     return result
+
+
+def _decode_quota_summary_wire(wire: Any, source: dict[str, Any]) -> dict[str, Any]:
+    """Resolve owner-selected transport references after bounded validation."""
+    error = "invalid TypeScript quota summary wire"
+    if not isinstance(wire, dict) or wire.get("schema_version") != "todo_quota_summary_wire_v1":
+        raise RuntimeError(error)
+    summary, copies, integers = wire.get("summary"), wire.get("source_copies"), wire.get("integer_copies")
+    if not isinstance(summary, dict) or not isinstance(copies, list) or not isinstance(integers, list):
+        raise RuntimeError(error)
+    expanded = len(json.dumps(summary, ensure_ascii=False, separators=(",", ":")).encode())
+    patches: list[tuple[Any, Any, Any]] = []
+    targets: set[tuple[Any, ...]] = set()
+
+    def path(value: Any, *, nonempty: bool = False) -> list[Any]:
+        if (not isinstance(value, list) or (nonempty and not value)
+                or any(type(key) not in (str, int) or isinstance(key, int) and key < 0 for key in value)):
+            raise RuntimeError(error)
+        return value
+
+    def child(parent: Any, key: Any) -> Any:
+        if not (isinstance(parent, dict) and isinstance(key, str)
+                or isinstance(parent, list) and type(key) is int):
+            raise RuntimeError(error)
+        return parent[key]
+
+    try:
+        for copy in [*copies, *integers]:
+            if not isinstance(copy, dict):
+                raise RuntimeError(error)
+            target_path = path(copy.get("target"), nonempty=True)
+            coordinate = tuple(target_path)
+            if coordinate in targets:
+                raise RuntimeError(error)
+            targets.add(coordinate)
+            parent: Any = summary
+            for key in target_path[:-1]:
+                parent = child(parent, key)
+            if child(parent, target_path[-1]) is not None:
+                raise RuntimeError(error)
+            if "source" in copy:
+                original: Any = source
+                for key in path(copy["source"]):
+                    original = child(original, key)
+            else:
+                raw = copy.get("value")
+                if not isinstance(raw, str) or not raw or raw.lstrip("-").strip("0123456789"):
+                    raise RuntimeError(error)
+                original = int(raw)
+            expanded += len(json.dumps(original, ensure_ascii=False, separators=(",", ":")).encode()) - 4
+            if expanded > MAX_LOCAL_SNAPSHOT_BYTES:
+                raise RuntimeError("TypeScript quota summary expanded response is oversized")
+            patches.append((parent, target_path[-1], original))
+    except (ValueError, KeyError, IndexError, TypeError):
+        raise RuntimeError(error) from None
+    if expanded > MAX_LOCAL_SNAPSHOT_BYTES:
+        raise RuntimeError("TypeScript quota summary expanded response is oversized")
+    for parent, key, original in patches:
+        parent[key] = original
+    if (not isinstance(summary.get("work_counts"), dict)
+            or any(not isinstance(summary.get(key), list) for key in (
+                "first_open_items", "first_executable_items", "monitor_open_items", "backlog_items"))):
+        raise RuntimeError(error)
+    return summary