refactor(todos): compose full readback in one TypeScript transaction

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

@@ -4,7 +4,7 @@
 - Supersedes / closes: none
 - Proposed by: LoopX maintainers
 - Date: 2026-08-15
-- Last revised: 2026-10-16
+- Last revised: 2026-10-19
 - Scope: an incremental, replacement-first migration of the LoopX control-plane
   core from Python to TypeScript without maintaining two semantic
   implementations
@@ -46,6 +46,31 @@ Each page reuses the existing executable-revision request scope; operation
 records remain live reads, and the next page resolves the current code again.
 
 
+### Todo read transaction source checkpoint
+
+`todo.read` now composes canonical provider reads, full-source resume and
+succession evidence, standing decisions, both role summaries, selection and the
+public result through `todos/read_projection.ts` and `todos/read_transaction.ts`.
+Ordinary canonical queries use one request. Large result continuations retain
+one observation and result digest; changed results refuse rather than combining
+pages. Dependency and closure decisions still use the complete source.
+Python retains registry, document/material IO, independent Vision-wait consumers
+and compatibility exports. Its replaced summary, filter and result coordination
+is retired. Persisted completion spellings use shared token data and compatibility codecs
+without scalar RPCs;
+`todo.summary.project`, `todo.completion_state.normalize` and
+`todo.completion_state.require_metadata` are retired after their production
+callers exit. Existing semantic owners execute in process inside the transaction.
+
+CLI query results carry a query-scoped `projection_envelope`. A read grants no
+execution, acceptance, lease or mutation authority. The shared wire and private
+snapshot budgets are unchanged. Oversized Python facts use an owned temporary
+source, not another durable store; complete results use digest-bound pieces.
+Remaining historical/source adapters still name their actual IO and crossings.
+Source/controlled-IPC evidence does not qualify normal installed CLI/Host use,
+full-command performance, R3 adoption or D2/D3/default delivery.
+
+
 <a id="current-delivery-frontier-2026-10-02"></a>
 
 ## Current delivery frontier (post-merge, 2026-10-16)

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

@@ -4,7 +4,7 @@
 - 替代 / 关闭：无
 - Proposed by：LoopX maintainers
 - Date：2026-08-15
-- Last revised：2026-10-16
+- Last revised：2026-10-19
 - Scope：LoopX 控制面核心从 Python 到 TypeScript 的增量、replacement-first
   迁移；不长期维护两份语义实现
 - Tracking issue：[#3225](https://github.com/loopx-project/loopx/issues/3225)
@@ -39,6 +39,26 @@ Python 保留有界地址扫描、请求者注册检查和实时 `Delegations.re
 每页复用既有代码版本请求作用域；操作记录仍实时读取，下一页重新解析当前代码。
 
 
+### Todo 读取事务源码检查点
+
+`todo.read` 现由 `todos/read_projection.ts` 与 `todos/read_transaction.ts`
+组合 canonical provider 读取、完整源上的恢复与后继证据、standing decisions、
+两个角色的摘要、筛选与公共结果。普通 canonical 查询使用一次请求。
+大结果的连续页保留同一次观察和结果摘要；结果变化即拒绝，不拼接新旧页。
+依赖与闭合判断仍使用完整源。
+Python 保留 registry、文档／材料 IO、独立的 Vision-wait 消费者及兼容导出。
+被替代的摘要、筛选与结果协调已退役。持久化 completion 拼写使用共享 token 数据与兼容 codec，
+不再逐标量调用 RPC；实际生产调用者退出后，退役 `todo.summary.project`、
+`todo.completion_state.normalize` 与 `todo.completion_state.require_metadata`。
+事务在进程内执行既有语义 owner。
+
+CLI 查询结果携带查询范围内的 `projection_envelope`。读取不授予执行、验收、
+租约或修改权限。共享 wire 与私有快照预算不变。超大 Python 事实使用私有临时
+源，不增加持久化存储；完整结果通过摘要绑定的片段传输。仍保留的历史／源适配
+明确记录实际 IO 与跨运行时调用。源码／受控 IPC 证据不代表正常安装的 CLI／Host
+使用、完整命令性能、R3 采用或 D2／D3／默认交付资格。
+
+
 <a id="当前交付边界2026-10-02"></a>
 
 ## 当前交付边界（合入后，2026-10-16）

docs/reference/todo-work-counts.md

@@ -51,8 +51,9 @@ are lower bounds while `complete=false`. Current capability re-evaluation
 replaces prior readiness for observed rows and cannot certify unseen rows.
 
 The TypeScript `todos/summary_lanes.ts` owner returns indexes into the one input
-array instead of repeating full Todo bodies for each lane. The Python adapter
-normalizes legacy fields/timestamps and validates the returned ordinal bounds.
+array instead of repeating full Todo bodies for each lane. The native read owner
+normalizes canonical fields/timestamps and materializes the selected records.
+Python retains the legacy source codecs.
 One observation time governs Monitor due and expiry classification within the
 batch. Source order, completed/deferred conventions and claimant visibility
 remain compatible; `done_count` still includes deferred rows as required by its
@@ -92,11 +93,14 @@ read and continue to show the whole Goal; no new configuration editor is needed.
 Resume and succession are evaluated on the complete source before selection.
 The typed batch filters rows without renumbering their original source indexes,
 then builds lanes/counts, and only then applies display limits. Status/identity
-filters do not recompute dependencies from their smaller view. One internal `todo.summary.project` batch now composes selection, counts,
-visibility allocation and closure. Its transient request replaces the separate
-lane and closure RPC calls; it does not change persisted Todo or public summary
-schemas. Python decodes legacy input, validates source ordinals and materializes
-public fields, with no independent summary count, cap or claimant-allocation rule.
+filters do not recompute dependencies from their smaller view. The `todo.read` transaction composes source evaluation, selection, counts,
+frontier identity, handoff context and result construction through the existing
+TypeScript owners. Canonical provider IO runs in that transaction; Markdown
+source codecs and material IO remain explicit Python adapters. Filtering reuses
+validated full-source evidence, never dependencies inferred from a small view.
+The old `todo.summary.project` transport and Python ordinal materializer are
+retired. Large source/result transport preserves complete decisions and refuses
+changed continuations; display limits still disclose shown and available counts.
 
 Route-continuation visibility also uses the existing typed quota planning batch.
 It shares the normal claim/exclusion rule; eligibility, first-identity deduplication,

loopx/capabilities/periodic_report/post_writeback_hook.py

@@ -16,10 +16,6 @@
     build_goal_frontier_projection_from_summaries,
 )
 from .todo_source import read_report_source_history, read_report_todo_source
-from ...control_plane.effect_runtime import (
-    CANONICAL_AUTHORITY_READ_TIMEOUT_SECONDS,
-    effect_runtime_result,
-)
 from ...control_plane.todos.quota_summary import summarize_user_todos_for_quota
 from ...control_plane.todos.todo_index import MAX_TODO_INDEX_ROLLOUT_EVENTS_PER_GOAL
 from ...history import collect_history, load_index_snapshot, load_registry
@@ -202,18 +198,18 @@ def build_periodic_report_post_writeback_projection(
     if available_capabilities is None and isinstance(payload.get("turn"), Mapping):
         available_capabilities = payload["turn"].get("available_capabilities")
     if source_todos is None and "todo_source" in payload:
-        source_todos = effect_runtime_result(
-            "coordination.local_authority.todo_source",
-            {
-                "schema_version": "loopx_local_coordination_todo_source_request_v0",
-                "runtime_root": str(runtime_root), "goal_id": goal_id,
-                "source": payload["todo_source"],
-            },
-            timeout=CANONICAL_AUTHORITY_READ_TIMEOUT_SECONDS,
-            large_local_snapshot=True,
-        )
+        from ...control_plane.todos.read_projection import project_todo_read, restore_todo_field_evaluations
+        from ...control_plane.runtime.time import now_utc
+
+        now = now_utc()
+        readback = project_todo_read({"kind": "historical_fields",
+            "source": {"runtime_root": str(runtime_root), "goal_id": goal_id, "original_source": payload["todo_source"]},
+            "rollout_events": [], "available_capabilities": available_capabilities,
+            "observed_at": now.timestamp(), "evaluated_at": now.isoformat()})
+        source_todos = readback["canonical"]
         if source_todos.get("status") != "loaded":
             raise ValueError("original canonical Todo source is unavailable")
+        source_todos["_todo_fields"] = restore_todo_field_evaluations(readback)
         source_runs = read_report_source_history(
             runtime_root=runtime_root, goal_id=goal_id, run=payload,
         )
@@ -228,7 +224,7 @@ def build_periodic_report_post_writeback_projection(
         state.get("path") if isinstance(state, Mapping) else None
     ) or payload.get("state_file")
     if source_todos is not None:
-        fields = canonical_todo_summary_fields(
+        fields = source_todos.get("_todo_fields") or canonical_todo_summary_fields(
             source_todos["todos"],
             rollout_events=events,
             available_capabilities=available_capabilities,

loopx/capabilities/periodic_report/todo_source.py

@@ -11,7 +11,6 @@
 from typing import Any
 
 from ...control_plane.coordination.local_authority import (
-    canonical_todo_summary_fields,
     read_canonical_todos_if_promoted,
 )
 from ...control_plane.todos.active_state_todo_parser import parse_active_state_todos
@@ -62,16 +61,11 @@ def read_report_todo_source(
         registry, None, registry_path=registry_path
     )
     canonical = read_canonical_todos_if_promoted(
-        runtime_root=runtime_root, goal_id=goal_id
+        runtime_root=runtime_root, goal_id=goal_id, include_summary=True,
+        rollout_events=rollout_events, available_capabilities=available_capabilities
     )
     if canonical is not None:
-        fields = canonical_todo_summary_fields(
-            canonical["todos"],
-            rollout_events=rollout_events,
-            available_capabilities=available_capabilities,
-            goal_acceptance_contract=canonical.get("goal_acceptance_contract"),
-            goal_acceptance_work_guards=canonical.get("goal_acceptance_work_guards"),
-        )
+        fields = canonical["_todo_fields"]
         return fields, [row for row in canonical["todos"] if row.get("role") == "user"]
     state_path = state_path or resolve_state_file(
         Path(str(goal.get("repo") or "")).expanduser(),

loopx/cli_commands/todo_list.py

@@ -42,6 +42,7 @@ def handle_todo_list_command(
             project=Path(args.project).expanduser() if args.project else None,
             state_file=Path(args.state_file).expanduser() if args.state_file else None,
             runtime_root_arg=runtime_root_arg,
+            json_references=(format_name or str(getattr(args, "format", None) or "markdown")) == "json",
         )
     except Exception as exc:
         from ..usage_ping import capture_failure
@@ -56,6 +57,7 @@ def handle_todo_list_command(
     )
     output_format = format_name or str(getattr(args, "format", None) or "markdown")
     if (payload.get("ok") and not args.todo_id and not args.todo_thin
+            and "todo_list_record_references" not in payload
             and output_format == "json"):
         try:
             payload = effect_runtime_result(

loopx/control_plane/coordination/local_authority.py

@@ -245,6 +245,7 @@ def claim_canonical_todo_if_promoted(
 def read_canonical_todos_if_promoted(
     *, runtime_root: Path, goal_id: str, include_leases: bool = False,
     projection_readback: Mapping[str, Any] | None = None,
+    include_summary: bool = False, rollout_events=None, available_capabilities=None,
 ) -> dict[str, Any] | None:
     """Return canonical Todos after cutover, or ``None`` before cutover.
 
@@ -256,13 +257,28 @@ def read_canonical_todos_if_promoted(
     if not local_authority_is_promoted(runtime_root=runtime_root, goal_id=goal_id):
         return None
 
-    result = read_canonical_snapshot(
-        rpc=effect_runtime_result,
-        runtime_root=str(runtime_root.expanduser().resolve(strict=False)),
-        goal_id=goal_id, include_leases=include_leases,
-        projection_readback=projection_readback,
-        timeout=LOCAL_COORDINATION_TODO_LIST_TIMEOUT_SECONDS,
-    )
+    projected = None
+    if include_summary:
+        from ..todos.read_projection import project_todo_read
+        from ..runtime.time import now_utc
+
+        now = now_utc()
+        projected = project_todo_read({"kind": "canonical_fields",
+            "source": {"runtime_root": str(runtime_root.expanduser().resolve(strict=False)), "goal_id": goal_id,
+                "include_leases": include_leases, "projection_readback": dict(projection_readback) if projection_readback is not None else None},
+            "rollout_events": list(rollout_events or []),
+            "rollout_event_source": getattr(rollout_events, "source", None),
+            "available_capabilities": sorted({str(value).strip().lower() for value in available_capabilities if str(value).strip()}) if available_capabilities is not None else None,
+            "observed_at": now.timestamp(), "evaluated_at": now.isoformat()})
+        result = projected["canonical"]
+    else:
+        result = read_canonical_snapshot(
+            rpc=effect_runtime_result,
+            runtime_root=str(runtime_root.expanduser().resolve(strict=False)),
+            goal_id=goal_id, include_leases=include_leases,
+            projection_readback=projection_readback,
+            timeout=LOCAL_COORDINATION_TODO_LIST_TIMEOUT_SECONDS,
+        )
     if not isinstance(result, Mapping):
         raise LocalCoordinationAuthorityUnavailable(
             "local coordination authority returned an invalid Todo list",
@@ -316,6 +332,9 @@ def read_canonical_todos_if_promoted(
                 "canonical projection confirmation is missing or invalid",
                 code="local_authority_projection_confirmation_invalid", payload=payload,
             )
+    if projected is not None and "fields" in projected:
+        from ..todos.read_projection import restore_todo_field_evaluations
+        payload["_todo_fields"] = restore_todo_field_evaluations(projected)
     return payload
 
 
@@ -338,8 +357,10 @@ def read(self, *, runtime_root: Path, goal_id: str) -> dict[str, Any] | None:
         key = (runtime_root.expanduser().resolve(strict=False), goal_id)
         if key not in self._reads:
             try:
+                from ..todos.resume_condition import TodoResumeRolloutEvents
                 self._reads[key] = read_canonical_todos_if_promoted(
-                    runtime_root=key[0], goal_id=goal_id,
+                    runtime_root=key[0], goal_id=goal_id, include_summary=True,
+                    rollout_events=TodoResumeRolloutEvents([], runtime_root=key[0], goal_id=goal_id),
                 )
             except LocalCoordinationAuthorityUnavailable as error:
                 # A failed first read cannot become a successful attention view
@@ -367,13 +388,10 @@ def read_canonical_todo_fields_if_promoted(
     read neither repairs Markdown nor grants mutation/promotion authority.
     Callers pass the same fields to all decisions in one planning operation.
     """
-    canonical = read_canonical_todos_if_promoted(runtime_root=runtime_root, goal_id=goal_id)
-    return (
-        canonical_todo_summary_fields(canonical["todos"], rollout_events=rollout_events,
-            goal_acceptance_contract=canonical.get("goal_acceptance_contract"),
-            goal_acceptance_work_guards=canonical.get("goal_acceptance_work_guards"))
-        if canonical is not None else None
-    )
+    canonical = read_canonical_todos_if_promoted(runtime_root=runtime_root, goal_id=goal_id,
+        include_summary=True, rollout_events=rollout_events)
+    return canonical["_todo_fields"] if canonical is not None else None
+
 
 
 def canonical_todo_summary_fields(
@@ -384,69 +402,12 @@ def canonical_todo_summary_fields(
     goal_acceptance_contract: dict[str, Any] | None = None,
     goal_acceptance_work_guards: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
-    """Adapt canonical records into the existing Todo summary read model."""
-
-    from ..todos.active_state_editing import TODO_SECTION_HEADINGS
-    from ..todos.standing_decision import build_standing_decision_authority
-    from ..todos.todo_summary import compact_todo_group, count_advancement_todos
+    from ..todos.read_projection import canonical_todo_fields
 
-    # Read standing decisions before assigning presentation-only indexes, and
-    # include retained history so archiving a revocation cannot revive approval.
-    standing_authority = build_standing_decision_authority(
-        [item for item in todos if item.get("role") == "user"], canonical_records=True,
-    )
-    archived_ids = {
-        item["todo_id"]
-        for item in todos
-        if item.get("archive_state") == "archive"
-    }
-    todos = canonical_todo_items(todos)
-    # These are native authority decisions, not persisted Todo fields. Keep the
-    # records visible while every summary/selection uses the same work guard.
-    if goal_acceptance_contract and goal_acceptance_contract.get("enabled") is True:
-        guards = goal_acceptance_work_guards or {}
-        todos = [{**item, "goal_acceptance_guard": guards[item["todo_id"]]}
-            if item.get("todo_id") in guards else item for item in todos]
-    fields: dict[str, Any] = {}
-    for role in ("user", "agent"):
-        items = [
-            item
-            for item in todos
-            if ("user" if item.get("role") == "user" else "agent") == role
-            and item.get("todo_id") not in archived_ids
-        ]
-        summary = compact_todo_group(
-            items,
-            source_section=TODO_SECTION_HEADINGS[role],
-            role=role,
-            include_empty_source=True,
-            resume_source_items=todos,
-            rollout_events=rollout_events,
-            available_capabilities=available_capabilities,
-            item_limit=None,
-        )
-        if summary:
-            if role == "agent":
-                if goal_acceptance_contract and goal_acceptance_contract.get("enabled") is True:
-                    summary["goal_acceptance_contract"] = goal_acceptance_contract
-                archived_done = count_advancement_todos(
-                    [
-                        item
-                        for item in todos
-                        if item.get("todo_id") in archived_ids
-                        and item.get("done") is True
-                    ]
-                )
-                if archived_done:
-                    summary["archived_advancement_done_count"] = archived_done
-                    summary["advancement_done_count"] = (
-                        int(summary.get("advancement_done_count") or 0) + archived_done
-                    )
-            fields[f"{role}_todos"] = summary
-        if role == "user":
-            if standing_authority:
-                fields["standing_decision_authority"] = standing_authority
-    return fields
+    return canonical_todo_fields(todos, rollout_events=rollout_events,
+        available_capabilities=available_capabilities,
+        goal_acceptance_contract=goal_acceptance_contract,
+        goal_acceptance_work_guards=goal_acceptance_work_guards)
 
 
 def canonical_todo_items(todos: list[dict[str, Any]]) -> list[dict[str, Any]]:

loopx/control_plane/effect_runtime.py

@@ -42,6 +42,7 @@
 MAX_REQUEST_BYTES = 2 * 1024 * 1024
 MAX_LOCAL_SNAPSHOT_BYTES = 64 * 1024 * 1024
 LOCAL_SNAPSHOT_METHODS = frozenset({
+    "todo.read",
     "collaboration.delegation.inventory_page",
     "benchmark.trajectory.codex-usage",
     "coordination.local_authority.todo_source",