diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
index 2280fd58a6..caab677cd5 100644
--- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
+++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
@@ -3111,7 +3111,15 @@ This closes that T3/L5 consumer family, not D1 permanent display freshness,
 D2 durability, D3 whole-Goal qualification or default-provider selection. The
 conditional 7–9 remaining delivery-package estimate is unchanged.
 
-Summary/work-lane counts now remain independent of display limits and retain incomplete-source knowledge through Agent scoping; canonical list acceptance holds match status. This closes one L5 read consumer, not permanent projection freshness or D1–D3. See [count semantics](../../reference/todo-work-counts.md).
+The Todo summary consumer now uses one TS batch for scope, lanes, counts,
+claimant-balanced display and closure. It retires Python count/cap/allocation
+branches and two separate internal lane/closure calls. Recent completion uses
+actual completion instants, not edit time or ISO-string order; partial source
+knowledge cannot become a whole-source closure proof after selection. Public
+summary/persisted record schemas and display budgets stay unchanged. Real CLI
+and complete-graph provider readback qualify this read-model boundary, not
+permanent projection freshness or all D1–D3. See
+[count and chronology semantics](../../reference/todo-work-counts.md).
 
 The Goal Channel ownership observation consumes one complete provider revision before bounding display. It never repairs Markdown or revives old local leases; provider failures and truncation stay visible. This is a T3 read closure with shared TS interpretation, not D1/D2 qualification or D3 cutover. See [coordination observation](../../reference/coordination-observation.md).
 
@@ -3296,7 +3304,15 @@ merges qualifies default-on; unresolved acceptance evidence keeps its hold.
 | 6. Capture plus whole-Goal migration/rollback (1–2 PRs) | L7/L8: combine mixed-writer/event continuity, drain, old-writer fencing, canonical readback, fenced export/rollback and cohort migration. | One D3 packet binds exact profile, lineage, source digest and command inventory; existing-Goal cohort cutover remains explicit. |
 | 7. Default/onboarding and bounded Python retirement | L9/T4: make new-Goal creation, settings, installer and packaged clients choose the qualified local profile; publish migration/disable guidance and delete replaced final business writers. | Integrated L8 qualification, rollback and affected user-entrypoint readback. Preserve active rendering, host execution and import/export adapters. |
 
-After boundary 3 lands, **six planned PRs remain, potentially eight** under those
+At the 2026-09-24 reconciliation, #4870/#4888/#4920, #4922 (snapshot pagination),
+#4960 (qualified SQLite runtime admission) and #4961 (refresh display recovery)
+are merged, while #4931 (SQLite proof encoding) remains a separate in-review
+dependency and SQLite #4224 still retains failing/missing D2 evidence. This
+summary slice closes shared read-model decisions and advances boundary 4, not
+those delivery gates: requalify the combined accepted head before updating the
+estimate, and do not count each helper migration as one complete package.
+
+With boundary 3 landed, **six planned PRs remain, potentially eight** under those
 two named splits. Caller audit can expose additional missing effects, so this is
 an implementation estimate, not a guarantee. Small Python business-rule
 retirements accompany their TS owner; deleting all Python is neither the exit
diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md
index 58999b49dc..00ab82ac3d 100644
--- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md
+++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md
@@ -2422,6 +2422,14 @@ Scoped fallback 的选择与门禁关系也已复用同一 TS decision owner，
 
 **D1 — 资格化永久投影交付，可与 T1/T2 重叠推进。**
 
+Todo 摘要现由一个 TS 批次决定范围、lanes、计数、claim 展示分配和收尾证明，删除
+Python 的重复汇总分支及分离的 lane／closure 内部调用。最近完成按真实完成时刻排序，
+不把编辑时间或 ISO 字符串顺序当作完成顺序；partial 来源不能在再次筛选后重新取得
+整源收尾证明。公开摘要／持久记录合同与展示预算保持不变。
+见[计数与时间语义](../../reference/todo-work-counts.md)。这关闭共享摘要决策，不宣称
+全部消费者、永久展示新鲜度或 D1–D3 已通过。
+
+
 摘要与 work-lane 计数已独立于展示上限，并在 Agent 筛选后保留来源不完整状态；canonical 列表的 acceptance 限制与 status 一致。这只闭合 L5 的一个读取消费者，不代表永久投影新鲜度或 D1–D3 完成。见[计数语义](../../reference/todo-work-counts.md)。
 
 Goal Channel 所有权观察先读取完整 provider revision，再限制展示；不修复 Markdown、不复活旧本地 lease，明确披露失败与截断。这是共用 TS 解释规则的 T3 读链路闭合，不完成 D1/D2 或 D3 切换，见 [coordination observation](../../reference/coordination-observation.md)。
@@ -2536,11 +2544,11 @@ provider 确认；权威空集合不回退到陈旧 Markdown。Legacy 与预览
 | D／L9：新 Goal 默认与有界退役 | 单独 default-change PR 让新建／onboarding 选择合格本地 profile，配齐 settings／readback、installer 和打包客户端；最后 caller 与迁移窗口退出才删除旧业务 writer。 | L8 整体产品／回滚资格；区分新 Goal 默认和已有 Goal 迁移。发布兼容／停用说明，保留显式 provider、永久 renderer 和合法 import/export。T4 可在默认启用后继续收尾。 |
 
 **2026-09-24 基线核对。** 保留 claim 的 #4870、reviewed cutover #4888、shadow drain
-规划 #4920 已合并，快照分页 #4922 也已合并，后续应验收组合 head，而不是继续沿用
-旧的 PR hold；SQLite 读取证明优化 #4931 仍在评审。#4910 已加入更大测量轴；#4224
-实测 1 MiB receipt/scan 超预算，恢复和自然时间资格仍有缺项，不能将优化 PR 当成 D2
-通过。本次刷新恢复推进下表中第 4 项投影恢复边界，但没有把其他调用方或默认切换
-标记完成。
+规划 #4920 已合并，快照分页 #4922、SQLite runtime 准入 #4960 与刷新显示恢复 #4961 也已
+合并，后续应验收组合 head，而不是继续沿用旧的 PR hold；SQLite 读取证明优化 #4931 仍在
+评审。#4224 实测 1 MiB receipt/scan 超预算，恢复和自然时间资格仍有缺项，不能将优化 PR
+当成 D2 通过。本次摘要规则收口推进下表中第 4 项投影恢复边界，但没有把其他调用方或默认
+切换标记完成；数量估计应在组合 head 验收后更新，不能按 helper 迁移数量机械扣减。
 
 **开发节奏以证据推进。** 2026-09-23 将旧计划的五类粗粒度交付包细化为
 **七个明确 PR 边界；D2 和迁移各拆两批时最多九个**，其中包含本批一致性分页。
diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
index 32d0d6b2fd..04c7ffa456 100644
--- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
+++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
@@ -1905,3 +1905,16 @@ This is a migration orchestration checkpoint, not completion of Stage 3 or a
 default-provider flip. Integrate claim-preserving migration separately, retain
 real-backend and captured-source qualification, and retire Python only where its
 actual callers have moved. [Operator contract](../../reference/reviewed-coordination-promotion.md).
+
+
+### Todo summary decision ownership
+
+One TS summary batch now owns selected-source counts, display allocation,
+recent-completion chronology, orchestration candidate positions and closure
+proofs. Python retains decoding, public field allowlists and rendering. The
+old Python claimant selector and aggregate branches are retired; the internal
+lane and closure RPC entries are replaced without retaining unused wire paths.
+Public `todo_summary_v0` and persisted records do not change. Full-source
+relationship evaluation is reused before selection, and source completeness is
+preserved independently of query matching. See [semantics and rollback](../../reference/todo-work-counts.md).
+This advances T3/L5; it does not replace D2/D3 or flip a provider default.
diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
index 9e26e9f6f8..21b5d13327 100644
--- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
+++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
@@ -1468,3 +1468,13 @@ This is a migration orchestration checkpoint, not completion of Stage 3 or a
 default-provider flip. Integrate claim-preserving migration separately, retain
 real-backend and captured-source qualification, and retire Python only where its
 actual callers have moved. [Operator contract](../../reference/reviewed-coordination-promotion.md).
+
+
+### Todo 摘要决策收口
+
+已选来源的计数、展示分配、最近完成时间顺序、编排候选位置与收尾证明，收口到一个
+TS 摘要批次；Python 保留旧格式解码、公开字段筛选及渲染。删除旧 Python claim 分配
+算法和汇总分支，用一个内部入口替换 lane／closure 两次调用，不保留无调用方的旧 wire。
+公开 `todo_summary_v0` 和持久记录不变；完整来源的关系求值先于筛选，来源完整性不被
+查询命中情况覆盖。见[语义及回滚](../../reference/todo-work-counts.md)。这是 T3/L5 的
+共享读取边界推进，不替代 D2/D3 或 provider 默认切换。
diff --git a/docs/reference/todo-work-counts.md b/docs/reference/todo-work-counts.md
index 27ebe38343..6d0386a631 100644
--- a/docs/reference/todo-work-counts.md
+++ b/docs/reference/todo-work-counts.md
@@ -79,10 +79,11 @@ 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. The v1 internal
-request composes this selection into the existing call; v0 unfiltered callers
-retain their wire contract. Python decodes legacy input and renders results,
-with no independent Agent-addressing rule.
+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.
 
 ## 中文说明
 
@@ -112,3 +113,48 @@ claim/exclusion 筛选，可见不代表获准执行。
 当前列表。未筛选的整 Goal 视图仍显示这些记录。依赖和 succession 先在完整来源求值，
 TS 再筛选并保留原数组位置，最后生成 lanes、计数和有界展示；筛选后的数组位置不是原
 来源位置。无需新增 capability、配置、前端或 Lark 编辑入口，不增加一次筛选 RPC。
+
+## Summary chronology and source completeness
+
+The same TS projection now supplies `recent_completed_advancement_items`,
+claimant-balanced display lanes, orchestration candidate positions and closure
+proofs to legacy and canonical consumers. Display budgets are unchanged;
+`items` limits never change full-source counts. Full-source resume/succession
+evaluation still precedes filtering, and returned positions refer to the original
+array, not a newly numbered subset. Python retains public field allowlists,
+warning text, privacy redaction and Markdown parsing/rendering.
+
+**Intentional read behavior changes:** recent completions are ordered by the
+actual `completed_at` instant, preserving timezone offsets and microseconds.
+Later `updated_at` edits no longer make an old completion recent. Missing or
+invalid completion times remain in completed-work counts/history but do not
+claim a place in the recent-completion lane. Equal instants retain reverse
+source-coordinate order and stable ties. Succession warnings keep their
+last-change ordering, now comparing instants rather than timestamp strings;
+unknown instants follow known ones without discarding the warning.
+
+A source already marked partial cannot regain `source_proof` or
+`terminal_closure_proof` simply because a later selection matches all visible
+rows. Query scope and source completeness are independent conditions. These
+proofs remain read-only observations, not permission to settle a Goal.
+
+This changes status, Todo-list and quota summary readback for both legacy and
+promoted Goals without a flag. Existing frontend and Lark views consume these
+Core projections; no new setting or frontend asset is required. No provider,
+lease, registry or display writer is added. Rollback requires the matching
+Python/TS package but no data migration. Full L5 consumer acceptance, projection
+freshness, SQLite D2 and default/cutover gates remain separate.
+
+### 中文补充
+
+摘要的计数、展示上限、领取者之间的展示分配、编排候选位置及收尾证明，现由一个 TS
+批次决定；删除 Python 的重复汇总分支和仅为旧内部调用保留的 claim 分配 helper。
+Python 继续负责旧数据解码、公开字段筛选、隐私处理与文本展示。
+
+这是有意的读取语义修复：最近完成列表按 `completed_at` 的真实时刻排序，保留时区和
+微秒，不再把较晚编辑误作较晚完成。缺失／非法时间仍计入已完成总数和历史，但不进入
+最近完成列表。后继缺口警告仍按最后更新时间排序，未知时间靠后，不丢弃警告。
+已有 partial 来源不会因为再次筛选命中所有可见行，就重新获得整个来源的收尾证明。
+
+覆盖 legacy 与 canonical 的 status、Todo 查询和 quota 摘要；展示预算保持原值。
+没有新增设置、权限或 writer，不改变 provider 默认值，也不宣称完成整 Goal 迁移。
diff --git a/examples/control_plane/todo-projection-shared-helper-smoke.py b/examples/control_plane/todo-projection-shared-helper-smoke.py
index 1f70c34dce..e77b11c1b3 100644
--- a/examples/control_plane/todo-projection-shared-helper-smoke.py
+++ b/examples/control_plane/todo-projection-shared-helper-smoke.py
@@ -33,7 +33,6 @@
     build_goal_frontier_projection_from_summaries,
 )
 from loopx.status import (  # noqa: E402
-    claimed_visibility_items as status_claimed_visibility_items,
     todo_item_is_deferred as status_todo_item_is_deferred,
     todo_projection_sort_key,
 )
@@ -42,7 +41,6 @@
     TODO_TASK_CLASS_MONITOR,
 )
 from loopx.control_plane.todos.projection import (  # noqa: E402
-    todo_claimed_visibility_items as shared_claimed_visibility_items,
     todo_item_claimed_by_agent_or_unclaimed as shared_todo_item_claimed_by_agent_or_unclaimed,
     todo_item_is_deferred as shared_todo_item_is_deferred,
     todo_projection_sort_key as shared_todo_projection_sort_key,
@@ -165,7 +163,7 @@ def assert_shared_ordering_parity(summary: dict) -> None:
     assert quota_todo_projection_sort_key(embedded_priority) == (50, 9), embedded_priority
 
 
-def assert_claimed_visibility_parity() -> None:
+def assert_claim_scope_parity() -> None:
     items = [
         quota_todo_item(
             todo_id="todo_a1",
@@ -199,18 +197,6 @@ def assert_claimed_visibility_parity() -> None:
             task_class=TODO_TASK_CLASS_ADVANCEMENT,
         ),
     ]
-    for selector in (
-        shared_claimed_visibility_items,
-        status_claimed_visibility_items,
-    ):
-        selected_two = selector(items, limit=2)
-        assert [item["todo_id"] for item in selected_two] == ["todo_a1", "todo_b1"], selected_two
-        selected_three = selector(items, limit=3)
-        assert [item["todo_id"] for item in selected_three] == [
-            "todo_a1",
-            "todo_a2",
-            "todo_b1",
-        ], selected_three
     claimed_by_current = items[0]
     claimed_by_other = items[2]
     unclaimed = items[3]
@@ -487,7 +473,7 @@ def main() -> int:
     summary = build_agent_todo_summary()
     assert_status_summary_lanes(summary)
     assert_shared_ordering_parity(summary)
-    assert_claimed_visibility_parity()
+    assert_claim_scope_parity()
     assert_agent_scope_frontier_routing_parity()
     assert_deferred_helper_parity()
     assert_monitor_item_collection_parity(summary)
diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index fba95a150b..61727926cd 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -1,3 +1,4 @@
+import {projectTodoSummary} from "./todos/summary_projection.ts";
 import {admitAutomationStart, confirmAutomationStart, manageAutomationCadence, projectCadenceSchedule} from "./quota/automation_cadence.ts";
 import {deliverShadowEntry} from "./coordination/shadow_entry_delivery.ts";
 import {readShadowDrainPlan} from "./coordination/shadow_drain_plan.ts";
@@ -9,8 +10,8 @@ import {projectPeerOrchestration} from "./quota/peer_orchestration.ts";
 import {inspectTaskLease} from "./work_items/task_lease_inspection.ts";
 import {evaluateTodoPriority} from "./todos/priority.ts";
 import {evaluateUserCompletion} from "./todos/user_completion.ts";
-import {projectTodoSuccession, projectTodoClosure} from "./todos/succession.ts";
-import {projectTodoSummaryLanes, projectLegacyTodoWorkCounts} from "./todos/summary_lanes.ts";
+import {projectTodoSuccession} from "./todos/succession.ts";
+import {projectLegacyTodoWorkCounts} from "./todos/summary_lanes.ts";
 import {recordDelegationAdoption, delegationInventoryItem, delegationInventoryQuery, delegationPreflight, delegationTurnPlanDecision, recoverValidatedDelegationSettlement, selectDelegationBinding, transitionDelegationObservation} from "./collaboration/delegation.ts";
 import {planChatMode} from "./collaboration/chat_mode.ts";
 import {resolveConversationScope} from "./collaboration/conversation_scope.ts";
@@ -442,11 +443,10 @@ export function createEffectRuntimeHandlers(
     ["todo.priority.plan", evaluateTodoPriority],
     ["todo.public_update.plan", planPublicTodoUpdate],
     ["todo.standing_decision.project", evaluateStandingDecisionProjection],
-    ["todo.summary_lanes.project", projectTodoSummaryLanes],
+    ["todo.summary.project", projectTodoSummary],
     ["capabilities.periodic_report.progress.select", selectPeriodicReportProgress],
     ["capabilities.periodic_report.approval_retry.select", selectPeriodicReportApprovalRetry],
     ["todo.succession.project", projectTodoSuccession],
-    ["todo.succession.closure", projectTodoClosure],
     ["todo.work_counts.project", projectLegacyTodoWorkCounts],
     ["todo.decision_scope.evaluate", evaluateDecisionScope],
     ["todo.user_completion.plan", evaluateUserCompletion],
diff --git a/loopx/control_plane/todos/summary_projection.ts b/loopx/control_plane/todos/summary_projection.ts
new file mode 100644
index 0000000000..f0c3af39e2
--- /dev/null
+++ b/loopx/control_plane/todos/summary_projection.ts
@@ -0,0 +1,182 @@
+/** Whole-source summary decisions. Python materializes these ordinals as public
+ * display records; counts and closure never depend on a presentation budget. */
+import type {JsonObject} from "../effect_program.ts";
+import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts";
+import {requireBoolean, requireJsonObject, requireStringLiteral} from "../runtime_decode.ts";
+import {parseTodoTimestampMicros} from "../runtime_timestamp.ts";
+import {projectTodoSummaryLanes, type TodoSummaryLane} from "./summary_lanes.ts";
+import {projectTodoClosure} from "./succession.ts";
+
+type Format = "raw" | "compact" | "active" | "recent" | "gap";
+interface DisplayLane {indices: number[]; format: Format}
+interface SummaryProjection {
+  schema_version: "todo_summary_projection_v0";
+  source_indices: number[];
+  full_selection: boolean;
+  fields: JsonObject;
+  lanes: Record<string, DisplayLane>;
+  orchestration: {candidate_items: number[]; user_blocker_items: number[]};
+}
+
+/** The declared order is this request version's schema, not a hint: the same
+ * cells in a different order would silently change meaning. */
+export const TODO_SUMMARY_PROJECTION_COLUMNS = [
+  "status", "done", "task_class", "has_resume", "resume_ready", "resume_evaluated",
+  "acceptance_blocked", "claimed", "preferred", "watch_only", "due_at", "expires_at",
+  "sort", "completed_at", "updated_at", "completion_index", "linked_user_action",
+  "no_followup", "successor_gap", "handoff_state", "replan", "todo_id", "claim",
+  "bound", "blocks", "global", "excluded",
+] as const;
+
+/** One whole-source batch carries every Todo, so the co-deployed adapter sends
+ * columnar facts. Decoding restores the row objects the lane and closure owners
+ * already validate; nothing is defaulted or inferred from absence. */
+function decodeRows(request: JsonObject): JsonObject[] {
+  const columns = request.columns;
+  if (!Array.isArray(columns) || columns.length !== TODO_SUMMARY_PROJECTION_COLUMNS.length ||
+      columns.some((name, index) => name !== TODO_SUMMARY_PROJECTION_COLUMNS[index])) {
+    throw new EffectRuntimeRequestError("Todo summary row columns do not match the typed adapter order");
+  }
+  if (!Array.isArray(request.rows)) {
+    throw new EffectRuntimeRequestError("Todo summary rows must be a list");
+  }
+  return request.rows.map((value, ordinal) => {
+    if (!Array.isArray(value) || value.length !== columns.length) {
+      throw new EffectRuntimeRequestError(`Todo summary row ${ordinal} does not match its declared columns`);
+    }
+    return Object.fromEntries(TODO_SUMMARY_PROJECTION_COLUMNS.map((name, index) => [name, value[index]]));
+  });
+}
+
+/** Allocate a bounded display across claimants, then restore source ordering. */
+function claimedVisibility(indices: readonly number[], rows: readonly JsonObject[], limit: number): number[] {
+  if (indices.length <= limit) return [...indices];
+  const buckets = new Map<string, number[]>();
+  for (const index of indices) {
+    const claim = rows[index].claim;
+    if (typeof claim !== "string" || !claim) continue;
+    const bucket = buckets.get(claim) ?? [];
+    bucket.push(index); buckets.set(claim, bucket);
+  }
+  if (!buckets.size) return indices.slice(0, limit);
+  const perClaimant = Math.max(1, Math.floor(limit / buckets.size));
+  const selected = new Set<number>();
+  for (const bucket of buckets.values()) {
+    for (const index of bucket.slice(0, perClaimant)) {
+      if (selected.size < limit) selected.add(index);
+    }
+  }
+  for (const index of indices) if (selected.size < limit) selected.add(index);
+  return indices.filter(index => selected.has(index));
+}
+
+export function projectTodoSummary(value: unknown): SummaryProjection {
+  const request = requireJsonObject(value, "Todo summary request");
+  if (request.schema_version !== "todo_summary_projection_request_v1") {
+    throw new EffectRuntimeRequestError("Todo summary request schema mismatch");
+  }
+  const role = request.role === null ? null : requireStringLiteral(request.role, ["user", "agent"], "role");
+  if (request.source_section !== null && typeof request.source_section !== "string") {
+    throw new EffectRuntimeRequestError("source_section must be text or null");
+  }
+  const limit = request.item_limit;
+  if (limit !== null && (typeof limit !== "number" || !Number.isSafeInteger(limit) || limit < 0)) {
+    throw new EffectRuntimeRequestError("item_limit must be a non-negative integer or null");
+  }
+  const full = requireBoolean(request.full_selection, "full_selection");
+  const rows = decodeRows(request);
+  // The co-deployed adapter sends source facts, not prose or full Todo bodies.
+  for (const row of rows) {
+    if ((row.claim !== null && typeof row.claim !== "string") || row.claimed !== Boolean(row.claim)) {
+      throw new EffectRuntimeRequestError("summary claimant facts disagree");
+    }
+    requireBoolean(row.linked_user_action, "linked_user_action");
+    if ((row.completed_at !== null && typeof row.completed_at !== "string") ||
+        (row.updated_at !== null && typeof row.updated_at !== "string") ||
+        typeof row.completion_index !== "number" || !Number.isSafeInteger(row.completion_index)) {
+      throw new EffectRuntimeRequestError("invalid summary completion coordinates");
+    }
+  }
+  const hasSelection = request.selection !== null;
+  const projected = projectTodoSummaryLanes({
+    schema_version: hasSelection ? "todo_summary_lanes_request_v1" : "todo_summary_lanes_request_v0",
+    rows, observed_at: request.observed_at, ...(hasSelection ? {selection: request.selection} : {}),
+  });
+  const source = hasSelection ? projected.source_indices as number[] : rows.map((_, index) => index);
+  const fullSelection = full && (!hasSelection || projected.full_selection === true);
+  const selected = projected.lanes as Record<TodoSummaryLane, number[]>;
+  const fields: JsonObject = {schema_version: "todo_summary_v0", source_section: request.source_section,
+    total_count: source.length, work_counts: projected.work_counts,
+    open_count: selected.open_items.length, done_count: selected.terminal_items.length,
+    advancement_done_count: selected.done_items.filter(index => rows[index].task_class === "advancement_task").length,
+    deferred_count: selected.deferred_items.length, monitor_due_count: selected.monitor_due_items.length,
+    monitor_schedule_gap_count: selected.monitor_schedule_gap_items.length,
+  };
+  const lanes: Record<string, DisplayLane> = {};
+  const lane = (name: string, indices: readonly number[], cap: number | null = null, format: Format = "compact") => {
+    lanes[name] = {indices: cap === null ? [...indices] : indices.slice(0, cap), format};
+  };
+  lane("first_open_items", selected.projected_open_items, 3);
+  lane("first_executable_items", selected.executable_items, 3);
+  lane("monitor_open_items", selected.monitor_items);
+  lane("monitor_due_items", selected.monitor_due_items, 1);
+  lane("monitor_schedule_gap_items", selected.monitor_schedule_gap_items, 1);
+  lane("unclaimed_priority_open_items", selected.unclaimed_open_items, 8);
+  lane("claimed_open_items", claimedVisibility(selected.claimed_open_items, rows, 16));
+  lane("claimed_advancement_open_items", claimedVisibility(selected.claimed_advancement_items, rows, 16));
+  lane("claimed_monitor_open_items", claimedVisibility(selected.claimed_monitor_items, rows, 16));
+  lane("backlog_items", selected.projected_open_items, 8);
+  lane("executable_backlog_items", selected.executable_items, 8);
+  lane("deferred_items", selected.projected_deferred_items, 8);
+  lane("deferred_resume_candidates", selected.projected_deferred_items.filter(index => rows[index].resume_ready === true), 8);
+  lane("items", selected.budgeted_items, limit, "raw");
+
+  // Recent completion is chronological, not ISO-string order or last-edit order.
+  // Unknown legacy times remain in totals, but cannot claim a recent timestamp.
+  const instant = (value: unknown): bigint | null => typeof value === "string" && value.trim()
+    ? parseTodoTimestampMicros(value.trim()) : null;
+  const completedAt = rows.map(row => instant(row.completed_at));
+  const changedAt = rows.map(row => instant(row.updated_at || row.completed_at));
+  const timestamp = (index: number, completionOnly: boolean) => (completionOnly ? completedAt : changedAt)[index];
+  const byTime = (completionOnly: boolean) => (left: number, right: number) => {
+    const a = timestamp(left, completionOnly), b = timestamp(right, completionOnly);
+    if (a !== b) return a === null ? 1 : b === null ? -1 : a > b ? -1 : 1;
+    return Number(rows[right].completion_index) - Number(rows[left].completion_index);
+  };
+  const recent = selected.done_items.filter(index => rows[index].task_class === "advancement_task" && timestamp(index, true) !== null)
+    .sort(byTime(true));
+  if (recent.length) lane("recent_completed_advancement_items", recent, 16, "recent");
+  const gaps = source.filter(index => rows[index].successor_gap === true).sort(byTime(false));
+  if (gaps.length) {
+    fields.completed_without_successor_count = gaps.length;
+    lane("completed_without_successor_items", gaps, 5, "gap");
+  }
+  if (selected.watch_only_monitor_items.length) {
+    fields.watch_only_monitor_count = selected.watch_only_monitor_items.length;
+    fields.watch_only_monitor_due_count = selected.watch_only_monitor_due_items.length;
+    fields.convergence_open_count = selected.convergent_open_items.length;
+  }
+  for (const [name, indices, cap] of [
+    ["blocker", selected.blocker_items, null], ["resume_blocked", selected.resume_blocked_items, 8],
+  ] as const) {
+    if (indices.length) {
+      fields[name === "blocker" ? "blocker_open_count" : "resume_blocked_count"] = indices.length;
+      lane(`${name}_items`, indices, cap);
+    }
+  }
+  if (selected.active_next_action_items.length) lane("active_next_action_items", selected.active_next_action_items, null, "active");
+  if (selected.active_next_action_executable_items.length) lane("active_next_action_executable_items", selected.active_next_action_executable_items, null, "active");
+  if (selected.claimed_open_items.length) {
+    fields.claimed_open_count = selected.claimed_open_items.length;
+    fields.unclaimed_open_count = selected.open_items.length - selected.claimed_open_items.length;
+    fields.claimed_advancement_open_count = selected.claimed_advancement_items.length;
+    fields.claimed_monitor_open_count = selected.claimed_monitor_items.length;
+  }
+  Object.assign(fields, projectTodoClosure({schema_version: "todo_closure_request_v0", role,
+    source_section: request.source_section, full_selection: fullSelection, rows: source.map(index => rows[index])}));
+  return {schema_version: "todo_summary_projection_v0", source_indices: source, full_selection: fullSelection, fields, lanes,
+    orchestration: {
+      candidate_items: role === "agent" ? selected.projected_open_items.filter(index => rows[index].task_class === "advancement_task") : [],
+      user_blocker_items: role === "user" ? selected.projected_open_items.filter(index => rows[index].linked_user_action === true) : [],
+    }};
+}
diff --git a/loopx/control_plane/todos/todo_semantics.py b/loopx/control_plane/todos/todo_semantics.py
index cda58b9cd1..b0667261e7 100644
--- a/loopx/control_plane/todos/todo_semantics.py
+++ b/loopx/control_plane/todos/todo_semantics.py
@@ -192,56 +192,6 @@ def todo_projection_sort_key(
     return (todo_priority_rank(item, text_mode=text_mode), todo_index_rank(item))
 
 
-def todo_claimed_visibility_items(
-    items: list[dict[str, Any]],
-    *,
-    limit: int,
-) -> list[dict[str, Any]]:
-    if limit <= 0 or len(items) <= limit:
-        return items[:limit]
-    claim_order: list[str] = []
-    buckets: dict[str, list[dict[str, Any]]] = {}
-    for item in items:
-        claimed_by = normalize_todo_claimed_by(item.get("claimed_by"))
-        if not claimed_by:
-            continue
-        if claimed_by not in buckets:
-            buckets[claimed_by] = []
-            claim_order.append(claimed_by)
-        buckets[claimed_by].append(item)
-    if not buckets:
-        return items[:limit]
-
-    original_index = {id(item): index for index, item in enumerate(items)}
-    per_claimant_cap = max(1, limit // len(buckets))
-    selected: list[dict[str, Any]] = []
-    selected_ids: set[int] = set()
-    for claimed_by in claim_order:
-        taken = 0
-        for item in buckets[claimed_by]:
-            if taken >= per_claimant_cap:
-                break
-            if len(selected) >= limit:
-                break
-            selected.append(item)
-            selected_ids.add(id(item))
-            taken += 1
-        if len(selected) >= limit:
-            break
-
-    if len(selected) < limit:
-        for item in items:
-            if id(item) in selected_ids:
-                continue
-            selected.append(item)
-            selected_ids.add(id(item))
-            if len(selected) >= limit:
-                break
-
-    return sorted(
-        selected, key=lambda item: original_index.get(id(item), TODO_MISSING_INDEX)
-    )[:limit]
-
 
 def todo_item_task_text(
     item: dict[str, Any],
diff --git a/loopx/control_plane/todos/todo_summary.py b/loopx/control_plane/todos/todo_summary.py
index 6b74353c78..5f739d34de 100644
--- a/loopx/control_plane/todos/todo_summary.py
+++ b/loopx/control_plane/todos/todo_summary.py
@@ -1,9 +1,8 @@
 from __future__ import annotations
 
-from dataclasses import dataclass
 from datetime import datetime
 import re
-from typing import Any, Callable, Optional
+from typing import Any, Callable, Optional, TypeGuard
 
 from ..goals.goal_vision_wait_projection import attach_active_vision_waits
 from .contract import (
@@ -45,7 +44,6 @@
 from .handoff_gate import build_todo_handoff_gate_states
 from .handoff_note import attach_todo_handoff_note
 from .todo_semantics import (
-    todo_claimed_visibility_items as projection_todo_claimed_visibility_items,
     todo_item_is_actionable_open as projection_todo_item_is_actionable_open,
     todo_item_is_deferred as projection_todo_item_is_deferred,
     todo_item_is_due_monitor as projection_todo_item_is_due_monitor,
@@ -84,10 +82,22 @@
 MAX_MONITOR_DUE_ITEMS = 1
 MAX_DEPENDENCY_BLOCKERS = 4
 MAX_COMPLETED_SUCCESSION_WARNING_ITEMS = 5
-MAX_RECENT_COMPLETED_ADVANCEMENT_ITEMS = MAX_TODO_VISIBILITY_LANE_ITEMS
 
 TASK_ORCHESTRATION_AUTHORITY_SCHEMA_VERSION = "task_orchestration_authority_v0"
 TODO_ARCHIVE_STATE_ACTIVE = "active"
+
+# One internal batch carries the whole source, so the adapter sends columnar
+# facts: repeating every key name per Todo pushed a long-history request past the
+# effect-runtime request budget. The typed owner decodes the declared columns
+# back into row objects before validating them, so no cell changes meaning.
+SUMMARY_PROJECTION_REQUEST_SCHEMA_VERSION = "todo_summary_projection_request_v1"
+SUMMARY_PROJECTION_COLUMNS = (
+    "status", "done", "task_class", "has_resume", "resume_ready", "resume_evaluated",
+    "acceptance_blocked", "claimed", "preferred", "watch_only", "due_at", "expires_at",
+    "sort", "completed_at", "updated_at", "completion_index", "linked_user_action",
+    "no_followup", "successor_gap", "handoff_state", "replan", "todo_id", "claim",
+    "bound", "blocks", "global", "excluded",
+)
 AttentionItemBuilder = Callable[..., dict[str, Any]]
 GoalLifecycleFields = Callable[[dict[str, Any], Optional[dict[str, Any]]], dict[str, Any]]
 PublicSafeText = Callable[..., Optional[str]]
@@ -95,31 +105,6 @@
 FirstOpenTodoText = Callable[[Optional[dict[str, Any]]], Optional[str]]
 
 
-@dataclass(frozen=True)
-class _TodoGroupLanes:
-    open_items: list[dict[str, Any]]
-    terminal_items: list[dict[str, Any]]
-    deferred_items: list[dict[str, Any]]
-    done_items: list[dict[str, Any]]
-    projected_open_items: list[dict[str, Any]]
-    projected_deferred_items: list[dict[str, Any]]
-    budgeted_items: list[dict[str, Any]]
-    claimed_open_items: list[dict[str, Any]]
-    unclaimed_open_items: list[dict[str, Any]]
-    executable_items: list[dict[str, Any]]
-    blocker_items: list[dict[str, Any]]
-    resume_blocked_items: list[dict[str, Any]]
-    monitor_items: list[dict[str, Any]]
-    monitor_due_items: list[dict[str, Any]]
-    watch_only_monitor_items: list[dict[str, Any]]
-    watch_only_monitor_due_items: list[dict[str, Any]]
-    non_watch_only_monitor_due_items: list[dict[str, Any]]
-    convergent_open_items: list[dict[str, Any]]
-    monitor_schedule_gap_items: list[dict[str, Any]]
-    claimed_advancement_items: list[dict[str, Any]]
-    claimed_monitor_items: list[dict[str, Any]]
-    active_next_action_items: list[dict[str, Any]]
-    active_next_action_executable_items: list[dict[str, Any]]
 TASK_ORCHESTRATION_CANDIDATE_FIELDS = (
     "todo_id",
     "status",
@@ -494,45 +479,13 @@ def canonical_todo_read_record(
     return record
 
 
-def _task_orchestration_authority(
-    lanes: _TodoGroupLanes,
-    *,
-    role: str | None,
-) -> dict[str, Any]:
-    candidate_items = (
-        [
-            {
-                key: compact[key]
-                for key in TASK_ORCHESTRATION_CANDIDATE_FIELDS
-                if key in compact
-            }
-            for item in lanes.projected_open_items
-            if todo_item_task_class(item) == TODO_TASK_CLASS_ADVANCEMENT
-            for compact in [compact_todo_item(item)]
-        ]
-        if role == "agent"
-        else []
-    )
-    user_blocker_items = (
-        [
-            {
-                key: compact[key]
-                for key in TASK_ORCHESTRATION_USER_BLOCKER_FIELDS
-                if key in compact
-            }
-            for item in lanes.projected_open_items
-            if normalize_todo_id(item.get("unblocks_todo_id"))
-            for compact in [compact_todo_item(item)]
-        ]
-        if role == "user"
-        else []
-    )
-    return {
-        "schema_version": TASK_ORCHESTRATION_AUTHORITY_SCHEMA_VERSION,
-        "role": role,
-        "candidate_items": candidate_items,
-        "user_blocker_items": user_blocker_items,
-    }
+def _task_orchestration_authority(lanes: dict[str, list[dict[str, Any]]], *, role: str | None) -> dict[str, Any]:
+    """Materialize the typed selection using the existing public field allowlist."""
+    return {"schema_version": TASK_ORCHESTRATION_AUTHORITY_SCHEMA_VERSION, "role": role,
+        **{name: [{key: compact[key] for key in fields if key in compact}
+                  for item in lanes[name] for compact in [compact_todo_item(item)]]
+           for name, fields in (("candidate_items", TASK_ORCHESTRATION_CANDIDATE_FIELDS),
+                                ("user_blocker_items", TASK_ORCHESTRATION_USER_BLOCKER_FIELDS))}}
 
 
 def compact_active_next_action_todo_item(item: dict[str, Any]) -> dict[str, Any]:
@@ -591,10 +544,6 @@ def todo_projection_sort_key(item: dict[str, Any]) -> tuple[int, int]:
     return projection_todo_projection_sort_key(item, text_mode="prefix")
 
 
-def claimed_visibility_items(items: list[dict[str, Any]], *, limit: int) -> list[dict[str, Any]]:
-    return projection_todo_claimed_visibility_items(items, limit=limit)
-
-
 def todo_item_is_deferred(item: dict[str, Any]) -> bool:
     return projection_todo_item_is_deferred(item)
 
@@ -825,32 +774,6 @@ def todo_item_is_succession_tracked_completion(item: dict[str, Any]) -> bool:
     return project_succession([item])[0]["tracked_completion"] is True
 
 
-def _completed_succession_sort_key(item: dict[str, Any]) -> tuple[str, int]:
-    raw_index = item.get("index")
-    try:
-        index = int(raw_index) if raw_index is not None else 0
-    except (TypeError, ValueError):
-        index = 0
-    timestamp = str(item.get("updated_at") or item.get("completed_at") or "")
-    return (timestamp, index)
-
-
-def completed_without_successor_items(
-    items: list[dict[str, Any]], *, evaluations: list[dict[str, Any]],
-) -> list[dict[str, Any]]:
-    gap_items = []
-    for item, evaluation in zip(items, evaluations, strict=True):
-        if not evaluation["successor_gap"]:
-            continue
-        compact = compact_todo_item(item)
-        for key in ("note", "evidence", "reason"):
-            compact.pop(key, None)
-        compact["succession_tracked"] = True
-        compact["recommended_action"] = "record no_followup=true or add/link a successor todo"
-        gap_items.append(compact)
-    return sorted(gap_items, key=_completed_succession_sort_key, reverse=True)
-
-
 def _structured_todo_group_items(
     items: list[dict[str, Any]],
     *,
@@ -895,21 +818,41 @@ def _structured_resume_source_items(
     ]
 
 
-def _project_summary_lanes(items: list[dict[str, Any]], preferred_todo_ids: set[str] | None,
-    selection: dict[str, Any] | None = None,
+def _resume_condition_evaluated(item: dict[str, Any], resume: str | None) -> bool:
+    """The source's own full-source resume evaluation for this condition."""
+    condition = item.get("resume_condition")
+    return (isinstance(condition, dict)
+        and condition.get("schema_version") == "todo_resume_condition_v0"
+        and condition.get("resume_when") == resume
+        and isinstance(condition.get("satisfied"), bool)
+        and item.get("resume_ready") is condition.get("satisfied"))
+
+
+def _project_summary(items: list[dict[str, Any]], preferred_todo_ids: set[str] | None,
+    *, selection: dict[str, Any] | None, role: str | None, source_section: str | None,
+    item_limit: int | None, full_selection: bool,
 ) -> dict[str, Any]:
-    """Adapt legacy facts and read batch ordinals from the typed lane owner."""
+    """Adapt evaluated facts and materialize one typed summary decision."""
     from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result
 
-    rows = []
+    from .succession_warning import project_succession
+
+    # Display may never invent the source's resume decision. Assert the
+    # full-source precondition before any RPC that reuses the evaluation, so an
+    # unevaluated source fails with its own diagnostic instead of a downstream
+    # "succession evaluation must be an object" from a later owner.
     for item in items:
         resume = normalize_todo_resume_when(item.get("resume_when"))
-        condition = item.get("resume_condition")
-        evaluated = (isinstance(condition, dict)
-            and condition.get("schema_version") == "todo_resume_condition_v0"
-            and condition.get("resume_when") == resume
-            and isinstance(condition.get("satisfied"), bool)
-            and item.get("resume_ready") is condition.get("satisfied"))
+        if resume and not _resume_condition_evaluated(item, resume):
+            raise ValueError("Todo display requires a matching full-source resume evaluation")
+    succession = project_succession(items, reuse=True)
+    handoff_gates = build_todo_handoff_gate_states(items, evaluations=succession)
+    replan_gates = {gate.get("todo_id") for gate in handoff_gates
+        if gate.get("route_continuation_replan_required") is True}
+    rows = []
+    for item, evaluation in zip(items, succession, strict=True):
+        resume = normalize_todo_resume_when(item.get("resume_when"))
+        evaluated = _resume_condition_evaluated(item, resume)
         due = projection_todo_item_next_due_at(item)
         expires = projection_todo_item_expires_at(item)
         guard = item.get("goal_acceptance_guard")
@@ -921,41 +864,82 @@ def _project_summary_lanes(items: list[dict[str, Any]], preferred_todo_ids: set[
             "watch_only": projection_todo_item_is_watch_only_monitor(item),
             "due_at": due.timestamp() if due else None, "expires_at": expires.timestamp() if expires else None,
             "sort": list(projection_todo_presentation_sort_key(item)),
-            **({"todo_id": normalize_todo_id(item.get("todo_id")),
+            "completed_at": str(item.get("completed_at") or "") or None,
+            "updated_at": str(item.get("updated_at") or "") or None,
+            "completion_index": int(item.get("index") or 0),
+            "linked_user_action": bool(normalize_todo_id(item.get("unblocks_todo_id"))),
+            "no_followup": normalize_todo_no_followup(item.get("no_followup")) is True,
+            "successor_gap": evaluation["successor_gap"], "handoff_state": evaluation["handoff_state"],
+            "replan": item.get("route_continuation_replan_required") is True or item.get("todo_id") in replan_gates,
+            **{"todo_id": normalize_todo_id(item.get("todo_id")),
                 "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")),
                 "global": bool(item.get("global_gate")),
-                "excluded": normalize_todo_excluded_agents(item.get("excluded_agents"))}
-                if selection is not None else {})})
+                "excluded": normalize_todo_excluded_agents(item.get("excluded_agents"))}})
     try:
-        result = effect_runtime_result("todo.summary_lanes.project", {
-            "schema_version": "todo_summary_lanes_request_v0" if selection is None else "todo_summary_lanes_request_v1",
-            "rows": rows, "observed_at": now_utc().timestamp(),
-            **({"selection": selection} if selection is not None else {}),
+        result = effect_runtime_result("todo.summary.project", {
+            "schema_version": SUMMARY_PROJECTION_REQUEST_SCHEMA_VERSION,
+            "columns": list(SUMMARY_PROJECTION_COLUMNS),
+            "rows": [[row[name] for name in SUMMARY_PROJECTION_COLUMNS] for row in rows],
+            "observed_at": now_utc().timestamp(),
+            "selection": selection, "role": role, "source_section": source_section,
+            "item_limit": item_limit, "full_selection": full_selection,
         })
     except EffectRuntimeRejected as error:
         raise ValueError(str(error)) from error
-    if not isinstance(result, dict) or result.get("schema_version") != "todo_summary_lanes_v0":
-        raise ValueError("invalid typed Todo summary lanes")
-    def valid_ordinals(value: Any) -> bool:
+    if not isinstance(result, dict) or result.get("schema_version") != "todo_summary_projection_v0":
+        raise ValueError("invalid typed Todo summary projection")
+
+    def valid_ordinals(value: Any) -> TypeGuard[list[int]]:
         return (isinstance(value, list)
             and all(type(index) is int and 0 <= index < len(items) for index in value)
             and len(set(value)) == len(value))
 
-    lanes = result["lanes"]
-    if not isinstance(lanes, dict) or any(not valid_ordinals(indices) for indices in lanes.values()):
-        raise ValueError("invalid Todo summary source ordinal")
-    selected = result.get("source_indices", list(range(len(items))))
-    if (not valid_ordinals(selected)
-        or selection is not None and ("source_indices" not in result or type(result.get("full_selection")) is not bool)):
+    selected = result.get("source_indices")
+    if not valid_ordinals(selected):
+        raise ValueError("invalid typed Todo selection ordinals")
+    if type(result.get("full_selection")) is not bool:
         raise ValueError("invalid typed Todo selection ordinals")
     selected_set = set(selected)
-    if any(not set(indices) <= selected_set for indices in lanes.values()):
-        raise ValueError("Todo summary lane escaped the selected source")
-    return {"lanes": {key: [items[index] for index in indices] for key, indices in lanes.items()},
-            "items": [items[index] for index in selected],
-            "full_selection": result.get("full_selection", True), "work_counts": result["work_counts"]}
+    lanes, orchestration = result.get("lanes"), result.get("orchestration")
+    if not isinstance(lanes, dict) or not isinstance(orchestration, dict):
+        raise ValueError("invalid typed Todo summary lanes")
+    for indices in [*(lane.get("indices") if isinstance(lane, dict) else None for lane in lanes.values()),
+                    *orchestration.values()]:
+        if not valid_ordinals(indices):
+            raise ValueError("invalid Todo summary source ordinal")
+        if not set(indices) <= selected_set:
+            raise ValueError("Todo summary lane escaped the selected source")
+    summary = result.get("fields")
+    if not isinstance(summary, dict) or summary.get("schema_version") != "todo_summary_v0":
+        raise ValueError("invalid typed Todo summary fields")
+    for name, lane in lanes.items():
+        mode = lane.get("format")
+        if mode not in {"raw", "active", "compact", "recent", "gap"}:
+            raise ValueError("invalid Todo summary display format")
+        formatted = []
+        for index in lane["indices"]:
+            item = items[index]
+            if mode == "raw":
+                compact = item
+            elif mode == "active":
+                compact = compact_active_next_action_todo_item(item)
+            elif mode in {"compact", "recent", "gap"}:
+                compact = compact_todo_item(item)
+                if mode in {"recent", "gap"}:
+                    for key in ("note", "evidence", "reason"):
+                        compact.pop(key, None)
+                if mode == "gap":
+                    compact.update(succession_tracked=True,
+                        recommended_action="record no_followup=true or add/link a successor todo")
+            else:
+                raise ValueError("invalid Todo summary display format")
+            formatted.append(compact)
+        summary[name] = formatted
+    return {"summary": summary, "items": [items[index] for index in selected],
+        "succession": [succession[index] for index in selected],
+        "orchestration": {name: [items[index] for index in indices] for name, indices in orchestration.items()}}
 
 
 def compact_todo_group(
@@ -1023,183 +1007,33 @@ def compact_evaluated_todo_group(
     """
     if not items and not include_empty_source:
         return None
-    projected = _project_summary_lanes(items, preferred_todo_ids, selection)
+    projected = _project_summary(items, preferred_todo_ids, selection=selection,
+        role=role, source_section=source_section, item_limit=item_limit, full_selection=full_selection)
     items = projected["items"]
-    if selection is not None:
-        full_selection = projected["full_selection"]
     if not items and not include_empty_source:
         return None
-    lanes = _TodoGroupLanes(**projected["lanes"])
-    from .succession_warning import project_succession
-
-    succession = project_succession(items, reuse=True)
-    successor_gap_items = completed_without_successor_items(items, evaluations=succession)
-    recent_completed_advancement_items = [
-        compact_todo_item(item)
-        for item in sorted(
-            (
-                item
-                for item in lanes.done_items
-                if todo_item_task_class(item) == TODO_TASK_CLASS_ADVANCEMENT
-                and str(item.get("completed_at") or "").strip()
-            ),
-            key=_completed_succession_sort_key,
-            reverse=True,
-        )[:MAX_RECENT_COMPLETED_ADVANCEMENT_ITEMS]
-    ]
-    for item in recent_completed_advancement_items:
-        for key in ("note", "evidence", "reason"):
-            item.pop(key, None)
-    handoff_gates = build_todo_handoff_gate_states(items, evaluations=succession)
-    watch_only_monitor_items = lanes.watch_only_monitor_items
-    watch_only_monitor_due_items = lanes.watch_only_monitor_due_items
-    convergent_open_items = lanes.convergent_open_items
-    summary: dict[str, Any] = {
-        "schema_version": "todo_summary_v0",
-        "source_section": source_section,
-        "total_count": len(items),
-        "work_counts": projected["work_counts"],
-        "open_count": len(lanes.open_items),
-        "done_count": len(lanes.terminal_items),
-        "advancement_done_count": count_advancement_todos(lanes.done_items),
-        "deferred_count": len(lanes.deferred_items),
-        "first_open_items": [
-            compact_todo_item(item) for item in lanes.projected_open_items[:3]
-        ],
-        "first_executable_items": [
-            compact_todo_item(item) for item in lanes.executable_items[:3]
-        ],
-        "monitor_open_items": [
-            compact_todo_item(item) for item in lanes.monitor_items
-        ],
-        "monitor_due_count": len(lanes.monitor_due_items),
-        "monitor_due_items": [
-            compact_todo_item(item)
-            for item in lanes.monitor_due_items[:MAX_MONITOR_DUE_ITEMS]
-        ],
-        "monitor_schedule_gap_count": len(lanes.monitor_schedule_gap_items),
-        "monitor_schedule_gap_items": [
-            compact_todo_item(item)
-            for item in lanes.monitor_schedule_gap_items[:MAX_MONITOR_DUE_ITEMS]
-        ],
-        "unclaimed_priority_open_items": [
-            compact_todo_item(item)
-            for item in lanes.unclaimed_open_items[:MAX_PROJECT_ASSET_TODO_BACKLOG_ITEMS]
-        ],
-        "claimed_open_items": [
-            compact_todo_item(item)
-            for item in claimed_visibility_items(
-                lanes.claimed_open_items,
-                limit=MAX_TODO_VISIBILITY_LANE_ITEMS,
-            )
-        ],
-        "claimed_advancement_open_items": [
-            compact_todo_item(item)
-            for item in claimed_visibility_items(
-                lanes.claimed_advancement_items,
-                limit=MAX_TODO_VISIBILITY_LANE_ITEMS,
-            )
-        ],
-        "claimed_monitor_open_items": [
-            compact_todo_item(item)
-            for item in claimed_visibility_items(
-                lanes.claimed_monitor_items,
-                limit=MAX_TODO_VISIBILITY_LANE_ITEMS,
-            )
-        ],
-        "backlog_items": [
-            compact_todo_item(item)
-            for item in lanes.projected_open_items[:MAX_PROJECT_ASSET_TODO_BACKLOG_ITEMS]
-        ],
-        "executable_backlog_items": [
-            compact_todo_item(item)
-            for item in lanes.executable_items[:MAX_PROJECT_ASSET_TODO_BACKLOG_ITEMS]
-        ],
-        "deferred_items": [
-            compact_todo_item(item)
-            for item in lanes.projected_deferred_items[:MAX_DEFERRED_TODO_VISIBILITY_ITEMS]
-        ],
-        "deferred_resume_candidates": [
-            compact_todo_item(item)
-            for item in lanes.projected_deferred_items
-            if item.get("resume_ready") is True
-        ][:MAX_DEFERRED_TODO_VISIBILITY_ITEMS],
-        "items": lanes.budgeted_items if item_limit is None else lanes.budgeted_items[:item_limit],
-    }
+    summary: dict[str, Any] = projected["summary"]
+    handoff_gates = build_todo_handoff_gate_states(items, evaluations=projected["succession"])
     attach_advancement_frontier_revision_index(summary, items, role=role)
     attach_active_vision_waits(
         summary, vision_runs, role=role, items=items,
         lineage_items=lineage_items,
     )
-    if watch_only_monitor_items:
-        summary["watch_only_monitor_count"] = len(watch_only_monitor_items)
-        summary["watch_only_monitor_due_count"] = len(watch_only_monitor_due_items)
-        summary["convergence_open_count"] = len(convergent_open_items)
-    if recent_completed_advancement_items:
-        summary["recent_completed_advancement_items"] = recent_completed_advancement_items
     if include_task_orchestration_authority:
         summary["task_orchestration_authority"] = _task_orchestration_authority(
-            lanes,
-            role=role,
-        )
-    if lanes.blocker_items:
-        summary["blocker_open_count"] = len(lanes.blocker_items)
-        summary["blocker_items"] = [
-            compact_todo_item(item) for item in lanes.blocker_items
-        ]
-    from ..effect_runtime import effect_runtime_result
-
-    replan_gates = {gate.get("todo_id") for gate in handoff_gates
-        if gate.get("route_continuation_replan_required") is True}
-    closure = effect_runtime_result("todo.succession.closure", {
-        "schema_version": "todo_closure_request_v0", "role": role,
-        "source_section": source_section, "full_selection": full_selection,
-        "rows": [{"status": item.get("status") or ("done" if item.get("done") else "open"),
-            "watch_only": projection_todo_item_is_watch_only_monitor(item),
-            "no_followup": normalize_todo_no_followup(item.get("no_followup")) is True,
-            "successor_gap": evaluation["successor_gap"], "handoff_state": evaluation["handoff_state"],
-            "replan": item.get("route_continuation_replan_required") is True or item.get("todo_id") in replan_gates}
-            for item, evaluation in zip(items, succession, strict=True)],
-    })
-    if not isinstance(closure, dict):
-        raise ValueError("invalid typed Todo closure projection")
-    summary.update(closure)
-    if lanes.resume_blocked_items:
-        summary["resume_blocked_count"] = len(lanes.resume_blocked_items)
-        summary["resume_blocked_items"] = [
-            compact_todo_item(item)
-            for item in lanes.resume_blocked_items[:MAX_DEFERRED_TODO_VISIBILITY_ITEMS]
-        ]
+            projected["orchestration"], role=role)
     if handoff_gates:
         summary["handoff_gates"] = handoff_gates
-    if successor_gap_items:
-        compact_gap_items = successor_gap_items[:MAX_COMPLETED_SUCCESSION_WARNING_ITEMS]
-        summary["completed_without_successor_count"] = len(successor_gap_items)
-        summary["completed_without_successor_items"] = compact_gap_items
+    if summary.get("completed_without_successor_count"):
         summary["todo_succession_warning"] = {
             "schema_version": TODO_SUCCESSION_WARNING_SCHEMA_VERSION,
             "reason_code": TODO_SUCCESSION_WARNING_REASON_CODE,
-            "count": len(successor_gap_items),
-            "items": compact_gap_items,
+            "count": summary["completed_without_successor_count"],
+            "items": summary["completed_without_successor_items"],
             "recommended_action": (
                 "run loopx todo complete --no-follow-up for the completed Todo, "
                 "or add/link a successor Todo before closing the slice; do not "
                 "invent a user gate"
             ),
         }
-    if lanes.active_next_action_items:
-        summary["active_next_action_items"] = [
-            compact_active_next_action_todo_item(item)
-            for item in lanes.active_next_action_items
-        ]
-    if lanes.active_next_action_executable_items:
-        summary["active_next_action_executable_items"] = [
-            compact_active_next_action_todo_item(item)
-            for item in lanes.active_next_action_executable_items
-        ]
-    if lanes.claimed_open_items:
-        summary["claimed_open_count"] = len(lanes.claimed_open_items)
-        summary["unclaimed_open_count"] = len(lanes.open_items) - len(lanes.claimed_open_items)
-        summary["claimed_advancement_open_count"] = len(lanes.claimed_advancement_items)
-        summary["claimed_monitor_open_count"] = len(lanes.claimed_monitor_items)
     return summary
diff --git a/loopx/status.py b/loopx/status.py
index 96290fa032..26ee24cf16 100644
--- a/loopx/status.py
+++ b/loopx/status.py
@@ -16,20 +16,20 @@
     collect_status as _collect_status_read_model,
 )
 from .control_plane.status.active_state_projection import (
-    STATE_EVENT_LOG_BASENAME,
+    STATE_EVENT_LOG_BASENAME as STATE_EVENT_LOG_BASENAME,
 )
 from .control_plane.status.contract_projection import (
-    STATUS_CONTRACT_RELOAD_HINT,
+    STATUS_CONTRACT_RELOAD_HINT as STATUS_CONTRACT_RELOAD_HINT,
 )
 from .control_plane.status.goal_attention_projection import (
-    PLANNED_CONTROLLER_OPT_IN_RECOMMENDED_ACTION,
+    PLANNED_CONTROLLER_OPT_IN_RECOMMENDED_ACTION as PLANNED_CONTROLLER_OPT_IN_RECOMMENDED_ACTION,
 )
 # Refs #4447: one definition for this vocabulary. The control_plane projection
 # owns it because it feeds the monitor/attention read models; this module keeps
 # re-exporting each name for existing callers instead of restating its value.
 from .control_plane.status.monitor_display_projection import (
-    MONITOR_DISPLAY_FALLBACK_ACTION,
-    MONITOR_DISPLAY_STOP_CONDITION,
+    MONITOR_DISPLAY_FALLBACK_ACTION as MONITOR_DISPLAY_FALLBACK_ACTION,
+    MONITOR_DISPLAY_STOP_CONDITION as MONITOR_DISPLAY_STOP_CONDITION,
     MONITOR_SIGNAL_WAITING_ON,
 )
 from .control_plane.status.registry_health_projection import (
@@ -189,7 +189,6 @@
     active_state_todo_attention_item as _active_state_todo_attention_item_read_model,
     active_next_action_todo_ids,
     attach_dependency_blockers,
-    claimed_visibility_items as claimed_visibility_items,
     compact_todo_group as compact_todo_group,
     compact_todo_item as compact_todo_item,
     first_open_todo_text,
@@ -234,7 +233,6 @@
     "TODO_PROJECTION_DETAIL_POINTER_SCHEMA_VERSION": "loopx.control_plane.work_items.project_asset",
     "TODO_PROJECTION_VIEW_SCHEMA_VERSION": "loopx.control_plane.work_items.project_asset",
     "project_asset_summary_is_public_safe": "loopx.control_plane.work_items.project_asset",
-    "claimed_visibility_items": "loopx.control_plane.todos.todo_summary",
     "compact_todo_group": "loopx.control_plane.todos.todo_summary",
     "compact_todo_item": "loopx.control_plane.todos.todo_summary",
     "todo_lane_items": "loopx.control_plane.todos.todo_summary",
diff --git a/tests/control_plane/test_todo_consumer_scope.py b/tests/control_plane/test_todo_consumer_scope.py
index f8fbaa0a49..1bae6b3aa5 100644
--- a/tests/control_plane/test_todo_consumer_scope.py
+++ b/tests/control_plane/test_todo_consumer_scope.py
@@ -97,16 +97,20 @@ def test_filter_composes_with_existing_lane_call_and_rejects_downgraded_response
     original = effect_runtime.effect_runtime_result
     requests = []
     def track(method, request, **kwargs):
-        if method == "todo.summary_lanes.project":
+        if method == "todo.summary.project":
             requests.append(request)
         return original(method, request, **kwargs)
     monkeypatch.setattr(effect_runtime, "effect_runtime_result", track)
     assert filtered_todo_summary(source, role="user", agent_id="agent-a")["total_count"] == 1
     assert len(requests) == 1
-    assert requests[0]["schema_version"] == "todo_summary_lanes_request_v1"
+    assert requests[0]["schema_version"] == "todo_summary_projection_request_v1"
+    # One whole-source batch stays columnar, so adding fields cannot silently
+    # push a long-history request past the runtime request budget.
+    assert requests[0]["columns"][:3] == ["status", "done", "task_class"]
+    assert all(len(cells) == len(requests[0]["columns"]) for cells in requests[0]["rows"])
     def downgrade(method, request, **kwargs):
         result = original(method, request, **kwargs)
-        if method == "todo.summary_lanes.project":
+        if method == "todo.summary.project":
             result.pop("source_indices", None)
         return result
     monkeypatch.setattr(effect_runtime, "effect_runtime_result", downgrade)
@@ -127,8 +131,8 @@ def test_typed_lane_response_cannot_alias_or_escape_selected_source(monkeypatch,
     original = effect_runtime.effect_runtime_result
     def corrupt(method, request, **kwargs):
         result = original(method, request, **kwargs)
-        if method == "todo.summary_lanes.project":
-            result["lanes"]["open_items"] = {"duplicate": [0, 0], "boolean": [False], "outside_selection": [outside]}[corruption]
+        if method == "todo.summary.project":
+            result["lanes"]["first_open_items"]["indices"] = {"duplicate": [0, 0], "boolean": [False], "outside_selection": [outside]}[corruption]
         return result
     monkeypatch.setattr(effect_runtime, "effect_runtime_result", corrupt)
     with pytest.raises(ValueError, match="source ordinal|escaped the selected source"):
diff --git a/tests/control_plane/test_todo_summary_projection.py b/tests/control_plane/test_todo_summary_projection.py
new file mode 100644
index 0000000000..0eac0bfc2e
--- /dev/null
+++ b/tests/control_plane/test_todo_summary_projection.py
@@ -0,0 +1,86 @@
+"""Whole-source summary decisions precede display limits and preserve chronology."""
+import json
+
+from loopx.control_plane.todos.todo_summary import compact_todo_group
+
+
+def row(index, **fields):
+    return {"todo_id": f"todo_summary_{index}", "text": f"Work {index}", "role": "agent",
+            "task_class": "advancement_task", "status": "open", "index": index,
+            "source_section": "Agent Todo", **fields}
+
+
+def summarize(items, **kwargs):
+    return compact_todo_group(items, role="agent", source_section="Agent Todo", **kwargs)
+
+
+def test_recent_completions_compare_instants_not_offset_strings():
+    items = [row(1, status="done", no_followup=True, completed_at="2026-01-01T10:00:00+08:00"),
+             row(2, status="done", no_followup=True, completed_at="2026-01-01T03:00:00Z")]
+    result = summarize(items)
+    assert [item["todo_id"] for item in result["recent_completed_advancement_items"]] == [
+        "todo_summary_2", "todo_summary_1"]
+
+
+def test_recent_completions_preserve_microseconds_across_offsets():
+    items = [row(2, status="done", no_followup=True, completed_at="2026-01-01T10:00:00.000001+08:00"),
+             row(1, status="done", no_followup=True, completed_at="2026-01-01T02:00:00.000002Z")]
+    result = summarize(items)
+    assert [item["todo_id"] for item in result["recent_completed_advancement_items"]] == [
+        "todo_summary_1", "todo_summary_2"]
+
+
+def test_invalid_completion_time_does_not_displace_known_recent_work():
+    result = summarize([row(1, status="done", no_followup=True, completed_at="unknown"),
+                        row(2, status="done", no_followup=True, completed_at="2026-01-01T03:00:00Z")])
+    assert [item["todo_id"] for item in result["recent_completed_advancement_items"]] == ["todo_summary_2"]
+    assert result["advancement_done_count"] == 2  # Still retained as completed work.
+
+
+def test_summary_caps_do_not_change_work_counts_or_hide_a_peer():
+    items = [row(i, claimed_by="agent-a" if i < 24 else "agent-b") for i in range(32)]
+    result = summarize(items, item_limit=1)
+    assert len(result["items"]) == 1 and result["work_counts"]["advancement"] == 32
+    assert len(result["claimed_open_items"]) == 16
+    assert {item["claimed_by"] for item in result["claimed_open_items"]} == {"agent-a", "agent-b"}
+    assert result["claimed_open_count"] == 32
+
+
+def test_late_edit_does_not_make_an_old_completion_recent():
+    result = summarize([row(1, status="done", no_followup=True,
+                            completed_at="2026-01-01T00:00:00Z", updated_at="2026-09-01T00:00:00Z"),
+                        row(2, status="done", no_followup=True,
+                            completed_at="2026-02-01T00:00:00Z")])
+    assert [item["todo_id"] for item in result["recent_completed_advancement_items"]] == [
+        "todo_summary_2", "todo_summary_1"]
+
+
+def test_a_selection_cannot_restore_a_lost_full_source_proof():
+    from loopx.control_plane.todos.todo_summary import compact_evaluated_todo_group
+    source = summarize([row(1, status="done", no_followup=True)], item_limit=None)
+    result = compact_evaluated_todo_group(source["items"], source_section="Agent Todo", role="agent",
+        full_selection=False, selection={"role": "agent", "status": None, "todo_id": None, "agent_id": None})
+    assert "source_proof" not in result and "terminal_closure_proof" not in result
+    assert result["done_count"] == 1
+
+
+def test_long_history_stays_inside_the_runtime_request_budget(monkeypatch):
+    """A whole-source batch must not outgrow the co-deployed runtime's request."""
+    from loopx.control_plane import effect_runtime
+    from loopx.control_plane.effect_runtime import MAX_REQUEST_BYTES
+    requests = []
+    original = effect_runtime.effect_runtime_result
+
+    def track(method, request, **kwargs):
+        if method == "todo.summary.project":
+            requests.append(request)
+        return original(method, request, **kwargs)
+
+    monkeypatch.setattr(effect_runtime, "effect_runtime_result", track)
+    items = [row(index, status="done", no_followup=True,
+                 completed_at="2026-01-01T00:00:00Z") for index in range(4096)]
+    result = summarize(items, item_limit=None)
+    assert result["done_count"] == 4096
+    request = requests[0]
+    encoded = json.dumps(request, separators=(",", ":")).encode()
+    assert len(encoded) < MAX_REQUEST_BYTES
diff --git a/tests/control_plane_ts/todo_consumer_scope_conformance.ts b/tests/control_plane_ts/todo_consumer_scope_conformance.ts
index 19e5754403..c44c08f2d3 100644
--- a/tests/control_plane_ts/todo_consumer_scope_conformance.ts
+++ b/tests/control_plane_ts/todo_consumer_scope_conformance.ts
@@ -1,6 +1,7 @@
 import assert from "node:assert/strict";
 import {spawnSync} from "node:child_process";
 import test from "node:test";
+import type {JsonObject} from "../../loopx/control_plane/effect_program.ts";
 import type {AuthorityStoreConformanceFactory} from "./authority_store_conformance.ts";
 import {productionScaleConsumerScopeFixture} from "./production_scale_coordination_fixture.ts";
 
@@ -24,6 +25,7 @@ prefix=lambda values: sorted(row['todo_id'] for row in values if row['todo_id'].
 result={'selected':prefix(selected['items']), 'quota_delta':quota['open_count']-base_quota['open_count'],
  'limited':len(limited['items']), 'counts_equal':selected['total_count']==limited['total_count'],
  'whole':prefix(summary['items']),
+ 'recent':[row['todo_id'] for row in fields['agent_todos'].get('recent_completed_advancement_items', [])[:2]],
  'filtered_peer':filtered_todo_summary(summary,role='user',agent_id='agent-a',todo_id='todo_scope_peer_gate')['items'],
  'succession_gap':filtered_todo_summary(fields['agent_todos'],role='agent',todo_id=p['cases']['inferred_source']).get('completed_without_successor_count',0)}
 print(json.dumps(result))
@@ -32,6 +34,12 @@ export function registerTodoConsumerScopeConformance(name: string, factory: Auth
   for (const schema of ["native", "legacy"] as const) test(`${name}: full-source Agent read addressing (${schema})`, async context => {
     const {store} = await factory(context);
     const {projection, cases} = productionScaleConsumerScopeFixture("consumer-scope", schema);
+    const recent = (projection.todos as JsonObject[]).filter(todo => todo.status === "done" &&
+      todo.task_class === "advancement_task" && todo.archive_state !== "archive").slice(0, 2);
+    assert.equal(recent.length, 2);
+    recent[0].completed_at = "2099-01-01T10:00:00.000001+08:00";
+    recent[0].updated_at = "2099-12-01T00:00:00Z";
+    recent[1].completed_at = "2099-01-01T02:00:00.000002Z";
     assert.equal((await store.commitAuthority({operation_id: "scope-source", expected_provider_revision: null,
       next_projection: projection, events: [], receipts: []})).status, "applied");
     const before = await store.loadAuthority(); assert.equal(before.status, "loaded");
@@ -45,6 +53,7 @@ export function registerTodoConsumerScopeConformance(name: string, factory: Auth
     assert.equal(result.whole.length, 5); assert.deepEqual(result.filtered_peer, []);
     assert.equal(result.limited, 1); assert.equal(result.counts_equal, true);
     assert.equal(result.succession_gap, 0);
+    assert.deepEqual(result.recent, [recent[1].todo_id, recent[0].todo_id]);
     assert.deepEqual(await store.loadAuthority(), before, "read consumers must never mutate authority");
   });
 }
diff --git a/tests/control_plane_ts/todo_summary_projection.test.ts b/tests/control_plane_ts/todo_summary_projection.test.ts
new file mode 100644
index 0000000000..53e4c91472
--- /dev/null
+++ b/tests/control_plane_ts/todo_summary_projection.test.ts
@@ -0,0 +1,105 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import type {JsonObject} from "../../loopx/control_plane/effect_program.ts";
+import {TODO_SUMMARY_PROJECTION_COLUMNS, projectTodoSummary} from "../../loopx/control_plane/todos/summary_projection.ts";
+import {productionScaleCoordinationFixture} from "./production_scale_coordination_fixture.ts";
+
+const row = (fields: JsonObject = {}): JsonObject => ({status: "open", done: false,
+  task_class: "advancement_task", has_resume: false, resume_ready: null, resume_evaluated: false,
+  acceptance_blocked: false, claimed: false, claim: null, preferred: false, watch_only: false,
+  due_at: null, expires_at: null, sort: [1, 1, "", ""], todo_id: null, bound: null,
+  blocks: null, global: false, excluded: [], completed_at: null, updated_at: null,
+  completion_index: 0, linked_user_action: false, no_followup: false,
+  successor_gap: false, handoff_state: null, replan: false, ...fields});
+const request = (rows: JsonObject[], fields: JsonObject = {}): JsonObject => ({
+  schema_version: "todo_summary_projection_request_v1",
+  columns: [...TODO_SUMMARY_PROJECTION_COLUMNS],
+  rows: rows.map(row => TODO_SUMMARY_PROJECTION_COLUMNS.map(name => row[name] ?? null)),
+  observed_at: 100,
+  selection: null, role: "agent", source_section: "Agent Todo", item_limit: 12, full_selection: true, ...fields});
+const done = (fields: JsonObject = {}) => row({status: "done", done: true, no_followup: true, ...fields});
+
+test("one whole-source projection computes counts, visibility and closure before limits", () => {
+  for (const limit of [null, 0, 1, 12]) {
+    const rows = Array.from({length: 32}, (_, index) => row({claimed: true, claim: index < 24 ? "a" : "b"}));
+    const before = structuredClone(rows), result = projectTodoSummary(request(rows, {item_limit: limit}));
+    assert.equal(result.fields.open_count, 32);
+    assert.equal((result.fields.work_counts as JsonObject).advancement, 32);
+    assert.equal(result.fields.claimed_open_count, 32);
+    assert.deepEqual(result.lanes.claimed_open_items.indices, [...Array(8).keys(), ...Array.from({length: 8}, (_, i) => 24 + i)]);
+    assert.equal(result.lanes.items.indices.length, limit === null ? 32 : limit);
+    assert.equal(result.fields.terminal_closure_proof, undefined);
+    assert.deepEqual(rows, before);
+  }
+});
+
+test("recent completion orders true microsecond instants and ignores later edits", () => {
+  const rows = [done({completed_at: "2026-01-01T10:00:00.000001+08:00", updated_at: "2026-12-01T00:00:00Z"}),
+    done({completed_at: "2026-01-01T02:00:00.000002Z"}), done({completed_at: "invalid"})];
+  const result = projectTodoSummary(request(rows));
+  assert.deepEqual(result.lanes.recent_completed_advancement_items.indices, [1, 0]);
+  assert.equal(result.fields.advancement_done_count, 3);
+  assert.equal(result.lanes.items.indices.length, 3);
+});
+
+test("equal instants retain reverse source coordinate and stable ties", () => {
+  const rows = [done({completed_at: "2026-01-01T10:00:00+08:00", completion_index: 2}),
+    done({completed_at: "2026-01-01T02:00:00Z", completion_index: 4}),
+    done({completed_at: "2026-01-01T02:00:00Z", completion_index: 4})];
+  assert.deepEqual(projectTodoSummary(request(rows)).lanes.recent_completed_advancement_items.indices, [1, 2, 0]);
+});
+
+test("selection cannot turn partial source knowledge into a closure proof", () => {
+  const select = {role: "agent", status: null, todo_id: null, agent_id: null};
+  const full = projectTodoSummary(request([done()], {selection: select}));
+  assert.ok(full.fields.terminal_closure_proof);
+  const partial = projectTodoSummary(request([done()], {selection: select, full_selection: false}));
+  assert.equal(partial.full_selection, false);
+  assert.equal(partial.fields.source_proof, undefined);
+  assert.equal(partial.fields.terminal_closure_proof, undefined);
+  const filtered = projectTodoSummary(request([done(), row()], {selection: {...select, status: "done"}}));
+  assert.equal(filtered.fields.terminal_closure_proof, undefined);
+  assert.deepEqual(filtered.source_indices, [0]);
+});
+
+test("scope preserves original ordinals and closure uses only the selected graph decisions", () => {
+  const rows = [row({todo_id: "peer", claimed: true, claim: "other"}),
+    done({todo_id: "own", claimed: true, claim: "me"})];
+  const result = projectTodoSummary(request(rows, {selection: {role: "agent", status: null, todo_id: null, agent_id: "me"}}));
+  assert.deepEqual(result.source_indices, [1]);
+  assert.deepEqual(result.lanes.items.indices, [1]);
+  assert.equal(result.fields.done_count, 1);
+  assert.equal(result.fields.terminal_closure_proof, undefined);
+});
+
+test("invalid source or budgets fail closed rather than hiding rows", () => {
+  for (const fields of [{item_limit: -1}, {item_limit: 1.5}, {item_limit: true}, {full_selection: null}]) {
+    assert.throws(() => projectTodoSummary(request([row()], fields)));
+  }
+  for (const fields of [{claimed: true}, {claim: "agent"}, {done: true}, {completed_at: 2},
+    {has_resume: true}, {successor_gap: "false"}, {handoff_state: "unrecognized"}]) {
+    assert.throws(() => projectTodoSummary(request([row(fields)])));
+  }
+});
+
+test("large native and imported corpora preserve source coverage across every display cap", () => {
+  for (const format of ["native", "legacy"] as const) {
+    const fixture = productionScaleCoordinationFixture("summary-source", format);
+    const records = fixture.projection.todos as JsonObject[];
+    const rows = records.map((todo, ordinal) => row({status: todo.status, done: todo.done,
+      task_class: todo.task_class, sort: [1, ordinal, "", ""], todo_id: todo.todo_id,
+      claim: todo.claimed_by ?? null, claimed: Boolean(todo.claimed_by)}));
+    const full = projectTodoSummary(request(rows, {item_limit: null}));
+    assert.equal(full.fields.total_count, records.length);
+    assert.equal(full.lanes.items.indices.length, records.length);
+    for (const limit of [0, 1, 12]) {
+      const limited = projectTodoSummary(request(rows, {item_limit: limit}));
+      assert.deepEqual(limited.fields, full.fields);
+      assert.equal(limited.lanes.items.indices.length, limit);
+      for (const lane of Object.values(limited.lanes)) {
+        assert.equal(new Set(lane.indices).size, lane.indices.length);
+        assert.ok(lane.indices.every(index => index >= 0 && index < records.length));
+      }
+    }
+  }
+});