diff --git a/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-09-19-summary-work-counts.md b/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-09-19-summary-work-counts.md
new file mode 100644
index 0000000000..e6dae95229
--- /dev/null
+++ b/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-09-19-summary-work-counts.md
@@ -0,0 +1,15 @@
+# Summary work-count consumer closure (2026-09-19)
+
+| Boundary | Delivered evidence |
+| --- | --- |
+| Goal/source | Overall roadmap R5/S2, TS T3 and shared-authority L5; baseline `96d98f3d4`. |
+| Observable gap | A complete source with 21 actionable advancement Todos becomes 8 when a consumer counts the bounded backlog; quota payload compaction reduces the apparent count further. A different legacy fallback classifies unseen rows as advancement. |
+| Owning change | `todos/summary_lanes.ts` selects lanes in one typed batch and supplies pre-limit work counts. Existing quota selection reuses the count owner after Agent filtering. Python retains normalization, timestamp/presentation adaptation and ordinal readback; its lane-selection and hidden-work inference loops are removed. |
+| Semantics | Complete counts survive list/status/quota compaction. Incomplete scope knowledge survives repeated projection; contradictory legacy fragments cannot prove completeness. Unknown tasks are not inferred as executable. Canonical `todo list` carries the read revision's acceptance guard, matching status without hiding held records. |
+| Compatibility | Full baseline/candidate role and scoped summaries agree apart from the disclosed additive counts; completed/deferred conventions, ordering, Monitor timing and successor/closure policies remain. Empty canonical sources stay authoritative and unavailable providers cannot fall back to Markdown. |
+| Real paths | File/SQLite CLI and installed-wheel CLI/Chat HTTP readback; isolated real PostgreSQL authority/service reads; both complete synthetic record schemas and an authorized frozen full-graph snapshot. Existing heads/Todos/leases and independent display bytes remain unchanged. |
+| Cost | Two compact ordinal-planning requests per two-role summary; no per-Todo RPC and no new persisted queue or provider state. The whole Python summary adapter still has other rule/effect callers. |
+| Remaining boundary | This closes the count consumer within L5, not permanent projection freshness, all T3 sources, event callers, executor-held effect fences, D2 capacity/elapsed soak or D3 integrated promotion. No default change or old-writer retirement is claimed. |
+
+See [the read contract](../../../../reference/todo-work-counts.md) for field
+meaning, public read commands, incomplete-source behavior and rollback.
diff --git a/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-09-19-summary-work-counts.zh-CN.md b/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-09-19-summary-work-counts.zh-CN.md
new file mode 100644
index 0000000000..f4bb2d341d
--- /dev/null
+++ b/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-09-19-summary-work-counts.zh-CN.md
@@ -0,0 +1,15 @@
+# 摘要工作计数消费者闭合（2026-09-19）
+
+| 边界 | 交付证据 |
+| --- | --- |
+| 目标 | 总纲 R5/S2、TS T3、shared-authority L5；基线 `96d98f3d4`。 |
+| 可观察缺口 | 完整来源中 21 条可执行推进 Todo 因 backlog 展示上限被下游报成 8 条，quota 压缩进一步缩小数量；另一旧路径把未看到的任务猜成 advancement。 |
+| 唯一 owner | `todos/summary_lanes.ts` 批量选择 lane 并计算裁剪前数量；已有 quota selection 在 Agent 筛选后复用计数 owner。Python 保留规范化、时间／展示适配及索引回读，删除原 lane 选择和隐藏任务推断循环。 |
+| 语义修复 | list/status/quota 压缩保留完整计数；重复投影保留来源不完整状态；矛盾旧片段不能证明完整，未知任务不再被推断为可执行。canonical `todo list` 保留同版本 acceptance 限制，与 status 一致但不隐藏受阻记录。 |
+| 兼容 | 除已声明新增计数，基线／候选的完整 role 与作用域摘要一致；完成／延期约定、排序、Monitor 时间规则及 successor／closure 策略保留。空 canonical 来源仍有权威性，provider 不可用时不回退 Markdown。 |
+| 真实路径 | File/SQLite CLI、wheel 安装后的 CLI／真实 Chat HTTP、隔离真实 PostgreSQL authority/service 读取；两种完整合成记录 schema 与授权冻结图快照。原 head/Todo/lease 及独立展示字节不变。 |
+| 成本 | 双 role 摘要增加两次紧凑索引规划请求，没有逐 Todo RPC、新持久队列或 provider 状态。Python 完整摘要 adapter 仍有其他规则／效果调用，不能整体删除。 |
+| 剩余 | 本次只闭合 L5 计数消费者；永久投影新鲜度、其他 T3 来源、event caller、执行器 effect fence、D2 容量／真实时间 soak 及 D3 集成晋升仍未完成。不改变默认 provider，也不宣称旧 writer 已退出。 |
+
+[读取合同](../../../../reference/todo-work-counts.md)说明字段、公有读取命令、
+不完整来源语义和回滚边界。
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 de7e65bd5f..6d354c00bb 100644
--- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
+++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
@@ -2957,6 +2957,8 @@ source paths, authorize monitor writeback, or change provider/promotion holds.
 
 **D1 — qualify permanent projection delivery; may overlap T1/T2.**
 
+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 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).
 
 The D1 document-ownership slice gives readers, editors and projection one visible-region
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 33c3aa55b6..e590669d0e 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
@@ -2340,6 +2340,8 @@ Scoped fallback 的选择与门禁关系也已复用同一 TS decision owner，
 
 **D1 — 资格化永久投影交付，可与 T1/T2 重叠推进。**
 
+摘要与 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)。
 
 D1 的文档归属切片把读取、编辑与投影放到同一可见区域／Todo 行解码边界，修复
diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
index bbd8e37082..e3e1fa9109 100644
--- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
+++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
@@ -738,6 +738,15 @@ all T2 commands or authorize whole-Goal promotion.
 
 **T3 — close remaining structured consumers, then remove their old reads.**
 
+Todo summary lanes and pre-limit work counts now share `todos/summary_lanes.ts`.
+Python's lane classification and hidden-work inference loops are removed; quota
+recomputes counts after scope selection and carries incomplete source knowledge
+through compaction/reprojection. Public canonical Todo lists retain the same
+revision's acceptance guard. See [count semantics](../../reference/todo-work-counts.md).
+This closes the summary-to-work-lane count consumer, not every T3 source or D1
+projection delivery; legacy codecs/renderers and other summary policies remain.
+
+
 Goal Channel ownership observation now reads a complete canonical Todo/lease revision and shares one TS batch policy with the legacy adapter. It retires display-layer lease time/generation/conflict decisions and local-file reads after promotion. Empty, unavailable and truncated observations remain distinct; see [coordination observation](../../reference/coordination-observation.md). This closes the Goal Channel ownership reader, not other channel panels or whole-Goal promotion.
 
 The D1 document-ownership slice gives readers, editors and projection one visible-region
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 98d69fabd3..551d2b57fb 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
@@ -577,6 +577,13 @@ delivery pending；这不代表全部 T2 命令或整 Goal promotion 已完成
 
 **T3 — 闭合剩余 structured consumer，删除各自旧读路径。**
 
+Todo 摘要 lane 与裁剪前工作计数现共用 `todos/summary_lanes.ts`，删除 Python 的
+lane 分类和隐藏任务推断循环。quota 在作用域筛选后重新计数，不完整来源状态贯穿
+压缩与重复投影；公开 canonical Todo 列表保留同版本 acceptance 限制。见
+[计数语义](../../reference/todo-work-counts.md)。本切片闭合摘要到 work-lane 的计数
+消费者，不代表所有 T3 来源或 D1 展示交付完成；旧格式解码、renderer 及其他摘要策略仍保留。
+
+
 Goal Channel 所有权观察现从完整 canonical Todo／lease revision 读取，并与 legacy adapter 共用 TS 批量规则；删除展示层的时间／代数／冲突判断和晋升后的本地文件读路径。空值、不可用与截断分别披露，见 [coordination observation](../../reference/coordination-observation.md)。这只闭合所有权观察 reader，不宣称其余面板或整 Goal 晋升完成。
 
 D1 的文档归属切片把读取、编辑与投影放到同一可见区域／Todo 行解码边界，修复
diff --git a/docs/reference/contracts/interface-budget-contract.md b/docs/reference/contracts/interface-budget-contract.md
index 8d6d0e0e11..efc5d1358c 100644
--- a/docs/reference/contracts/interface-budget-contract.md
+++ b/docs/reference/contracts/interface-budget-contract.md
@@ -10,7 +10,7 @@ and size/count budgets.
 | --- | --- | --- | --- | --- | --- | --- |
 | `heartbeat_prompt_json` | heartbeat automation | wake and route one bounded turn | `quota should-run`, `status`, or `review-packet --handoff-only` | `json_chars <= 4800` plus `interface_budget.within_budget=true` | `nested_keys <= 40` | `top_level_keys <= 30` |
 | `review_packet_handoff_only_json` | project-agent handoff | forward the smallest sufficient task packet | full `review-packet` or run-history artifact | `json_chars <= 3000` plus `handoff_interface_budget.within_budget=true` | `nested_keys <= 40` | `top_level_keys <= 18` |
-| `quota_should_run_json` | quota guard | decide whether the selected goal may spend compute | `status`, `history`, or active state | `json_chars <= 14000` | `nested_keys <= 350` | `top_level_keys <= 52` |
+| `quota_should_run_json` | quota guard | decide whether the selected goal may spend compute | `status`, `history`, or active state | `json_chars <= 14500` | `nested_keys <= 360` | `top_level_keys <= 52` |
 | `dashboard_status_json` | operator dashboard | render first-screen operator state | `history`, run artifacts, or project-local adapter output | `json_chars <= 19500` | `nested_keys <= 260` | `top_level_keys <= 25` |
 
 These four budgets measure compact machine payloads. For
@@ -52,6 +52,19 @@ route, pending-selection qualification, and hard-lane preemption evidence. The
 budget retains modest headroom for those enforceable semantics; repeated action
 details and command prefixes still belong in compact references or cold paths.
 
+The work-count projection adds scope and completeness facts that a bounded Todo
+list cannot supply. Its observed-row count is derived from `open - hidden`,
+rather than repeated in the wire object. The quota ceiling moves from 14,000
+to 14,500 characters and from 350 to 360 nested keys to retain modest headroom
+for this useful semantic growth; the top-level ceiling stays 52. Existing
+repeated Todo bodies across named lanes have distinct consumers and cannot be
+removed without a separately validated caller migration.
+
+工作计数增加了展示列表无法提供的完整性与作用域信息；已观察行数由 `open - hidden`
+推导，不重复传输。quota 字符预算从 14,000 调至 14,500，嵌套键从 350 调至 360，
+保留适量余量；顶层键上限仍为 52。不同 lane 重复携带的 Todo 有既有消费者，后续
+去重应配合调用方迁移，不能仅为通过尺寸测试而删除。
+
 | Emitted Surface | Default Qualification | Scale / Limit Contract | Cold Path |
 | --- | --- | --- | --- |
 | `start-goal --guided` | baseline and growth | small, crowded, and multi-agent goals; objective/command duplication | `packet_summary.detail_refs` and `bootstrap-command-pack` |
diff --git a/docs/reference/todo-work-counts.md b/docs/reference/todo-work-counts.md
new file mode 100644
index 0000000000..b941a72c9f
--- /dev/null
+++ b/docs/reference/todo-work-counts.md
@@ -0,0 +1,79 @@
+# Todo work counts and bounded display
+
+Todo lists, status and quota summaries carry `work_counts` with schema
+`todo_work_counts_v0`. Counts are computed before display limits; quota
+recomputes them **after** the existing Agent scope and resume selection.
+The contract is read-only. A count never grants a claim, lease, capability,
+validation exemption or execution permission.
+
+```sh
+loopx --format json todo list --goal-id example --role agent --limit 1 --thin
+loopx --format json status --goal-id example
+loopx --format json quota should-run --goal-id example --agent-id agent-a
+```
+
+The first command returns one Todo while its role summary retains the matched
+source's counts. A filtered list describes its filtered source. This is not a
+new provider setting: legacy inputs and promoted File/SQLite inputs share the
+same typed summary owner. PostgreSQL uses the same provider-neutral records;
+service deployment and whole-Goal promotion remain separately qualified.
+
+| Field | Meaning |
+| --- | --- |
+| `open` | Nonterminal source rows, including blocked work; it is not executable work |
+| `advancement` | Observed actionable advancement rows; acceptance-denied and unsatisfied resume rows do not qualify |
+| `monitor` | Observed actionable Monitor rows, including future/expired observation context; due/schedule-gap fields retain their existing separate meanings |
+| `hidden` | Declared source rows not available for classification, not rows hidden by UI pagination |
+| `complete` | Whether the available source covers the declared scope; false counts are lower bounds for classified task kinds |
+| `agent_id` | Agent execution scope, or null for an unscoped/role summary |
+
+The observed row count is derived as `open - hidden`; the payload does not
+repeat it as a second value that could drift.
+
+`complete=false` survives repeated quota projection, even when the surviving
+subset fits on one screen. It cannot certify “Monitor-only work remains.”
+Legacy display-only inputs are deduplicated by Todo identity; contradictory
+fragments cannot certify completeness. A missing task is never guessed to be
+advancement work. Invalid count envelopes and differently scoped count reuse
+fail explicitly.
+
+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.
+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
+existing summary contract.
+
+Intentional corrections: 21 executable Todos no longer become 8 because the
+backlog display limit is 8; a compact quota payload no longer turns that count
+into 2. Unknown hidden rows are not classified. Public canonical `todo list`
+also retains the acceptance guard from the same read revision, so held work
+cannot appear executable there while status says it is held. Acceptance-off
+reads keep their existing selection behavior.
+
+Markdown stays a permanent display. These reads do not rewrite stale/missing
+Markdown, create receipts or mutate canonical records. The additive count
+field is not persisted in Todo authority. Older readers can ignore it, but
+retain their old undercount behavior; rollback does not require data migration.
+T1/T2 caller closure, D1 projection recovery, D2 capacity/elapsed soak and D3
+fenced whole-Goal cutover remain separate work.
+
+## 中文说明
+
+`work_counts` 由完整来源计算，随后才裁剪展示。Agent quota 先按原有归属、排除、
+能力与作用域规则筛选，再重新计数，不能沿用整个 Goal 的数量。`open` 包含 blocked
+任务；`advancement` 才是已观察到的可执行推进任务。Monitor 是否到期仍使用独立字段。
+
+`hidden` 表示未取得、无法分类的来源行，不是界面折叠的行数。`complete=false` 时，
+已分类数量只是下界，重复投影也不能把未知变成完整，更不能据此声称“只剩 Monitor”。
+旧摘要按 Todo 身份去重，矛盾片段不能证明完整；缺失任务不再被猜成 advancement。
+
+TS 统一批量 lane 分类与计数，Python 保留旧格式解码、时间适配和展示。返回数组位置
+索引减少同一任务在多个 lane 的重复传输；一个批次使用同一观察时刻。排序、延期与
+完成计数约定、claim 展示保留。canonical `todo list` 同时修复了漏传同版本 acceptance
+限制的问题，验收受阻的任务仍可见，但不会被列为可执行。
+
+这不改变 provider 默认值，不授予执行权限，不写回 Markdown 或 canonical 状态。
+新增计数字段不进入持久化 Todo；回滚无需数据迁移。默认切换、存量迁移、D1–D3 和旧
+Python writer 退出仍有各自的验收条件，不能按本 PR 合并数量推定完成。
diff --git a/examples/control_plane/cli-output-probe-runner.py b/examples/control_plane/cli-output-probe-runner.py
index 878187ccf1..4e1ca37bad 100644
--- a/examples/control_plane/cli-output-probe-runner.py
+++ b/examples/control_plane/cli-output-probe-runner.py
@@ -116,6 +116,11 @@ def _receipt_row(
             if isinstance(payload, dict)
             else []
         ),
+        "todo_work_counts_schema_versions": (
+            semantics.todo_work_counts_schema_versions(payload)
+            if isinstance(payload, dict)
+            else []
+        ),
     }
 
 
diff --git a/examples/control_plane/hot-path-interface-budget-smoke.py b/examples/control_plane/hot-path-interface-budget-smoke.py
index 1ee80ecfeb..5172693732 100644
--- a/examples/control_plane/hot-path-interface-budget-smoke.py
+++ b/examples/control_plane/hot-path-interface-budget-smoke.py
@@ -70,8 +70,10 @@
         "cold_path": "status, history, or active state",
         # Codex keeps a lossless codex_app compatibility alias while the
         # provider-neutral app_automation packet becomes canonical.
-        "max_json_chars": 14_000,
-        "max_nested_keys": 350,
+        # Pre-limit work counts add useful scope/completeness evidence; allow
+        # modest headroom after removing the redundant observed-row count.
+        "max_json_chars": 14_500,
+        "max_nested_keys": 360,
         "max_top_level_keys": 52,
     },
     "dashboard_status_json": {
diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index f1a2760aec..89aaf55915 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -1,3 +1,4 @@
+import {projectTodoSummaryLanes, projectLegacyTodoWorkCounts} from "./todos/summary_lanes.ts";
 import {selectDelegationBinding, transitionDelegationObservation} from "./collaboration/delegation.ts";
 import {resolveConversationScope} from "./collaboration/conversation_scope.ts";
 import {previewTeamPlan, planTeamTransaction, teamTransactionIdentity} from "./work_items/team_plan.ts";
@@ -408,6 +409,8 @@ export function createEffectRuntimeHandlers(
     ["todo.field_update.plan", planTodoFieldUpdate],
     ["todo.public_update.plan", planPublicTodoUpdate],
     ["todo.standing_decision.project", evaluateStandingDecisionProjection],
+    ["todo.summary_lanes.project", projectTodoSummaryLanes],
+    ["todo.work_counts.project", projectLegacyTodoWorkCounts],
     ["todo.decision_scope.evaluate", evaluateDecisionScope],
     ["agent.capability_gate.evaluate", evaluateCapabilityGate],
     ["agent.capability_memory", agentCapabilityMemory],
diff --git a/loopx/control_plane/scheduler/external_evidence_observation.py b/loopx/control_plane/scheduler/external_evidence_observation.py
index e8be659e77..0510af08be 100644
--- a/loopx/control_plane/scheduler/external_evidence_observation.py
+++ b/loopx/control_plane/scheduler/external_evidence_observation.py
@@ -112,7 +112,8 @@ def scoped_monitor_watch_without_advancement(summary: dict[str, Any] | None) ->
         return False
     if not todo_summary_monitor_items(summary):
         return False
-    return todo_summary_open_task_counts(summary).get("advancement", 0) <= 0
+    counts = todo_summary_open_task_counts(summary)
+    return counts["complete"] is True and counts["advancement"] == 0
 
 
 def _monitor_item_matches_handle(
diff --git a/loopx/control_plane/testing/cli_output_differential.py b/loopx/control_plane/testing/cli_output_differential.py
index 415cc2775e..305f96409c 100644
--- a/loopx/control_plane/testing/cli_output_differential.py
+++ b/loopx/control_plane/testing/cli_output_differential.py
@@ -24,6 +24,7 @@
 PLANNING_HORIZON_SCHEMA_VERSION_V0 = "quota_planning_horizon_v0"
 GUIDED_TODO_DELTA_SCHEMA_VERSION_V0 = "loopx_guided_todo_delta_v0"
 PLANNING_INVENTORY_DETAIL_SCHEMA_VERSION_V0 = "todo_planning_inventory_detail_v0"
+TODO_WORK_COUNTS_SCHEMA_VERSION_V0 = "todo_work_counts_v0"
 
 Metric = Literal["chars", "utf8_bytes", "lines", "compact_payload_chars"]
 
@@ -152,6 +153,16 @@ class GrowthAllowance:
     "compact_payload_chars": 1_024,
 }
 
+# Source-complete Todo counts add one compact scope/completeness envelope to
+# the existing Todo, status, and quota projections. The allowance is bound to
+# the none-to-v0 schema transition; v0-to-v0 changes use the ordinary budgets.
+_TODO_WORK_COUNTS_V0_MIGRATION_GROWTH_ALLOWANCE: dict[Metric, int] = {
+    "chars": 320,
+    "utf8_bytes": 320,
+    "lines": 10,
+    "compact_payload_chars": 192,
+}
+
 # Explicit runtime-root command routing repeats one bounded command prefix per
 # executable action. The allowance covers the prefix and its JSON projection;
 # it is per newly observed route, not per row, so unrelated output growth still
@@ -412,6 +423,16 @@ def _planning_inventory_detail_schema_migration(
     return None
 
 
+def _todo_work_counts_schema_migration(
+    base: dict[str, Any], candidate: dict[str, Any]
+) -> str | None:
+    base_versions = tuple(base.get("todo_work_counts_schema_versions") or [])
+    candidate_versions = tuple(candidate.get("todo_work_counts_schema_versions") or [])
+    if base_versions == () and candidate_versions == (TODO_WORK_COUNTS_SCHEMA_VERSION_V0,):
+        return f"none -> {TODO_WORK_COUNTS_SCHEMA_VERSION_V0}"
+    return None
+
+
 @dataclass(frozen=True)
 class _SchemaMigrationState:
     signature_changed: bool
@@ -424,11 +445,14 @@ class _SchemaMigrationState:
     inventory_detail_schema_migration: str | None
     guided_todo_delta_schema_changed: bool
     guided_todo_delta_schema_migration: str | None
+    todo_work_counts_schema_changed: bool
+    todo_work_counts_schema_migration: str | None
     portfolio_growth_migration: bool
     horizon_growth_migration: bool
     agent_context_growth_migration: bool
     inventory_detail_growth_migration: bool
     guided_todo_delta_growth_migration: bool
+    todo_work_counts_growth_migration: bool
 
 
 def _schema_migration_state(
@@ -479,6 +503,14 @@ def _schema_migration_state(
         if guided_todo_delta_schema_changed
         else None
     )
+    todo_work_counts_schema_changed = tuple(
+        base.get("todo_work_counts_schema_versions") or []
+    ) != tuple(candidate.get("todo_work_counts_schema_versions") or [])
+    todo_work_counts_schema_migration = (
+        _todo_work_counts_schema_migration(base, candidate)
+        if todo_work_counts_schema_changed
+        else None
+    )
     return _SchemaMigrationState(
         signature_changed=signature_changed,
         signature_migration=signature_migration,
@@ -490,6 +522,8 @@ def _schema_migration_state(
         inventory_detail_schema_migration=inventory_detail_schema_migration,
         guided_todo_delta_schema_changed=guided_todo_delta_schema_changed,
         guided_todo_delta_schema_migration=guided_todo_delta_schema_migration,
+        todo_work_counts_schema_changed=todo_work_counts_schema_changed,
+        todo_work_counts_schema_migration=todo_work_counts_schema_migration,
         portfolio_growth_migration=bool(
             output_format == "json"
             and (
@@ -525,9 +559,34 @@ def _schema_migration_state(
         guided_todo_delta_growth_migration=bool(
             output_format == "json" and guided_todo_delta_schema_migration
         ),
+        todo_work_counts_growth_migration=bool(
+            output_format == "json" and todo_work_counts_schema_migration
+        ),
     )
 
 
+def _schema_migration_growth_allowance(
+    migration: _SchemaMigrationState,
+    metric: Metric,
+) -> int:
+    allowances: list[int] = []
+    if migration.portfolio_growth_migration:
+        allowances.append(_ACTION_PORTFOLIO_V0_MIGRATION_GROWTH_ALLOWANCE[metric])
+    if migration.horizon_growth_migration:
+        allowances.append(_PLANNING_HORIZON_V0_MIGRATION_GROWTH_ALLOWANCE[metric])
+    if migration.agent_context_growth_migration:
+        allowances.append(_AGENT_CONTEXT_V4_MIGRATION_GROWTH_ALLOWANCE[metric])
+    if migration.inventory_detail_growth_migration:
+        allowances.append(
+            _PLANNING_INVENTORY_DETAIL_V0_MIGRATION_GROWTH_ALLOWANCE[metric]
+        )
+    if migration.guided_todo_delta_growth_migration:
+        allowances.append(_GUIDED_TODO_DELTA_V0_MIGRATION_GROWTH_ALLOWANCE[metric])
+    if migration.todo_work_counts_growth_migration:
+        allowances.append(_TODO_WORK_COUNTS_V0_MIGRATION_GROWTH_ALLOWANCE[metric])
+    return max(allowances, default=0)
+
+
 def _compare_row(base: dict[str, Any], candidate: dict[str, Any]) -> dict[str, Any]:
     row_id = str(base["row_id"])
     failures: list[str] = []
@@ -591,6 +650,7 @@ def _compare_row(base: dict[str, Any], candidate: dict[str, Any]) -> dict[str, A
                 candidate,
                 metric,
             ),
+            _schema_migration_growth_allowance(migration, metric),
         )
         # Thin installed prompts contain bilingual lifecycle instructions. A
         # small character-level clarification can cost three bytes per CJK
@@ -625,30 +685,6 @@ def _compare_row(base: dict[str, Any], candidate: dict[str, Any]) -> dict[str, A
                     "compact_payload_chars": 512,
                 }[metric],
             )
-        if migration.portfolio_growth_migration:
-            allowance = max(
-                allowance,
-                _ACTION_PORTFOLIO_V0_MIGRATION_GROWTH_ALLOWANCE[metric],
-            )
-        if migration.horizon_growth_migration:
-            allowance = max(
-                allowance,
-                _PLANNING_HORIZON_V0_MIGRATION_GROWTH_ALLOWANCE[metric],
-            )
-        if migration.agent_context_growth_migration:
-            allowance = max(
-                allowance, _AGENT_CONTEXT_V4_MIGRATION_GROWTH_ALLOWANCE[metric]
-            )
-        if migration.inventory_detail_growth_migration:
-            allowance = max(
-                allowance,
-                _PLANNING_INVENTORY_DETAIL_V0_MIGRATION_GROWTH_ALLOWANCE[metric],
-            )
-        if migration.guided_todo_delta_growth_migration:
-            allowance = max(
-                allowance,
-                _GUIDED_TODO_DELTA_V0_MIGRATION_GROWTH_ALLOWANCE[metric],
-            )
         if runtime_root_route_allowances:
             allowance = max(
                 allowance,
@@ -729,6 +765,14 @@ def _compare_row(base: dict[str, Any], candidate: dict[str, Any]) -> dict[str, A
                 "guided todo delta schema migrated: "
                 f"{migration.guided_todo_delta_schema_migration}"
             )
+    if migration.todo_work_counts_schema_changed:
+        if migration.todo_work_counts_schema_migration is None:
+            failures.append("Todo work-count schema coverage changed")
+        else:
+            review_signals.append(
+                "Todo work-count schema migrated: "
+                f"{migration.todo_work_counts_schema_migration}"
+            )
 
     return {
         "row_id": row_id,
diff --git a/loopx/control_plane/testing/cli_output_semantics.py b/loopx/control_plane/testing/cli_output_semantics.py
index fa32f1f19f..603d0fc68f 100644
--- a/loopx/control_plane/testing/cli_output_semantics.py
+++ b/loopx/control_plane/testing/cli_output_semantics.py
@@ -179,6 +179,10 @@ def guided_todo_delta_schema_versions(value: Any) -> list[str]:
     return _schema_versions_for_key(value, "todo_delta")
 
 
+def todo_work_counts_schema_versions(value: Any) -> list[str]:
+    return _schema_versions_for_key(value, "work_counts")
+
+
 def markdown_headings(text: str) -> list[str]:
     return [line.strip() for line in text.splitlines() if _MARKDOWN_HEADING.match(line)]
 
diff --git a/loopx/control_plane/testing/quota_fixtures.py b/loopx/control_plane/testing/quota_fixtures.py
index 3f7fcd3bd7..dd937c25d6 100644
--- a/loopx/control_plane/testing/quota_fixtures.py
+++ b/loopx/control_plane/testing/quota_fixtures.py
@@ -93,6 +93,15 @@ def quota_todo_summary(
             "monitor_due_count": 0,
         }
     if claim_scope_agent_id:
+        from ..todos.quota_summary import summarize_project_asset_todos_for_quota
+
+        scoped = summarize_project_asset_todos_for_quota(
+            summary,
+            agent_identity={"agent_id": claim_scope_agent_id},
+        )
+        if scoped is None:
+            raise RuntimeError("agent-scoped Todo fixture projection is unavailable")
+        summary["work_counts"] = scoped["work_counts"]
         summary["claim_scope"] = {"agent_id": claim_scope_agent_id}
     return summary
 
diff --git a/loopx/control_plane/todos/list_projection.py b/loopx/control_plane/todos/list_projection.py
index d0337a51a3..f630dda4bc 100644
--- a/loopx/control_plane/todos/list_projection.py
+++ b/loopx/control_plane/todos/list_projection.py
@@ -41,6 +41,7 @@
     "active_next_action_executable_items": 3,
 }
 _RETAINED_DICTS = {
+    "work_counts",
     "monitor_writeback",
     "source_proof",
     "terminal_closure_proof",
@@ -208,7 +209,9 @@ def compact_thin_todo_summary(
                 omitted_nonempty_lane_count += bool(value)
             continue
         if isinstance(value, dict):
-            if key == "payload_compaction":
+            if key == "work_counts":
+                compact[key] = value
+            elif key == "payload_compaction":
                 source_view = _summary_source_view(value)
             else:
                 omitted_nonempty_dict_count += bool(value)
diff --git a/loopx/control_plane/todos/quota_selection.py b/loopx/control_plane/todos/quota_selection.py
index 0e2df5c202..421b00ead8 100644
--- a/loopx/control_plane/todos/quota_selection.py
+++ b/loopx/control_plane/todos/quota_selection.py
@@ -72,6 +72,7 @@ def active(key: str) -> list[dict[str, Any]]:
                 "user_gate_scope": filter_user_gate_blocks_agent,
                 "monitor_supported": todo_summary_monitor_writeback_supported(value),
                 "source_open_count": source_open_count,
+                "source_complete": (value.get("work_counts") or {}).get("complete", True),
                 "diagnostic_limit": 3, "backlog_limit": 8, "visibility_limit": 16,
             },
         })
diff --git a/loopx/control_plane/todos/quota_selection.ts b/loopx/control_plane/todos/quota_selection.ts
index c71e4a16bb..582d728bc5 100644
--- a/loopx/control_plane/todos/quota_selection.ts
+++ b/loopx/control_plane/todos/quota_selection.ts
@@ -1,3 +1,4 @@
+import {countTodoWork} from "./summary_lanes.ts";
 import type { JsonObject } from "../effect_program.ts";
 import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts";
 import { requireJsonObject, requireBoolean, requireInteger, requireStringArray,
@@ -158,7 +159,12 @@ export function projectQuotaSelection(value: unknown): JsonObject {
     policy: "user actions bound to another agent remain diagnostic-only and must not enter this agent's reminder channel",
     current_agent_user_action_open_count: actions.length, other_agent_bound_user_action_open_count: otherActions.length,
   } : null;
-  return {lanes: {
+  const displayed = userMode ? [...open, ...actions] : open;
+  const sourceComplete = request.source_open_count === source.length &&
+    (request.source_complete === undefined || requireBoolean(request.source_complete, "source_complete"));
+  const countOpen = !agent && !userMode && Number.isSafeInteger(request.source_open_count)
+    ? Math.max(Number(request.source_open_count), displayed.length) : displayed.length;
+  return {work_counts: countTodoWork(displayed, countOpen, sourceComplete, agent && !userMode ? agent : null), lanes: {
     all_open_items: payloads(source), blocking_open_items: payloads(blocking),
     user_action_open_items: payloads(actions), other_agent_bound_user_action_items: payloads(otherActions),
     user_action_agent_scope_filter: actionFilter, other_agent_scoped_items: payloads(otherGates),
diff --git a/loopx/control_plane/todos/quota_summary.py b/loopx/control_plane/todos/quota_summary.py
index 30a16d2e26..214da11486 100644
--- a/loopx/control_plane/todos/quota_summary.py
+++ b/loopx/control_plane/todos/quota_summary.py
@@ -391,6 +391,7 @@ def summarize_user_todos_for_quota(
         "source_section": value.get("source_section"),
         "total_count": value.get("total_count"),
         "open_count": lanes.open_count,
+        "work_counts": planning["work_counts"],
         "done_count": value.get("done_count"),
         "deferred_count": value.get("deferred_count"),
         "source_completeness": source_completeness,
@@ -668,7 +669,7 @@ def _compact_agent_lane_status_todo_summary(
                 }
             continue
         if isinstance(value, dict):
-            if key == "monitor_writeback":
+            if key in {"monitor_writeback", "work_counts"}:
                 compact[key] = _compact_quota_payload_nested_warning(value)
             continue
         compact[key] = value
@@ -813,6 +814,7 @@ def summarize_project_asset_todos_for_quota(
         "source_section": value.get("source_section") or "project_asset",
         "total_count": value.get("total", value.get("total_count")),
         "open_count": lanes.open_count,
+        "work_counts": planning["work_counts"],
         "done_count": value.get("done", value.get("done_count")),
         "first_open_items": lanes.display_open_items[:3],
         "first_executable_items": lanes.executable_items[:3],
diff --git a/loopx/control_plane/todos/summary_lanes.ts b/loopx/control_plane/todos/summary_lanes.ts
new file mode 100644
index 0000000000..ae2a045798
--- /dev/null
+++ b/loopx/control_plane/todos/summary_lanes.ts
@@ -0,0 +1,127 @@
+/** Read-only lane selection over one evaluated source, before display limits. */
+import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts";
+import type {JsonObject} from "../effect_program.ts";
+import {requireBoolean, requireJsonObject, requireStringLiteral} from "../runtime_decode.ts";
+import {authorityUnicodeCompare} from "../coordination/authority_store_codec.ts";
+
+export const TODO_SUMMARY_LANES = [
+  "open_items", "terminal_items", "deferred_items", "done_items", "projected_open_items",
+  "projected_deferred_items", "budgeted_items", "claimed_open_items", "unclaimed_open_items",
+  "executable_items", "blocker_items", "resume_blocked_items", "monitor_items",
+  "monitor_due_items", "monitor_schedule_gap_items", "claimed_advancement_items",
+  "claimed_monitor_items", "active_next_action_items", "active_next_action_executable_items",
+] as const;
+export type TodoSummaryLane = typeof TODO_SUMMARY_LANES[number];
+const TASK_CLASSES = ["advancement_task", "continuous_monitor", "user_gate", "user_action", "blocker"] as const;
+
+interface Row {
+  ordinal: number; status: "open" | "blocked" | "done" | "deferred";
+  taskClass: typeof TASK_CLASSES[number]; done: boolean; actionable: boolean;
+  claim: boolean; resumeBlocked: boolean; preferred: boolean; watchOnly: boolean;
+  dueAt: number | null; expiresAt: number | null; sort: readonly [number, number, string, string];
+}
+function finite(value: unknown, label: string): number {
+  if (typeof value !== "number" || !Number.isFinite(value)) throw new EffectRuntimeRequestError(`${label} must be finite`);
+  return value;
+}
+function optionalTime(value: unknown, label: string): number | null {
+  return value === null ? null : finite(value, label);
+}
+function decodeRow(value: unknown, ordinal: number): Row {
+  const row = requireJsonObject(value, `rows[${ordinal}]`);
+  const status = requireStringLiteral(row.status, ["open", "blocked", "done", "deferred"], "status");
+  const done = requireBoolean(row.done, "done");
+  if (done !== (status === "done" || status === "deferred")) throw new EffectRuntimeRequestError("Todo status/done disagree");
+  const hasResume = requireBoolean(row.has_resume, "has_resume");
+  const ready = row.resume_ready === true;
+  if (hasResume && !requireBoolean(row.resume_evaluated, "resume_evaluated")) {
+    throw new EffectRuntimeRequestError("Todo display requires a matching full-source resume evaluation");
+  }
+  const sort = row.sort;
+  if (!Array.isArray(sort) || sort.length !== 4 || typeof sort[2] !== "string" || typeof sort[3] !== "string") {
+    throw new EffectRuntimeRequestError("Todo presentation sort coordinate is invalid");
+  }
+  return {ordinal, status, done, taskClass: requireStringLiteral(row.task_class, TASK_CLASSES, "task_class"),
+    actionable: status === "open" && (!hasResume || ready) && !requireBoolean(row.acceptance_blocked, "acceptance_blocked"),
+    claim: requireBoolean(row.claimed, "claimed"), resumeBlocked: hasResume && row.resume_ready === false,
+    preferred: requireBoolean(row.preferred, "preferred"), watchOnly: requireBoolean(row.watch_only, "watch_only"),
+    dueAt: optionalTime(row.due_at, "due_at"), expiresAt: optionalTime(row.expires_at, "expires_at"),
+    sort: [finite(sort[0], "priority"), finite(sort[1], "index"), sort[2], sort[3]]};
+}
+function compare(left: Row, right: Row): number {
+  return left.sort[0] - right.sort[0] || left.sort[1] - right.sort[1] ||
+    authorityUnicodeCompare(left.sort[2], right.sort[2]) || authorityUnicodeCompare(left.sort[3], right.sort[3]);
+}
+export interface WorkCountRow {readonly actionable: boolean; readonly taskClass: string}
+
+/** Counts describe the observed source, never the size of a display lane.
+ * Incomplete legacy sources yield lower bounds; unseen work is not classified. */
+export function countTodoWork(rows: readonly WorkCountRow[], sourceOpenCount: number,
+  complete: boolean, agentId: string | null = null): JsonObject {
+  if (!Number.isSafeInteger(sourceOpenCount) || sourceOpenCount < rows.length) {
+    throw new EffectRuntimeRequestError("source open count cannot be smaller than observed work");
+  }
+  return {schema_version: "todo_work_counts_v0", open: sourceOpenCount,
+    advancement: rows.filter(row => row.actionable && row.taskClass === "advancement_task").length,
+    monitor: rows.filter(row => row.actionable && row.taskClass === "continuous_monitor").length,
+    hidden: sourceOpenCount - rows.length,
+    complete: complete && sourceOpenCount === rows.length, agent_id: agentId};
+}
+
+export function projectTodoSummaryLanes(value: unknown): JsonObject {
+  const request = requireJsonObject(value, "Todo summary lane request");
+  if (request.schema_version !== "todo_summary_lanes_request_v0" || !Array.isArray(request.rows)) {
+    throw new EffectRuntimeRequestError("Todo summary lane request schema mismatch");
+  }
+  const rows = request.rows.map(decodeRow), now = finite(request.observed_at, "observed_at");
+  const open = rows.filter(row => !row.done), terminal = rows.filter(row => row.done);
+  const deferred = terminal.filter(row => row.status === "deferred"), done = terminal.filter(row => row.status !== "deferred");
+  const ordered = [...open].sort(compare), orderedDeferred = [...deferred].sort(compare);
+  const claimed = ordered.filter(row => row.claim);
+  const executable = ordered.filter(row => row.actionable && row.taskClass === "advancement_task");
+  const monitors = ordered.filter(row => row.actionable && row.taskClass === "continuous_monitor");
+  const activeMonitor = (row: Row) => row.expiresAt === null || row.expiresAt > now;
+  const due = monitors.filter(row => activeMonitor(row) && row.dueAt !== null && row.dueAt <= now);
+  const missing = monitors.filter(row => activeMonitor(row) && !row.watchOnly && row.dueAt === null);
+  const selected = {
+    open_items: open, terminal_items: terminal, deferred_items: deferred, done_items: done,
+    projected_open_items: ordered, projected_deferred_items: orderedDeferred,
+    budgeted_items: [...ordered, ...orderedDeferred, ...done], claimed_open_items: claimed,
+    unclaimed_open_items: ordered.filter(row => !row.claim), executable_items: executable,
+    blocker_items: ordered.filter(row => row.status === "blocked" && row.taskClass === "blocker"),
+    resume_blocked_items: ordered.filter(row => row.resumeBlocked), monitor_items: monitors,
+    monitor_due_items: due, monitor_schedule_gap_items: missing,
+    claimed_advancement_items: executable.filter(row => row.claim), claimed_monitor_items: monitors.filter(row => row.claim),
+    active_next_action_items: ordered.filter(row => row.preferred),
+    active_next_action_executable_items: executable.filter(row => row.preferred),
+  } satisfies Record<TodoSummaryLane, readonly Row[]>;
+  const lanes = Object.fromEntries(TODO_SUMMARY_LANES.map(key => [key, selected[key].map(row => row.ordinal)]));
+  return {schema_version: "todo_summary_lanes_v0", lanes,
+    work_counts: countTodoWork(open, open.length, true)};
+}
+
+/** Compatibility summaries may contain only display fragments. Preserve their
+ * declared open total while classifying only unique observed rows. */
+export function projectLegacyTodoWorkCounts(value: unknown): JsonObject {
+  const request = requireJsonObject(value, "legacy Todo work counts");
+  if (request.schema_version !== "todo_work_counts_request_v0" || !Array.isArray(request.rows)) {
+    throw new EffectRuntimeRequestError("Todo work count request schema mismatch");
+  }
+  const observed = new Map<string, WorkCountRow>();
+  let consistent = true;
+  for (const value of request.rows) {
+    const row = requireJsonObject(value, "work count row");
+    if (typeof row.identity !== "string" || !row.identity) throw new EffectRuntimeRequestError("work count identity is required");
+    const current = {
+      actionable: requireBoolean(row.actionable, "actionable"),
+      taskClass: requireStringLiteral(row.task_class, TASK_CLASSES, "task_class"),
+    };
+    const previous = observed.get(row.identity);
+    if (previous && (previous.actionable !== current.actionable || previous.taskClass !== current.taskClass)) consistent = false;
+    if (!previous) observed.set(row.identity, current);
+  }
+  const declared = request.source_open_count;
+  const known = typeof declared === "number" && Number.isSafeInteger(declared) && declared >= observed.size;
+  return countTodoWork([...observed.values()], known ? declared : observed.size,
+    known && consistent && declared === observed.size, typeof request.agent_id === "string" ? request.agent_id : null);
+}
diff --git a/loopx/control_plane/todos/todo_semantics.py b/loopx/control_plane/todos/todo_semantics.py
index 41aa1748e3..f99519009b 100644
--- a/loopx/control_plane/todos/todo_semantics.py
+++ b/loopx/control_plane/todos/todo_semantics.py
@@ -768,80 +768,41 @@ def todo_summary_open_count(summary: dict[str, Any] | None) -> int:
         return 0
 
 
-def todo_summary_open_task_counts(summary: dict[str, Any] | None) -> dict[str, int]:
-    open_count = todo_summary_open_count(summary)
-    classified_items: list[dict[str, Any]] = []
-    seen: set[tuple[Any, str]] = set()
-    executable_backlog_items: list[dict[str, Any]] | None = None
-    monitor_open_items: list[dict[str, Any]] | None = None
-    if isinstance(summary, dict):
-        raw_executable_backlog = summary.get("executable_backlog_items")
-        if isinstance(raw_executable_backlog, list):
-            executable_backlog_items = [
-                item
-                for item in raw_executable_backlog
-                if isinstance(item, dict)
-                if todo_item_is_actionable_open(item)
-                if todo_item_task_class(item) == TODO_TASK_CLASS_ADVANCEMENT
-            ]
-        raw_monitor_open = summary.get("monitor_open_items")
-        if isinstance(raw_monitor_open, list):
-            monitor_open_items = [
-                item
-                for item in raw_monitor_open
-                if isinstance(item, dict)
-                if todo_item_is_actionable_open(item)
-                if todo_item_task_class(item) == TODO_TASK_CLASS_MONITOR
-            ]
-        for key in (
-            "first_executable_items",
-            "first_open_items",
-            "monitor_open_items",
-        ):
-            source_items = summary.get(key)
-            if not isinstance(source_items, list):
-                continue
-            for item in source_items:
-                if not isinstance(item, dict):
+def todo_summary_open_task_counts(summary: dict[str, Any] | None) -> dict[str, Any]:
+    """Consume pre-limit counts; old display-only summaries provide lower bounds."""
+    from ..effect_runtime import effect_runtime_result
+
+    summary = summary if isinstance(summary, dict) else {}
+    counts = summary.get("work_counts")
+    if counts is None:
+        rows = []
+        for key in ("items", "executable_backlog_items", "first_executable_items", "first_open_items", "monitor_open_items"):
+            for item in summary.get(key) or []:
+                if not isinstance(item, dict) or item.get("done") is True:
                     continue
                 text = str(item.get("text") or "").strip()
                 if not text:
                     continue
-                identity = (item.get("index"), text)
-                if identity in seen:
-                    continue
-                seen.add(identity)
-                classified_items.append(item)
-    if executable_backlog_items is not None:
-        advancement_count = len(executable_backlog_items)
-    else:
-        visible_open = min(open_count, len(classified_items))
-        advancement_visible_count = sum(
-            1
-            for item in classified_items[:visible_open]
-            if todo_item_is_actionable_open(item)
-            and todo_item_task_class(item) == TODO_TASK_CLASS_ADVANCEMENT
-        )
-        hidden_count = max(0, open_count - visible_open)
-        advancement_count = advancement_visible_count + hidden_count
-    if monitor_open_items is not None:
-        monitor_visible_count = len(monitor_open_items)
-    else:
-        visible_open = min(open_count, len(classified_items))
-        monitor_visible_count = sum(
-            1
-            for item in classified_items[:visible_open]
-            if todo_item_is_actionable_open(item)
-            and todo_item_task_class(item) == TODO_TASK_CLASS_MONITOR
-        )
-    hidden_count = max(0, open_count - len(classified_items))
-    return {
-        "open": open_count,
-        "advancement": advancement_count,
-        "monitor": monitor_visible_count,
+                rows.append({"identity": str(item.get("todo_id") or (str(item.get("index")) + ":" + text)),
+                    "actionable": todo_item_is_actionable_open(item), "task_class": todo_item_task_class(item)})
+        counts = effect_runtime_result("todo.work_counts.project", {
+            "schema_version": "todo_work_counts_request_v0", "rows": rows,
+            "source_open_count": summary.get("open_count"),
+            "agent_id": todo_summary_claim_scope_agent_id(summary),
+        })
+    if (not isinstance(counts, dict) or counts.get("schema_version") != "todo_work_counts_v0"
+        or not isinstance(counts.get("complete"), bool)
+        or any(type(counts.get(key)) is not int or counts[key] < 0
+               for key in ("open", "advancement", "monitor", "hidden"))
+        or counts.get("agent_id") != todo_summary_claim_scope_agent_id(summary)):
+        raise ValueError("invalid or differently scoped Todo work counts")
+    if (counts["hidden"] > counts["open"]
+        or counts["advancement"] + counts["monitor"] > counts["open"] - counts["hidden"]
+        or (counts["complete"] and counts["hidden"] != 0)):
+        raise ValueError("inconsistent Todo work count envelope")
+    return {key: counts[key] for key in ("open", "advancement", "monitor", "hidden", "complete")} | {
         "monitor_due": todo_summary_monitor_due_count(summary),
         "monitor_schedule_gap": todo_summary_monitor_schedule_gap_count(summary),
-        "hidden": hidden_count,
     }
 
 
@@ -855,6 +816,9 @@ def todo_summary_has_only_future_scoped_monitor_work(
         return False
     if not todo_summary_monitor_items(summary):
         return False
+    counts = todo_summary_open_task_counts(summary)
+    if counts["complete"] is not True or counts["advancement"] > 0:
+        return False
     if todo_summary_monitor_due_count(summary) > 0:
         return False
     if todo_summary_monitor_schedule_gap_count(summary) > 0:
diff --git a/loopx/control_plane/todos/todo_summary.py b/loopx/control_plane/todos/todo_summary.py
index a6deeb2a14..168a7c3e20 100644
--- a/loopx/control_plane/todos/todo_summary.py
+++ b/loopx/control_plane/todos/todo_summary.py
@@ -11,8 +11,6 @@
     TODO_STATUS_DONE,
     TODO_STATUS_OPEN,
     TODO_TASK_CLASS_ADVANCEMENT,
-    TODO_TASK_CLASS_BLOCKER,
-    TODO_TASK_CLASS_MONITOR,
     TODO_TASK_CLASS_USER_ACTION,
     build_todo_id,
     normalize_required_capabilities,
@@ -68,7 +66,7 @@
     TODO_SUCCESSION_WARNING_SCHEMA_VERSION,
 )
 from .resume_condition import evaluate_todo_resume_conditions
-from ..runtime.time import now_utc_iso
+from ..runtime.time import now_utc, now_utc_iso
 from ..work_items.project_asset import build_project_asset_todo_summary
 from .user_gate import open_user_gate_todo_items
 from ..coordination.coordination_state_contract import (
@@ -962,105 +960,46 @@ def _structured_resume_source_items(
     ]
 
 
-def _todo_group_lanes(
-    items: list[dict[str, Any]],
-    *,
-    preferred_todo_ids: set[str] | None,
-) -> _TodoGroupLanes:
-    open_items = [item for item in items if not item.get("done")]
-    terminal_items = [item for item in items if item.get("done")]
-    deferred_items = [item for item in terminal_items if todo_item_is_deferred(item)]
-    done_items = [item for item in terminal_items if not todo_item_is_deferred(item)]
-    projected_open_items = sorted(open_items, key=projection_todo_presentation_sort_key)
-    projected_deferred_items = sorted(
-        deferred_items,
-        key=projection_todo_presentation_sort_key,
-    )
-    budgeted_items = [
-        *projected_open_items,
-        *projected_deferred_items,
-        *done_items,
-    ]
-    claimed_open_items = [item for item in projected_open_items if item.get("claimed_by")]
-    unclaimed_open_items = [item for item in projected_open_items if not item.get("claimed_by")]
-    executable_items = [
-        item
-        for item in projected_open_items
-        if todo_item_is_actionable_open(item)
-        if todo_item_task_class(item) == TODO_TASK_CLASS_ADVANCEMENT
-    ]
-    blocker_items = [
-        item
-        for item in projected_open_items
-        if normalize_todo_status(item.get("status")) == "blocked"
-        if todo_item_task_class(item) == TODO_TASK_CLASS_BLOCKER
-    ]
-    resume_blocked_items = [
-        item
-        for item in projected_open_items
-        if normalize_todo_resume_when(item.get("resume_when"))
-        if item.get("resume_ready") is False
-    ]
-    monitor_items = [
-        item
-        for item in projected_open_items
-        if todo_item_is_actionable_open(item)
-        if todo_item_task_class(item) == TODO_TASK_CLASS_MONITOR
-    ]
-    monitor_due_items = [item for item in monitor_items if todo_item_is_due_monitor(item)]
-    monitor_schedule_gap_items = [
-        item
-        for item in monitor_items
-        if todo_item_missing_monitor_schedule(item)
-    ]
-    claimed_advancement_items = [
-        item
-        for item in claimed_open_items
-        if todo_item_is_actionable_open(item)
-        if todo_item_task_class(item) == TODO_TASK_CLASS_ADVANCEMENT
-    ]
-    claimed_monitor_items = [
-        item
-        for item in claimed_open_items
-        if todo_item_is_actionable_open(item)
-        if todo_item_task_class(item) == TODO_TASK_CLASS_MONITOR
-    ]
-    preferred_ids = {
-        todo_id
-        for todo_id in (preferred_todo_ids or set())
-        if normalize_todo_id(todo_id)
-    }
-    active_next_action_items = [
-        item
-        for item in projected_open_items
-        if normalize_todo_id(item.get("todo_id")) in preferred_ids
-    ]
-    active_next_action_executable_items = [
-        item
-        for item in executable_items
-        if normalize_todo_id(item.get("todo_id")) in preferred_ids
-    ]
-    return _TodoGroupLanes(
-        open_items=open_items,
-        terminal_items=terminal_items,
-        deferred_items=deferred_items,
-        done_items=done_items,
-        projected_open_items=projected_open_items,
-        projected_deferred_items=projected_deferred_items,
-        budgeted_items=budgeted_items,
-        claimed_open_items=claimed_open_items,
-        unclaimed_open_items=unclaimed_open_items,
-        executable_items=executable_items,
-        blocker_items=blocker_items,
-        resume_blocked_items=resume_blocked_items,
-        monitor_items=monitor_items,
-        monitor_due_items=monitor_due_items,
-        monitor_schedule_gap_items=monitor_schedule_gap_items,
-        claimed_advancement_items=claimed_advancement_items,
-        claimed_monitor_items=claimed_monitor_items,
-        active_next_action_items=active_next_action_items,
-        active_next_action_executable_items=active_next_action_executable_items,
-    )
+def _project_summary_lanes(items: list[dict[str, Any]], preferred_todo_ids: set[str] | None) -> dict[str, Any]:
+    """Adapt legacy facts and read batch ordinals from the typed lane owner."""
+    from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result
+
+    rows = []
+    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"))
+        due = projection_todo_item_next_due_at(item)
+        expires = projection_todo_item_expires_at(item)
+        guard = item.get("goal_acceptance_guard")
+        rows.append({"status": item.get("status") or ("done" if item.get("done") else "open"),
+            "done": bool(item.get("done")), "task_class": projection_todo_item_task_class(item),
+            "has_resume": bool(resume), "resume_ready": item.get("resume_ready"),
+            "resume_evaluated": evaluated, "acceptance_blocked": isinstance(guard, dict) and guard.get("allowed") is False,
+            "claimed": bool(item.get("claimed_by")), "preferred": item.get("todo_id") in (preferred_todo_ids or 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))})
+    try:
+        result = effect_runtime_result("todo.summary_lanes.project", {
+            "schema_version": "todo_summary_lanes_request_v0", "rows": rows, "observed_at": now_utc().timestamp(),
+        })
+    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")
+    lanes = result["lanes"]
+    if not isinstance(lanes, dict) or any(
+        not isinstance(indices, list) or any(type(index) is not int or not 0 <= index < len(items) for index in indices)
+        for indices in lanes.values()
+    ):
+        raise ValueError("invalid Todo summary source ordinal")
+    return {"lanes": {key: [items[index] for index in indices] for key, indices in lanes.items()},
+            "work_counts": result["work_counts"]}
 
 
 def compact_todo_group(
@@ -1103,19 +1042,6 @@ def compact_todo_group(
     )
 
 
-def _require_full_source_resume_evaluations(items: list[dict[str, Any]]) -> None:
-    for item in items:
-        resume_when = normalize_todo_resume_when(item.get("resume_when"))
-        if not resume_when:
-            continue
-        condition = item.get("resume_condition")
-        if (not isinstance(condition, dict)
-            or condition.get("schema_version") != "todo_resume_condition_v0"
-            or condition.get("resume_when") != resume_when
-            or not isinstance(condition.get("satisfied"), bool)
-            or item.get("resume_ready") is not condition.get("satisfied")):
-            raise ValueError("Todo display requires a matching full-source resume evaluation")
-
 
 def compact_evaluated_todo_group(
     items: list[dict[str, Any]],
@@ -1136,8 +1062,8 @@ def compact_evaluated_todo_group(
     """
     if not items and not include_empty_source:
         return None
-    _require_full_source_resume_evaluations(items)
-    lanes = _todo_group_lanes(items, preferred_todo_ids=preferred_todo_ids)
+    projected = _project_summary_lanes(items, preferred_todo_ids)
+    lanes = _TodoGroupLanes(**projected["lanes"])
     source_valid = role in {"user", "agent"} and bool(str(source_section or "").strip())
     no_followup_items = [
         item
@@ -1193,6 +1119,7 @@ def compact_evaluated_todo_group(
         "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),
diff --git a/loopx/control_plane/work_items/work_lane.py b/loopx/control_plane/work_items/work_lane.py
index 3f64e88fd0..ab64eb244d 100644
--- a/loopx/control_plane/work_items/work_lane.py
+++ b/loopx/control_plane/work_items/work_lane.py
@@ -540,6 +540,7 @@ def build_work_lane_contract(
     has_monitor_todos = monitor_count > 0
     monitor_only_schedule = (
         has_agent_todos and has_monitor_todos and not has_advancement_todos
+        and todo_counts.get("complete", True) is True
     )
     non_runnable_non_monitor_count = max(0, open_count - monitor_count)
     first_due_monitor = due_monitor_items[0] if due_monitor_items else None
diff --git a/loopx/todos.py b/loopx/todos.py
index aa806f8eb8..649f850e6b 100644
--- a/loopx/todos.py
+++ b/loopx/todos.py
@@ -216,6 +216,8 @@ def list_goal_todos(
             fields=canonical_todo_summary_fields(
                 canonical_read["todos"],
                 rollout_events=rollout_events,
+                goal_acceptance_contract=canonical_read.get("goal_acceptance_contract"),
+                goal_acceptance_work_guards=canonical_read.get("goal_acceptance_work_guards"),
             ),
             source="file_authority",
             projection_fields={},
diff --git a/tests/control_plane/test_cli_output_differential.py b/tests/control_plane/test_cli_output_differential.py
index 2a3a72b372..b584c39d4b 100644
--- a/tests/control_plane/test_cli_output_differential.py
+++ b/tests/control_plane/test_cli_output_differential.py
@@ -19,6 +19,7 @@
     planning_horizon_schema_versions,
     planning_inventory_detail_schema_versions,
     runtime_root_command_route_count,
+    todo_work_counts_schema_versions,
 )
 
 
@@ -44,6 +45,7 @@ def _row(**overrides: object) -> dict[str, object]:
         "planning_horizon_schema_versions": [],
         "guided_todo_delta_schema_versions": [],
         "planning_inventory_detail_schema_versions": [],
+        "todo_work_counts_schema_versions": [],
         "runtime_root_command_route_count": 0,
     }
     row.update(overrides)
@@ -333,6 +335,44 @@ def test_growth_above_policy_allowance_fails() -> None:
     assert "chars grew" in result["rows"][0]["failures"][0]
 
 
+def test_todo_work_count_schema_migration_has_one_time_bounded_budget() -> None:
+    payload = {"agent_todos": {"work_counts": {"schema_version": "todo_work_counts_v0"}}}
+    assert todo_work_counts_schema_versions(payload) == ["todo_work_counts_v0"]
+    candidate = _row(
+        chars=40_300,
+        utf8_bytes=40_300,
+        lines=1_010,
+        compact_payload_chars=20_180,
+        todo_work_counts_schema_versions=["todo_work_counts_v0"],
+    )
+
+    result = compare_cli_output_receipts(_receipt(_row()), _receipt(candidate))
+
+    assert result["ok"] is True
+    assert result["review_required"] is True
+    assert result["rows"][0]["allowances"] == {
+        "chars": 320,
+        "utf8_bytes": 320,
+        "lines": 10,
+        "compact_payload_chars": 192,
+    }
+    assert result["rows"][0]["review_signals"] == [
+        "Todo work-count schema migrated: none -> todo_work_counts_v0"
+    ]
+
+
+def test_todo_work_count_schema_migration_still_fails_above_bounded_growth() -> None:
+    candidate = _row(
+        chars=40_321,
+        todo_work_counts_schema_versions=["todo_work_counts_v0"],
+    )
+
+    result = compare_cli_output_receipts(_receipt(_row()), _receipt(candidate))
+
+    assert result["ok"] is False
+    assert "chars grew by 321; allowance is 320" in result["rows"][0]["failures"]
+
+
 def test_shrink_with_semantic_shape_retained_passes() -> None:
     base = _receipt(_row())
     candidate = _receipt(
diff --git a/tests/control_plane/test_goal_acceptance_runtime.py b/tests/control_plane/test_goal_acceptance_runtime.py
index 2143313449..c7a7c32d62 100644
--- a/tests/control_plane/test_goal_acceptance_runtime.py
+++ b/tests/control_plane/test_goal_acceptance_runtime.py
@@ -97,3 +97,18 @@ def test_absent_acceptance_does_not_change_summary_shape_or_executability(tmp_pa
     assert "goal_acceptance_contract" not in summary
     assert summary["first_executable_items"][0]["todo_id"] == "todo_work"
     assert "goal_acceptance_guard" not in json.dumps(summary)
+
+
+def test_public_todo_list_counts_and_selection_share_acceptance_guard(tmp_path):
+    from loopx.todos import list_goal_todos
+    from loopx.control_plane.todos.todo_semantics import todo_summary_open_task_counts
+
+    goal, runtime, state, registry = seed(tmp_path, enabled=True, monitor=True)
+    state.unlink()
+    result = list_goal_todos(registry_path=registry, runtime_root_arg=str(runtime), goal_id=goal["id"])
+    summary = result["agent_todos"]
+    assert summary["first_executable_items"] == []
+    assert todo_summary_open_task_counts(summary)["advancement"] == 0
+    assert todo_summary_open_task_counts(summary)["monitor"] == 1
+    assert "validation_argv" not in json.dumps(result)
+    assert not state.exists()
diff --git a/tests/control_plane/test_todo_work_counts.py b/tests/control_plane/test_todo_work_counts.py
new file mode 100644
index 0000000000..8f9bb3714f
--- /dev/null
+++ b/tests/control_plane/test_todo_work_counts.py
@@ -0,0 +1,106 @@
+"""Semantic counts survive source selection and every presentation budget."""
+from __future__ import annotations
+
+from loopx.control_plane.todos.todo_summary import compact_todo_group
+from loopx.control_plane.todos.todo_semantics import todo_summary_open_task_counts
+from loopx.control_plane.todos.quota_summary import (
+    summarize_user_todos_for_quota, compact_quota_todo_summary_for_payload,
+)
+
+
+def work(index, **fields):
+    return {"todo_id": f"todo_work_{index:03}", "text": f"[P1] Work {index}",
+            "status": "open", "task_class": "advancement_task", "role": "agent",
+            "index": index + 1, "source_section": "Agent Todo", **fields}
+
+
+def test_complete_counts_do_not_equal_backlog_display_length():
+    summary = compact_todo_group([work(i) for i in range(21)], source_section="Agent Todo", role="agent")
+    assert len(summary["executable_backlog_items"]) == 8
+    assert todo_summary_open_task_counts(summary)["advancement"] == 21
+
+
+def test_scoped_counts_survive_quota_payload_compaction():
+    records = [work(i, claimed_by="agent-a" if i < 21 else "agent-b") for i in range(35)]
+    summary = compact_todo_group(records, source_section="Agent Todo", role="agent", item_limit=None)
+    scoped = summarize_user_todos_for_quota(summary, agent_identity={"agent_id": "agent-a"})
+    compact = compact_quota_todo_summary_for_payload(scoped)
+    assert len(compact["executable_backlog_items"]) == 2
+    assert todo_summary_open_task_counts(compact)["advancement"] == 21
+
+
+def test_incomplete_legacy_summary_cannot_invent_hidden_advancement():
+    summary = {"open_count": 10, "first_open_items": [work(0, task_class="blocker", status="blocked")]}
+    counts = todo_summary_open_task_counts(summary)
+    assert counts["advancement"] == 0
+    assert counts["complete"] is False
+
+
+def test_list_and_status_compactors_preserve_semantic_counts():
+    from loopx.control_plane.todos.list_projection import (
+        compact_agent_lane_todo_summary, compact_explicit_limit_todo_summary, compact_thin_todo_summary,
+    )
+    from loopx.control_plane.todos.quota_summary import _compact_agent_lane_status_todo_summary
+
+    summary = compact_todo_group([work(i) for i in range(21)], source_section="Agent Todo", role="agent")
+    projections = [compact_agent_lane_todo_summary(summary, role="agent"),
+        compact_explicit_limit_todo_summary(summary, role="agent", item_limit=1),
+        compact_thin_todo_summary(summary, role="agent", items_matched=21, items_returned=1, item_limit_per_role=1),
+        _compact_agent_lane_status_todo_summary(summary, role="agent")]
+    for projected in projections:
+        assert todo_summary_open_task_counts(projected)["advancement"] == 21
+        assert projected["work_counts"]["complete"] is True
+
+
+def test_incomplete_scoped_snapshot_does_not_certify_monitor_only():
+    from loopx.control_plane.scheduler.external_evidence_observation import scoped_monitor_watch_without_advancement
+    from loopx.control_plane.todos.todo_semantics import todo_summary_has_only_future_scoped_monitor_work
+
+    monitor = work(0, task_class="continuous_monitor", claimed_by="agent-a", next_due_at="2099-01-01T00:00:00Z")
+    scoped = summarize_user_todos_for_quota({"schema_version": "todo_summary_v0", "open_count": 12,
+        "items": [monitor]}, agent_identity={"agent_id": "agent-a"})
+    assert scoped["work_counts"]["complete"] is False
+    assert scoped_monitor_watch_without_advancement(scoped) is False
+    assert todo_summary_has_only_future_scoped_monitor_work(scoped) is False
+    repeated = summarize_user_todos_for_quota(scoped, agent_identity={"agent_id": "agent-a"})
+    assert repeated["work_counts"]["complete"] is False
+    assert todo_summary_has_only_future_scoped_monitor_work(repeated) is False
+    complete = summarize_user_todos_for_quota({"schema_version": "todo_summary_v0", "open_count": 1,
+        "items": [monitor]}, agent_identity={"agent_id": "agent-a"})
+    assert todo_summary_has_only_future_scoped_monitor_work(complete) is True
+
+
+def test_wrong_scope_count_envelope_is_not_reused():
+    import pytest
+    summary = compact_todo_group([work(0)], source_section="Agent Todo", role="agent")
+    summary["claim_scope"] = {"schema_version": "agent_claim_scope_v0", "agent_id": "agent-b"}
+    with pytest.raises(ValueError, match="differently scoped"):
+        todo_summary_open_task_counts(summary)
+
+
+def test_real_cli_count_contract_after_promoted_display_loss(tmp_path, monkeypatch):
+    import json
+    from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime
+    from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
+    from loopx.control_plane.testing.canary_harness import run_json_cli
+
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    for provider in ("file", "sqlite"):
+        root = tmp_path / provider
+        root.mkdir()
+        state = root / "state.md"
+        state.write_text("# Independent narrative\n\n## Agent Todo\n- [ ] Stale display\n")
+        runtime = root / "runtime"
+        registry = root / "registry.json"
+        registry.write_text(json.dumps({"common_runtime_root": str(runtime), "goals": [
+            {"id": "goal-count", "repo": str(root), "state_file": str(state)}]}))
+        records = [work(i, schema_version="todo_item_v0", archive_state="active", done=False) for i in range(29)]
+        projection = build_todo_runtime_shadow_projection(goal_id="goal-count", todos=records)
+        initialize_canonical_authority(runtime, "goal-count", projection, state_path=state, provider=provider)
+        state.unlink()
+        listed = run_json_cli("todo", "list", "--goal-id", "goal-count", "--role", "agent", "--limit", "1", "--thin",
+            registry_path=registry, runtime_root=runtime)
+        assert len(listed["todos"]) == 1
+        assert listed["agent_todos"]["work_counts"]["advancement"] == 29
+        assert listed["agent_todos"]["work_counts"]["complete"] is True
+        assert not state.exists()
diff --git a/tests/control_plane_ts/todo_summary_lanes.test.ts b/tests/control_plane_ts/todo_summary_lanes.test.ts
new file mode 100644
index 0000000000..29feaf084f
--- /dev/null
+++ b/tests/control_plane_ts/todo_summary_lanes.test.ts
@@ -0,0 +1,81 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import {projectTodoSummaryLanes, projectLegacyTodoWorkCounts, countTodoWork} from "../../loopx/control_plane/todos/summary_lanes.ts";
+import type {JsonObject} from "../../loopx/control_plane/effect_program.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, preferred: false, watch_only: false,
+  due_at: null, expires_at: null, sort: [1, 1, "", ""], ...fields});
+const project = (rows: JsonObject[]) => projectTodoSummaryLanes({schema_version: "todo_summary_lanes_request_v0", rows, observed_at: 100});
+
+test("lane algebra distinguishes open, blocked, deferred, completed and acceptance denial", () => {
+  const rows = [row(), row({status: "blocked", task_class: "blocker"}), row({done: true, status: "deferred"}),
+    row({done: true, status: "done"}), row({acceptance_blocked: true}),
+    row({has_resume: true, resume_ready: false, resume_evaluated: true}),
+    row({has_resume: true, resume_ready: true, resume_evaluated: true, claimed: true, preferred: true})];
+  const before = structuredClone(rows), result = project(rows), lanes = result.lanes as JsonObject;
+  assert.deepEqual(lanes.open_items, [0, 1, 4, 5, 6]);
+  assert.deepEqual(lanes.done_items, [3]);
+  assert.deepEqual(lanes.deferred_items, [2]);
+  assert.deepEqual(lanes.executable_items, [0, 6]);
+  assert.deepEqual(lanes.blocker_items, [1]);
+  assert.deepEqual(lanes.resume_blocked_items, [5]);
+  assert.deepEqual(lanes.active_next_action_executable_items, [6]);
+  assert.deepEqual(lanes.budgeted_items, [0, 1, 4, 5, 6, 2, 3]);
+  assert.equal((result.work_counts as JsonObject).advancement, 2);
+  assert.deepEqual(rows, before);
+});
+
+test("one observation time fences due, expiry, missing schedule and watch-only monitors", () => {
+  const monitor = (fields: JsonObject) => row({task_class: "continuous_monitor", ...fields});
+  const lanes = project([monitor({due_at: 100}), monitor({due_at: 101}),
+    monitor({due_at: 10, expires_at: 100}), monitor({}), monitor({watch_only: true}),
+    monitor({due_at: 90, acceptance_blocked: true})]).lanes as JsonObject;
+  assert.deepEqual(lanes.monitor_due_items, [0]);
+  assert.deepEqual(lanes.monitor_schedule_gap_items, [3]);
+  assert.deepEqual(lanes.monitor_items, [0, 1, 2, 3, 4]);
+});
+
+test("display ordering preserves stable legacy ties and Python Unicode ordering", () => {
+  const lanes = project([row({sort: [1, 2, "", ""]}), row({sort: [0, 9, "", ""]}),
+    row({sort: [1, 2, "", ""]}), row({sort: [1, 999999, "", "\u{10000}"]}),
+    row({sort: [1, 999999, "", "\ue000"]})]).lanes as JsonObject;
+  assert.deepEqual(lanes.projected_open_items, [1, 0, 2, 4, 3]);
+});
+
+test("invalid evaluated source fails before any lane can escape", () => {
+  for (const fields of [{done: true}, {status: "unknown"}, {task_class: "unknown"},
+    {has_resume: true}, {sort: [true, 1, "", ""]}, {due_at: Infinity}]) {
+    assert.throws(() => project([row(fields)]));
+  }
+});
+
+test("legacy fragments deduplicate identity and cannot invent unseen advancement", () => {
+  const item = {identity: "todo_blocker", task_class: "blocker", actionable: false};
+  const result = projectLegacyTodoWorkCounts({schema_version: "todo_work_counts_request_v0",
+    rows: [item, item], source_open_count: 12});
+  assert.equal(result.advancement, 0); assert.equal(result.hidden, 11); assert.equal(result.complete, false);
+  assert.throws(() => countTodoWork([{taskClass: "blocker", actionable: false}], 0, true));
+  assert.throws(() => countTodoWork([], Number.MAX_SAFE_INTEGER + 1, true));
+});
+
+test("production corpus classification uses every row, independent of display budget", () => {
+  for (const native of [false, true]) {
+    const fixture = productionScaleCoordinationFixture("goal-summary", native ? "native" : "legacy");
+    const todos = fixture.projection.todos as JsonObject[];
+    const result = project(todos.map((todo, ordinal) => row({status: todo.status, done: todo.done,
+      task_class: todo.task_class, claimed: !!todo.claimed_by, sort: [1, ordinal, "", ""]})));
+    assert.equal((result.lanes as JsonObject).budgeted_items instanceof Array, true);
+    assert.equal(((result.lanes as JsonObject).budgeted_items as number[]).length, todos.length);
+    assert.equal((result.work_counts as JsonObject).complete, true);
+  }
+});
+
+test("conflicting display fragments cannot certify a complete source", () => {
+  const result = projectLegacyTodoWorkCounts({schema_version: "todo_work_counts_request_v0", source_open_count: 1,
+    rows: [{identity: "todo_same", task_class: "advancement_task", actionable: false},
+      {identity: "todo_same", task_class: "advancement_task", actionable: true}]});
+  assert.equal(result.complete, false);
+});
diff --git a/tsconfig.control-plane.json b/tsconfig.control-plane.json
index 12c246341e..b0acc3d1b2 100644
--- a/tsconfig.control-plane.json
+++ b/tsconfig.control-plane.json
@@ -97,6 +97,7 @@
     "tests/control_plane_ts/coordination_state_contract.test.ts",
     "tests/control_plane_ts/local_authority_runtime.test.ts",
     "tests/control_plane_ts/todo_completion_update_runtime.test.ts",
+    "tests/control_plane_ts/todo_summary_lanes.test.ts",
     "tests/control_plane_ts/local_archive_attempt.test.ts",
     "tests/control_plane_ts/local_authority_shadow_outbox.test.ts",
     "tests/control_plane_ts/authority_store_conformance.ts",