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 fa8ad2a80c..b94de05b45 100644
--- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
+++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
@@ -3211,6 +3211,15 @@ soak, release, merge and live promotion retain their respective authorization.
 | I. Binding and qualification integration | After C and the selected profile's qualification | Bind one exact provider lineage, field manifest, source revision, digest, and cursor; qualify explicit v0 import, ordering/archival/consumer parity, and recovery/capacity without consulting legacy state for missing fields. | Long-goal local integration requires L and does not wait for P. PostgreSQL joins only when its own P holds pass. |
 | F. Promotion and cleanup | After I and explicit maintainer approval | Complete provider-first CLI routing, the lock-owning promotion orchestrator, compatibility projection outbox, post-promotion fenced export/rollback, then delete duplicate reference aggregates and flip the reviewed stage/hold declarations. | Each profile must pass C, I, and its own provider qualification; long-goal local promotion additionally requires L, and PostgreSQL requires P. |
 
+Agent-addressed read checkpoint: Todo list filtering now joins the typed summary
+batch and shares User gate/action and Agent claim addressing with quota. The
+Python list predicate is retired on both legacy and canonical consumers; full
+source resume/succession and post-filter counts survive display limits. This is
+one L5 consumer closure, not D1 projection freshness or provider promotion. See
+[read semantics](../../reference/todo-work-counts.md). Remaining caller/executor,
+consumer recovery, contributor D2, capture/whole-Goal and default onboarding
+boundaries retain the conditional **5–8 cohesive PR** estimate.
+
 ## Appendix D: Execution ledger
 
 Delivery records for this RFC are files under
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 2e7c50c9a1..6b9634c89e 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
@@ -2523,6 +2523,13 @@ adapter，也不依赖 PostgreSQL service 部署。
 | I. Binding 与资格集成 | C 与选定 profile 的资格化完成后 | 绑定一个精确 provider lineage、field manifest、source revision、digest 与 cursor；资格化显式 v0 import、排序/归档/consumer parity 与 recovery/capacity；缺字段时不得查询 legacy state 补齐。 | 长程本地集成需要 L，不等待 P；PostgreSQL 仅在自己的 P hold 全通过后汇合。 |
 | F. Promotion 与清理 | I 完成且 maintainer 显式批准后 | 完成 provider-first CLI routing、持锁 promotion orchestrator、兼容投影 outbox、晋升后 fenced export/rollback；随后删除重复 reference aggregate，并翻转经评审的 stage/hold 声明。 | 每个 profile 必须通过 C、I 与自身 provider 资格化；长程本地晋升还需 L，PostgreSQL 还需 P。 |
 
+Agent 定向读取检查点：Todo list 筛选已进入现有 TS summary 批次，与 quota 共用
+User gate/action 及 Agent claim 范围规则；legacy 和 canonical 消费者中的 Python
+列表谓词已删除。完整来源上的 resume/succession 与筛选后的计数不受展示上限影响。
+这只闭合 L5 的一个消费者，不代表 D1 永久新鲜度或 provider 晋升。见[读取合同](../../reference/todo-work-counts.md)。
+剩余 caller/executor、consumer recovery、contributor D2、capture/整 Goal 演练和默认
+onboarding 仍按 **5–8 个完整 PR** 条件估计，不能按本次修复机械递减。
+
 ## 附录 D：执行账本
 
 本 RFC 的交付记录是 [`ledger/shared-goal-authority-state-provider-v0/`](ledger/shared-goal-authority-state-provider-v0/) 下的文件，
diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
index 29ce277639..7cb7f305d1 100644
--- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
+++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
@@ -1752,3 +1752,17 @@ behavior.
 Measured delivery records live in the [per-entry ledger](ledger/typescript-control-plane-migration-v0/).
 Each entry names its delivered boundary and remaining acceptance gaps; the T1–T4
 checkpoints above remain the current migration plan.
+
+### T2 Agent-addressed read checkpoint
+
+Todo list selection now composes with the existing typed summary-lanes batch.
+The Python role/status/id/Agent predicates and independent User scope rule are
+removed; legacy and promoted consumers share `todos/agent_scope.ts` with quota
+and decision scope. Explicit gate scope retains precedence over execution claim,
+while retained User claims now correctly restrict scoped list visibility.
+Full-source resume/succession stays evaluated before selection; original array
+ordinals survive filters and display limits. No extra selection runtime crossing,
+new capability/provider, or Python storage migration is introduced. Python keeps
+input normalization and rendering until their actual host consumers migrate.
+See [the read contract](../../reference/todo-work-counts.md); broader L5/D1 and
+local-default qualifications remain open.
diff --git a/docs/development/testing-and-quality.md b/docs/development/testing-and-quality.md
index d6f50bb4c1..cf8f2adebb 100644
--- a/docs/development/testing-and-quality.md
+++ b/docs/development/testing-and-quality.md
@@ -595,6 +595,19 @@ it does not grant execution quota, spending, or provider authority.
    证据中同时保留原失败与新结果。纯预算调整不必捆绑无关清理。已冻结的实验或
    promotion 阈值不能追溯放宽；新阈值属于新一轮验证，不能改写历史结论。
 
+A base/head differential must remain able to measure a syntactically and
+semantically valid base that already exceeds its own historical ceiling;
+otherwise the gate deadlocks the repair before observing the candidate. The
+base-only probe may skip absolute size assertions while retaining parse,
+required-key, anchor, and semantic differential checks. The candidate always
+runs the current absolute budgets. Measurement-only mode is never a candidate
+override or merge bypass.
+
+当 base 已超过自身历史上限但输出仍可解析且语义完整时，base/head differential 必须
+仍能采集它；否则门禁会在观察修复候选之前形成死锁。仅 base 的 probe 可跳过绝对尺寸
+断言，但必须保留解析、必需字段、锚点和语义差异检查；candidate 始终执行当前绝对
+预算。measurement-only 不能用于 candidate，也不是合并旁路。
+
 The PR-review packet's `semantic_alignment` rule consumes this evidence through
 the existing `validation_matrix` and `observable_semantics` rows. It does not
 add a separate budget receipt or approval gate. The result checker verifies
diff --git a/docs/reference/protocols/quota-cli-hot-path-compaction-v0.md b/docs/reference/protocols/quota-cli-hot-path-compaction-v0.md
index 2f2ff687a1..11541e3b41 100644
--- a/docs/reference/protocols/quota-cli-hot-path-compaction-v0.md
+++ b/docs/reference/protocols/quota-cli-hot-path-compaction-v0.md
@@ -67,6 +67,15 @@ projection. Candidate lists and peer action lists retain counts and point to
 `--include-detail vision`; `--include-detail all` restores every supported
 detail section.
 
+When a replan action carries a complete `vision_authoring` schema, the default
+`quota should-run` packet keeps its executable writeback summary (`required_fields`,
+accepted path outcomes, and rule) and replaces only that nested schema with a
+`vision_authoring_detail_ref`. `--include-detail vision` restores the schema.
+`turn plan` is different: its TurnEnvelope preserves the complete schema because
+the plan must be capable of authoring the exact input its validator accepts.
+The crowded Turn budget therefore accounts for that fixed contract without
+relaxing Todo-count growth or the small and multi-Agent ceilings.
+
 ## Qualification Contract
 
 Deterministic tests own exact full-versus-compact parity, cold-path restoration,
diff --git a/docs/reference/todo-work-counts.md b/docs/reference/todo-work-counts.md
index b941a72c9f..27ebe38343 100644
--- a/docs/reference/todo-work-counts.md
+++ b/docs/reference/todo-work-counts.md
@@ -59,6 +59,31 @@ 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.
 
+## Agent-addressed reads
+
+`todo list --agent-id` now composes selection in the same typed summary batch,
+sharing scope rules with quota. For User gates, explicit `global_gate` wins,
+then `blocks_agent`, then the retained `claimed_by` fallback. User actions use
+`bound_agent` first and retained `claimed_by` second. Unscoped records remain
+visible. Gate addressing is independent of executor exclusion: an Agent cannot
+ignore an explicitly addressed human gate because another Agent owns it.
+Agent work still filters by claim and exclusions. A visible row grants no
+mutation or execution permission; quota retains its additional eligibility rules.
+
+This intentionally removes other-Agent, claim-only User records from scoped
+lists; the old Python list rule ignored their claim while quota honored it.
+Unfiltered Goal views retain those records. There is no feature flag or provider
+default change. Existing frontend/Lark manager views use the unfiltered Core
+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.
+
 ## 中文说明
 
 `work_counts` 由完整来源计算，随后才裁剪展示。Agent quota 先按原有归属、排除、
@@ -77,3 +102,13 @@ TS 统一批量 lane 分类与计数，Python 保留旧格式解码、时间适
 这不改变 provider 默认值，不授予执行权限，不写回 Markdown 或 canonical 状态。
 新增计数字段不进入持久化 Todo；回滚无需数据迁移。默认切换、存量迁移、D1–D3 和旧
 Python writer 退出仍有各自的验收条件，不能按本 PR 合并数量推定完成。
+
+Agent 定向列表现与 quota 共用 TS 范围规则：User gate 按 global_gate → blocks_agent →
+旧 claimed_by 依次判定，User action 按 bound_agent → 旧 claimed_by 判定。无作用域的
+旧记录仍可见；显式人类 gate 不会被执行者 claim/exclusion 消除。Agent 工作仍按
+claim/exclusion 筛选，可见不代表获准执行。
+
+这是有意纠正：旧列表忽略仅声明 claimed_by 的 User 记录，导致其他 Agent 的工作混入
+当前列表。未筛选的整 Goal 视图仍显示这些记录。依赖和 succession 先在完整来源求值，
+TS 再筛选并保留原数组位置，最后生成 lanes、计数和有界展示；筛选后的数组位置不是原
+来源位置。无需新增 capability、配置、前端或 Lark 编辑入口，不增加一次筛选 RPC。
diff --git a/examples/control_plane/cli-output-base-head-differential-smoke.py b/examples/control_plane/cli-output-base-head-differential-smoke.py
index 570246195b..92f49795d2 100644
--- a/examples/control_plane/cli-output-base-head-differential-smoke.py
+++ b/examples/control_plane/cli-output-base-head-differential-smoke.py
@@ -58,11 +58,11 @@ def _run_probe(
     fixture_root: Path,
     receipt_path: Path,
     cwd: Path,
+    enforce_budget: bool,
 ) -> None:
     env = os.environ.copy()
     env["PYTHONPATH"] = str(source_root)
-    _run(
-        [
+    command = [
             sys.executable,
             str(probe_runner),
             "--test-source",
@@ -73,7 +73,11 @@ def _run_probe(
             str(fixture_root),
             "--receipt",
             str(receipt_path),
-        ],
+        ]
+    if not enforce_budget:
+        command.append("--measurement-only")
+    _run(
+        command,
         cwd=cwd,
         env=env,
     )
@@ -171,6 +175,7 @@ def main() -> int:
                 fixture_root=fixture_root,
                 receipt_path=base_receipt,
                 cwd=temp_root,
+                enforce_budget=False,
             )
             _run_probe(
                 source_root=REPO_ROOT,
@@ -180,6 +185,7 @@ def main() -> int:
                 fixture_root=fixture_root,
                 receipt_path=candidate_receipt,
                 cwd=temp_root,
+                enforce_budget=True,
             )
         finally:
             _run(
diff --git a/examples/control_plane/cli-output-probe-runner.py b/examples/control_plane/cli-output-probe-runner.py
index 4e1ca37bad..c530dd49bd 100644
--- a/examples/control_plane/cli-output-probe-runner.py
+++ b/examples/control_plane/cli-output-probe-runner.py
@@ -124,10 +124,36 @@ def _receipt_row(
     }
 
 
+def _assert_output_contract(
+    *,
+    output_format: str,
+    text: str,
+    measurement: dict,
+    semantic_json_keys: tuple[str, ...],
+    markdown_anchor: str | None,
+) -> None:
+    """Validate shape while allowing a red base to remain measurable."""
+
+    if output_format == "markdown":
+        if markdown_anchor and markdown_anchor not in text:
+            raise AssertionError(
+                f"markdown output lost semantic anchor {markdown_anchor!r}"
+            )
+        return
+    payload = measurement.get("payload")
+    if not isinstance(payload, dict):
+        raise AssertionError("JSON output did not emit an object")
+    missing = [key for key in semantic_json_keys if key not in payload]
+    if missing:
+        raise AssertionError(f"JSON output lost semantic key(s): {', '.join(missing)}")
+
+
 def _default_rows(
     probe: ModuleType,
     semantics: ModuleType,
     fixture_root: Path,
+    *,
+    enforce_budget: bool = True,
 ) -> list[dict]:
     rows: list[dict] = []
     for scenario in probe.SCENARIOS:
@@ -151,13 +177,22 @@ def _default_rows(
                     text, output_format=output_format
                 )
                 surface = probe.CLI_OUTPUT_BUDGET_BY_ID[surface_id]
-                probe.assert_cli_output_baseline(
-                    surface,
-                    scenario=scenario.name,
-                    output_format=output_format,
-                    text=text,
-                    measurement=measurement,
-                )
+                if enforce_budget:
+                    probe.assert_cli_output_baseline(
+                        surface,
+                        scenario=scenario.name,
+                        output_format=output_format,
+                        text=text,
+                        measurement=measurement,
+                    )
+                else:
+                    _assert_output_contract(
+                        output_format=output_format,
+                        text=text,
+                        measurement=measurement,
+                        semantic_json_keys=surface.semantic_json_keys,
+                        markdown_anchor=surface.markdown_anchor,
+                    )
                 rows.append(
                     _receipt_row(
                         semantics=semantics,
@@ -184,6 +219,8 @@ def _variant_rows(
     probe: ModuleType,
     semantics: ModuleType,
     fixture_root: Path,
+    *,
+    enforce_budget: bool = True,
 ) -> list[dict]:
     project, runtime, registry_path, state_file = probe._write_fixture(
         fixture_root / "mode_variants",
@@ -210,12 +247,21 @@ def _variant_rows(
             if exit_code != 0:
                 raise AssertionError(f"{variant_id}/{output_format} failed")
             measurement = probe.measure_cli_output(text, output_format=output_format)
-            probe.assert_cli_output_mode_variant(
-                variant,
-                output_format=output_format,
-                text=text,
-                measurement=measurement,
-            )
+            if enforce_budget:
+                probe.assert_cli_output_mode_variant(
+                    variant,
+                    output_format=output_format,
+                    text=text,
+                    measurement=measurement,
+                )
+            else:
+                _assert_output_contract(
+                    output_format=output_format,
+                    text=text,
+                    measurement=measurement,
+                    semantic_json_keys=variant.semantic_json_keys,
+                    markdown_anchor=variant.markdown_anchor,
+                )
             rows.append(
                 _receipt_row(
                     semantics=semantics,
@@ -238,6 +284,8 @@ def _blocking_gate_rows(
     probe: ModuleType,
     semantics: ModuleType,
     fixture_root: Path,
+    *,
+    enforce_budget: bool = True,
 ) -> list[dict]:
     project, runtime, registry_path, state_file = probe._write_fixture(
         fixture_root / "blocking_user_gate",
@@ -266,12 +314,21 @@ def _blocking_gate_rows(
     variant = probe.CLI_OUTPUT_MODE_VARIANT_BY_ID[
         "quota_should_run_turn_envelope"
     ]
-    probe.assert_cli_output_mode_variant(
-        variant,
-        output_format="json",
-        text=output,
-        measurement=measurement,
-    )
+    if enforce_budget:
+        probe.assert_cli_output_mode_variant(
+            variant,
+            output_format="json",
+            text=output,
+            measurement=measurement,
+        )
+    else:
+        _assert_output_contract(
+            output_format="json",
+            text=output,
+            measurement=measurement,
+            semantic_json_keys=variant.semantic_json_keys,
+            markdown_anchor=variant.markdown_anchor,
+        )
     return [
         _receipt_row(
             semantics=semantics,
@@ -293,7 +350,7 @@ def _blocking_gate_rows(
 
 
 
-def _multi_subagent_rows(probe, semantics, fixture_root):
+def _multi_subagent_rows(probe, semantics, fixture_root, *, enforce_budget=True):
     """Run the same enabled public fixture on base and head, not default-off only."""
     project, runtime, registry_path, state_file = probe._write_fixture(
         fixture_root / "multi_subagent_enabled", probe.SCENARIOS[0]
@@ -314,9 +371,16 @@ def _multi_subagent_rows(probe, semantics, fixture_root):
         raise AssertionError("enabled multi_subagent turn envelope failed")
     measurement = probe.measure_cli_output(output, output_format="json")
     variant = probe.CLI_OUTPUT_MODE_VARIANT_BY_ID[variant_id]
-    probe.assert_cli_output_mode_variant(
-        variant, output_format="json", text=output, measurement=measurement,
-    )
+    if enforce_budget:
+        probe.assert_cli_output_mode_variant(
+            variant, output_format="json", text=output, measurement=measurement,
+        )
+    else:
+        _assert_output_contract(
+            output_format="json", text=output, measurement=measurement,
+            semantic_json_keys=variant.semantic_json_keys,
+            markdown_anchor=variant.markdown_anchor,
+        )
     return [_receipt_row(
         semantics=semantics,
         row_id="variant/quota_should_run_turn_envelope_multi_subagent/small/json",
@@ -334,6 +398,11 @@ def main() -> int:
     parser.add_argument("--semantics-source", type=Path, required=True)
     parser.add_argument("--fixture-root", type=Path, required=True)
     parser.add_argument("--receipt", type=Path, required=True)
+    parser.add_argument(
+        "--measurement-only",
+        action="store_true",
+        help="measure a historical base without requiring its retired ceilings to pass",
+    )
     args = parser.parse_args()
     _install_pytest_import_stub()
     probe = _load_module("loopx_cli_output_probe_fixture", args.test_source)
@@ -343,10 +412,22 @@ def main() -> int:
     args.fixture_root.mkdir(parents=True)
     with probe._stable_budget_fixture_root(args.fixture_root) as stable_root:
         rows = [
-            *_default_rows(probe, semantics, stable_root),
-            *_variant_rows(probe, semantics, stable_root),
-            *_blocking_gate_rows(probe, semantics, stable_root),
-            *_multi_subagent_rows(probe, semantics, stable_root),
+            *_default_rows(
+                probe, semantics, stable_root,
+                enforce_budget=not args.measurement_only,
+            ),
+            *_variant_rows(
+                probe, semantics, stable_root,
+                enforce_budget=not args.measurement_only,
+            ),
+            *_blocking_gate_rows(
+                probe, semantics, stable_root,
+                enforce_budget=not args.measurement_only,
+            ),
+            *_multi_subagent_rows(
+                probe, semantics, stable_root,
+                enforce_budget=not args.measurement_only,
+            ),
         ]
     args.receipt.parent.mkdir(parents=True, exist_ok=True)
     args.receipt.write_text(
diff --git a/loopx/control_plane/quota/cli_projection.py b/loopx/control_plane/quota/cli_projection.py
index c5a32b02b4..c977c9005c 100644
--- a/loopx/control_plane/quota/cli_projection.py
+++ b/loopx/control_plane/quota/cli_projection.py
@@ -26,6 +26,9 @@
     "quota_cli_vision_continuation_compaction_v0"
 )
 QUOTA_CLI_VISION_DETAIL_COMMAND = "quota should-run --include-detail vision"
+QUOTA_CLI_REPLAN_ACTION_COMPACTION_SCHEMA_VERSION = (
+    "quota_cli_replan_action_compaction_v0"
+)
 QUOTA_CLI_CAPABILITY_GATE_COMPACTION_SCHEMA_VERSION = (
     "quota_cli_capability_gate_compaction_v0"
 )
@@ -454,6 +457,28 @@ def _compact_vision_continuation_audit(
     return compact
 
 
+def _compact_replan_action_packet(packet: dict[str, Any]) -> dict[str, Any]:
+    """Keep the executable writeback summary hot and move its schema cold."""
+
+    writeback = packet.get("writeback_contract")
+    if not isinstance(writeback, dict) or not isinstance(
+        writeback.get("vision_authoring"), dict
+    ):
+        return packet
+    compact_writeback = dict(writeback)
+    compact_writeback.pop("vision_authoring")
+    compact_writeback["vision_authoring_detail_ref"] = QUOTA_CLI_VISION_DETAIL_COMMAND
+    compact = dict(packet)
+    compact["writeback_contract"] = compact_writeback
+    compact["payload_compaction"] = {
+        "schema_version": QUOTA_CLI_REPLAN_ACTION_COMPACTION_SCHEMA_VERSION,
+        "mode": "compact_hot_path",
+        "compacted_fields": ["writeback_contract.vision_authoring"],
+        "full_detail_cold_path": QUOTA_CLI_VISION_DETAIL_COMMAND,
+    }
+    return compact
+
+
 def _vision_continuation_ref(audit: dict[str, Any]) -> dict[str, Any]:
     compact = {
         key: audit[key]
@@ -727,6 +752,12 @@ def compact_quota_should_run_cli_payload(
             source_audit=vision_audit,
             audit_ref=_vision_continuation_ref(compact_vision_audit),
         )
+    replan_action = payload.get("replan_action_packet")
+    if not include_vision_detail and isinstance(replan_action, dict):
+        compact_replan_action = _compact_replan_action_packet(replan_action)
+        if compact_replan_action is not replan_action:
+            compact = dict(compact)
+            compact["replan_action_packet"] = compact_replan_action
     if not include_todo_summary_detail:
         action_portfolio = payload.get("action_portfolio")
         if isinstance(action_portfolio, dict):
diff --git a/loopx/control_plane/testing/cli_output_budget.py b/loopx/control_plane/testing/cli_output_budget.py
index 2c81da4245..7d58113171 100644
--- a/loopx/control_plane/testing/cli_output_budget.py
+++ b/loopx/control_plane/testing/cli_output_budget.py
@@ -43,6 +43,7 @@ class CliOutputBudgetSpec:
     max_lines: dict[str, dict[OutputFormat, int]]
     scale_axis: str | None = None
     max_json_growth_chars_per_unit: int | None = None
+    max_json_fixed_semantic_growth_chars: int = 0
     output_contract_version: str | None = None
 
 
@@ -164,16 +165,32 @@ class CliOutputCommandClassification:
         markdown_anchor="# LoopX Turn Plan",
         max_chars={
             "small": {"json": 12_000, "markdown": 300},
-            "crowded": {"json": 12_000, "markdown": 300},
+            # The crowded fixture exercises the required-vision route. Its
+            # TurnEnvelope intentionally carries the complete authoring schema
+            # that the validator accepts, plus the typed executor and selection
+            # facts needed to decide whether execution is authorized. The
+            # latest-main fixture measures 14,159 chars, so 14,500 retains a
+            # narrow 341-char regression margin without relaxing Todo growth.
+            # The over-target TurnEnvelope diagnostic remains visible instead
+            # of hiding authority overflow; latest main renders it in 542
+            # characters, leaving a narrow 58-character presentation margin.
+            "crowded": {"json": 14_500, "markdown": 600},
             "multi_agent": {"json": 12_000, "markdown": 300},
         },
         max_lines={
             "small": {"json": 320, "markdown": 12},
-            "crowded": {"json": 320, "markdown": 12},
+            # The same latest-main fixture measures 389 lines. Keep a bounded
+            # 11-line formatting margin while the semantic character budget
+            # above remains the primary cost guard.
+            "crowded": {"json": 400, "markdown": 12},
             "multi_agent": {"json": 320, "markdown": 12},
         },
         scale_axis="todo_count",
         max_json_growth_chars_per_unit=60,
+        # The complete validator-owned vision-authoring schema appears only on
+        # the required-vision route. Account for that fixed semantic packet
+        # separately so it does not relax the per-Todo growth budget.
+        max_json_fixed_semantic_growth_chars=3_800,
     ),
     CliOutputBudgetSpec(
         surface_id="status",
diff --git a/loopx/control_plane/todos/agent_scope.ts b/loopx/control_plane/todos/agent_scope.ts
new file mode 100644
index 0000000000..b418b516c8
--- /dev/null
+++ b/loopx/control_plane/todos/agent_scope.ts
@@ -0,0 +1,27 @@
+/** Shared read addressing; visibility never grants mutation or execution. */
+export interface GateScope {
+  global: boolean;
+  blocks: string | null;
+  claim: string | null;
+}
+
+export function gateAddressesAgent(gate: GateScope, agent: string | null): boolean {
+  if (gate.global) return true;
+  if (gate.blocks) return gate.blocks === agent;
+  return !gate.claim || gate.claim === agent;
+}
+
+
+export interface ActionScope {bound: string | null; claim: string | null}
+export interface ClaimScope {claim: string | null; excluded: readonly string[]}
+
+/** Explicit action binding wins; retained unbound actions use their claim. */
+export function actionAddressesAgent(action: ActionScope, agent: string | null): boolean {
+  const bound = action.bound ?? action.claim;
+  return !agent || !bound || bound === agent;
+}
+
+/** Claim/exclusion address Agent work, not permission to disregard User gates. */
+export function claimAllowsAgent(work: ClaimScope, agent: string | null): boolean {
+  return !agent || (!work.excluded.includes(agent) && (!work.claim || work.claim === agent));
+}
diff --git a/loopx/control_plane/todos/decision_scope.ts b/loopx/control_plane/todos/decision_scope.ts
index 7cc4ac78ee..8585447cf7 100644
--- a/loopx/control_plane/todos/decision_scope.ts
+++ b/loopx/control_plane/todos/decision_scope.ts
@@ -2,7 +2,7 @@
  * A consistent dependency is not approval, a lease, or a mutation receipt. */
 import type {JsonObject} from "../effect_program.ts";
 import {requireJsonObject, optionalNonEmptyString, requireBoolean, requireInteger} from "../runtime_decode.ts";
-import {gateAddressesAgent} from "./gate_scope.ts";
+import {gateAddressesAgent} from "./agent_scope.ts";
 import {
   TODO_DECISION_SCOPE_GRANULARITY_SET,
   TODO_DECISION_SCOPE_KEY_PATTERN,
diff --git a/loopx/control_plane/todos/gate_scope.ts b/loopx/control_plane/todos/gate_scope.ts
deleted file mode 100644
index d0bf3931d9..0000000000
--- a/loopx/control_plane/todos/gate_scope.ts
+++ /dev/null
@@ -1,12 +0,0 @@
-/** Addressed permission scope is independent of execution ownership. */
-export interface GateScope {
-  global: boolean;
-  blocks: string | null;
-  claim: string | null;
-}
-
-export function gateAddressesAgent(gate: GateScope, agent: string | null): boolean {
-  if (gate.global) return true;
-  if (gate.blocks) return gate.blocks === agent;
-  return !gate.claim || gate.claim === agent;
-}
diff --git a/loopx/control_plane/todos/goal_todo_projection.py b/loopx/control_plane/todos/goal_todo_projection.py
index bb86b7c643..736f5f0081 100644
--- a/loopx/control_plane/todos/goal_todo_projection.py
+++ b/loopx/control_plane/todos/goal_todo_projection.py
@@ -18,14 +18,11 @@
 from .succession_warning import public_todo_summary
 from .contract import (
     build_todo_id,
-    normalize_todo_blocks_agent,
-    normalize_todo_bound_agent,
     normalize_todo_claimed_by,
-    normalize_todo_excluded_agents,
     normalize_todo_id,
     normalize_todo_status,
 )
-from .todo_summary import compact_evaluated_todo_group, compact_todo_group, todo_item_status
+from .todo_summary import compact_evaluated_todo_group, compact_todo_group
 
 
 def empty_todo_summary(*, role: str) -> dict[str, Any]:
@@ -40,17 +37,6 @@ def empty_todo_summary(*, role: str) -> dict[str, Any]:
         "first_open_items": [],
     }
 
-def _user_todo_visible_to_agent(item: dict[str, Any], agent_id: str) -> bool:
-    if bool(item.get("global_gate")):
-        return True
-    blocks_agent = normalize_todo_blocks_agent(item.get("blocks_agent"))
-    if blocks_agent:
-        return blocks_agent == agent_id
-    bound_agent = normalize_todo_bound_agent(item.get("bound_agent"))
-    if bound_agent:
-        return bound_agent == agent_id
-    return True
-
 def filtered_todo_summary(
     summary: dict[str, Any] | None,
     *,
@@ -61,36 +47,9 @@ def filtered_todo_summary(
     item_limit: int | None = None,
 ) -> dict[str, Any]:
     items = list((summary or {}).get("items") or [])
-    normalized_status = normalize_todo_status(status)
-    if normalized_status:
-        items = [item for item in items if todo_item_status(item) == normalized_status]
-    normalized_todo_id = normalize_todo_id(todo_id) if todo_id else None
-    if normalized_todo_id:
-        items = [
-            item
-            for item in items
-            if normalize_todo_id(item.get("todo_id")) == normalized_todo_id
-        ]
-    normalized_agent_id = normalize_todo_claimed_by(agent_id) if agent_id else None
-    if normalized_agent_id:
-        if role == "agent":
-            items = [
-                item
-                for item in items
-                if normalized_agent_id
-                not in normalize_todo_excluded_agents(item.get("excluded_agents"))
-                and (
-                    not normalize_todo_claimed_by(item.get("claimed_by"))
-                    or normalize_todo_claimed_by(item.get("claimed_by"))
-                    == normalized_agent_id
-                )
-            ]
-        elif role == "user":
-            items = [
-                item
-                for item in items
-                if _user_todo_visible_to_agent(item, normalized_agent_id)
-            ]
+    selection = {"role": role, "status": normalize_todo_status(status),
+        "todo_id": normalize_todo_id(todo_id) if todo_id else None,
+        "agent_id": normalize_todo_claimed_by(agent_id) if agent_id else None}
     source_section = str((summary or {}).get("source_section") or TODO_SECTION_HEADINGS[role])
     return (
         compact_evaluated_todo_group(
@@ -98,7 +57,7 @@ def filtered_todo_summary(
             source_section=source_section,
             role=role,
             item_limit=item_limit,
-            full_selection=not (normalized_status or normalized_todo_id or normalized_agent_id),
+            selection=selection,
         )
         or empty_todo_summary(role=role)
     )
diff --git a/loopx/control_plane/todos/quota_selection.ts b/loopx/control_plane/todos/quota_selection.ts
index 6bc5295d3f..3b6bd1fa54 100644
--- a/loopx/control_plane/todos/quota_selection.ts
+++ b/loopx/control_plane/todos/quota_selection.ts
@@ -4,7 +4,7 @@ import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts";
 import { requireJsonObject, requireBoolean, requireInteger, requireStringArray,
   optionalNonEmptyString } from "../runtime_decode.ts";
 import { projectTodoResumePlanning } from "./resume_planning.ts";
-import { gateAddressesAgent } from "./gate_scope.ts";
+import { gateAddressesAgent, actionAddressesAgent, claimAllowsAgent } from "./agent_scope.ts";
 import { missingRequiredCapabilities } from "../agents/capability_gate.ts";
 
 interface Row {
@@ -45,12 +45,8 @@ const bucket = (row: Row, agent: string) => row.claim === agent ? 0 : row.claim
 function gateApplies(row: Row, agent: string | null): boolean {
   return !agent || gateAddressesAgent(row, agent);
 }
-function actionApplies(row: Row, agent: string | null): boolean {
-  const bound = row.bound ?? row.claim;
-  return !agent || !bound || bound === agent;
-}
 function executableBy(row: Row, agent: string | null): boolean {
-  return !agent || (!row.removed && !row.excluded.includes(agent) && bucket(row, agent) !== 2);
+  return !agent || (!row.removed && claimAllowsAgent(row, agent));
 }
 
 /** Presentation-only claimant coverage; never changes eligible work or counts. */
@@ -139,8 +135,8 @@ export function projectQuotaSelection(value: unknown): JsonObject {
   const gates = userMode ? source.filter(row => row.gate) : source;
   const blocking = userMode ? gates.filter(row => gateApplies(row, agent)) : gates;
   const otherGates = userMode ? gates.filter(row => !gateApplies(row, agent)) : [];
-  const actions = userMode ? source.filter(row => !row.gate && actionApplies(row, agent)) : [];
-  const otherActions = userMode ? source.filter(row => !row.gate && !actionApplies(row, agent)) : [];
+  const actions = userMode ? source.filter(row => !row.gate && actionAddressesAgent(row, agent)) : [];
+  const otherActions = userMode ? source.filter(row => !row.gate && !actionAddressesAgent(row, agent)) : [];
   // Explicit User gate scope has already decided blocking. Claim/exclusion
   // governs Agent execution, not permission to disregard that human gate.
   const open = userMode ? blocking : blocking.filter(row => executableBy(row, agent));
@@ -151,7 +147,7 @@ export function projectQuotaSelection(value: unknown): JsonObject {
   const due = supported ? monitors.filter(row => row.due && executableBy(row, agent)) : [];
   const admittedDue = due.filter(row => !row.missing.length);
   const watchOnlyMonitors = monitors.filter(row => row.watchOnly);
-  const activeVisible = (row: Row) => userMode ? (row.gate ? gateApplies(row, agent) : actionApplies(row, agent)) : executableBy(row, agent);
+  const activeVisible = (row: Row) => userMode ? (row.gate ? gateApplies(row, agent) : actionAddressesAgent(row, agent)) : executableBy(row, agent);
   const gateFilter = otherGates.length ? {
     schema_version: "agent_scoped_user_gate_filter_v0", agent_id: agent,
     policy: "user todos scoped to another agent by blocks_agent or claimed_by remain visible but do not block this agent's quota lane",
diff --git a/loopx/control_plane/todos/summary_lanes.ts b/loopx/control_plane/todos/summary_lanes.ts
index 7051465bb3..f046a6bec6 100644
--- a/loopx/control_plane/todos/summary_lanes.ts
+++ b/loopx/control_plane/todos/summary_lanes.ts
@@ -1,7 +1,8 @@
 /** 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 {requireBoolean, requireJsonObject, requireStringLiteral, optionalNonEmptyString, requireStringArray} from "../runtime_decode.ts";
+import {gateAddressesAgent, actionAddressesAgent, claimAllowsAgent} from "./agent_scope.ts";
 import {authorityUnicodeCompare} from "../coordination/authority_store_codec.ts";
 
 export const TODO_SUMMARY_LANES = [
@@ -70,12 +71,37 @@ export function countTodoWork(rows: readonly WorkCountRow[], sourceOpenCount: nu
     complete: complete && sourceOpenCount === rows.length, agent_id: agentId};
 }
 
+/** Filter already evaluated full-source rows. Keep their original ordinals:
+ * consumers must not confuse a selected position with a source identity. */
+function selectRows(rows: readonly Row[], source: readonly unknown[], value: unknown): {rows: Row[]; full: boolean} {
+  const selection = requireJsonObject(value, "Todo read selection");
+  const role = requireStringLiteral(selection.role, ["user", "agent"], "selection role");
+  const status = selection.status == null ? null :
+    requireStringLiteral(selection.status, ["open", "blocked", "done", "deferred"], "selection status");
+  const todo = optionalNonEmptyString(selection.todo_id, "selection todo_id");
+  const agent = optionalNonEmptyString(selection.agent_id, "selection agent_id");
+  const addressed = rows.filter(row => {
+    const raw = requireJsonObject(source[row.ordinal], "Todo source row");
+    const optional = (key: string) => optionalNonEmptyString(raw[key], key);
+    const id = optional("todo_id");
+    const scope = {claim: optional("claim"), bound: optional("bound"), blocks: optional("blocks"),
+      global: requireBoolean(raw.global, "global"), excluded: requireStringArray(raw.excluded, "excluded")};
+    const visible = !agent || (role === "agent" ? claimAllowsAgent(scope, agent) :
+      row.taskClass === "user_gate" ? gateAddressesAgent(scope, agent) : actionAddressesAgent(scope, agent));
+    return (!status || row.status === status) && (!todo || id === todo) && visible;
+  });
+  return {rows: addressed, full: !status && !todo && !agent};
+}
+
 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)) {
+  if (!["todo_summary_lanes_request_v0", "todo_summary_lanes_request_v1"].includes(String(request.schema_version)) || !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 decoded = request.rows.map(decodeRow), now = finite(request.observed_at, "observed_at");
+  const selection = request.schema_version === "todo_summary_lanes_request_v1"
+    ? selectRows(decoded, request.rows, request.selection) : {rows: decoded, full: true};
+  const rows = selection.rows;
   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);
@@ -106,6 +132,8 @@ export function projectTodoSummaryLanes(value: unknown): JsonObject {
   } 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,
+    ...(request.schema_version === "todo_summary_lanes_request_v1" ? {
+      source_indices: rows.map(row => row.ordinal), full_selection: selection.full} : {}),
     work_counts: countTodoWork(open, open.length, true)};
 }
 
diff --git a/loopx/control_plane/todos/todo_summary.py b/loopx/control_plane/todos/todo_summary.py
index d2581dee46..6b74353c78 100644
--- a/loopx/control_plane/todos/todo_summary.py
+++ b/loopx/control_plane/todos/todo_summary.py
@@ -895,7 +895,9 @@ def _structured_resume_source_items(
     ]
 
 
-def _project_summary_lanes(items: list[dict[str, Any]], preferred_todo_ids: set[str] | None) -> dict[str, Any]:
+def _project_summary_lanes(items: list[dict[str, Any]], preferred_todo_ids: set[str] | None,
+    selection: dict[str, Any] | None = None,
+) -> dict[str, Any]:
     """Adapt legacy facts and read batch ordinals from the typed lane owner."""
     from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result
 
@@ -918,23 +920,42 @@ def _project_summary_lanes(items: list[dict[str, Any]], preferred_todo_ids: set[
             "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))})
+            "sort": list(projection_todo_presentation_sort_key(item)),
+            **({"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 {})})
     try:
         result = effect_runtime_result("todo.summary_lanes.project", {
-            "schema_version": "todo_summary_lanes_request_v0", "rows": rows, "observed_at": now_utc().timestamp(),
+            "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 {}),
         })
     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:
+        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 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()
-    ):
+    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)):
+        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()},
-            "work_counts": result["work_counts"]}
+            "items": [items[index] for index in selected],
+            "full_selection": result.get("full_selection", True), "work_counts": result["work_counts"]}
 
 
 def compact_todo_group(
@@ -993,6 +1014,7 @@ def compact_evaluated_todo_group(
     vision_runs: list[dict[str, Any]] | None = None,
     lineage_items: list[dict[str, Any]] | None = None,
     full_selection: bool = True,
+    selection: dict[str, Any] | None = None,
 ) -> dict[str, Any] | None:
     """Filter/display an already evaluated snapshot, never re-evaluate topology.
 
@@ -1001,7 +1023,12 @@ def compact_evaluated_todo_group(
     """
     if not items and not include_empty_source:
         return None
-    projected = _project_summary_lanes(items, preferred_todo_ids)
+    projected = _project_summary_lanes(items, preferred_todo_ids, 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
 
diff --git a/loopx/todos.py b/loopx/todos.py
index 321b647ede..2b24e86e08 100644
--- a/loopx/todos.py
+++ b/loopx/todos.py
@@ -303,8 +303,8 @@ def list_goal_todos(
         payload["unfiltered_todo_count"] = unfiltered_count
         payload["filter_semantics"] = (
             "agent todos include unclaimed items plus claimed_by=<agent>; "
-            "user todos include global, unscoped legacy, blocks_agent=<agent> gates, "
-            "and bound_agent=<agent> actions"
+            "User gates use global_gate, then blocks_agent, then legacy claimed_by scope; "
+            "User actions use bound_agent, then legacy claimed_by scope; unscoped items remain visible"
         )
     if agent_lane_hot_path:
         payload["returned_todo_count"] = len(todos)
diff --git a/tests/control_plane/test_cli_output_budget.py b/tests/control_plane/test_cli_output_budget.py
index 8c16665160..fef2bcbc6a 100644
--- a/tests/control_plane/test_cli_output_budget.py
+++ b/tests/control_plane/test_cli_output_budget.py
@@ -1092,6 +1092,22 @@ def test_quota_cli_bounds_real_scale_vision_audit_and_keeps_cold_detail(
     )
     assert "registry_read_instruction" not in compact_audit["vision_gap_judge"]
     assert "registry_read_instruction" in detailed_audit["vision_gap_judge"]
+    compact_replan = default_payload["replan_action_packet"]
+    detailed_replan = detail_payload["replan_action_packet"]
+    assert compact_replan["payload_compaction"] == {
+        "schema_version": "quota_cli_replan_action_compaction_v0",
+        "mode": "compact_hot_path",
+        "compacted_fields": ["writeback_contract.vision_authoring"],
+        "full_detail_cold_path": "quota should-run --include-detail vision",
+    }
+    assert "vision_authoring" not in compact_replan["writeback_contract"]
+    assert compact_replan["writeback_contract"]["vision_authoring_detail_ref"] == (
+        "quota should-run --include-detail vision"
+    )
+    assert detailed_replan["writeback_contract"]["vision_authoring"][
+        "schema_version"
+    ] == "goal_vision_replan_contract_v0"
+    assert "payload_compaction" not in detailed_replan
     assert default_payload["goal_frontier_projection"][
         "vision_continuation_audit"
     ]["projection_ref"] == "$.vision_continuation_audit"
@@ -1126,6 +1142,38 @@ def test_quota_cli_bounds_real_scale_vision_audit_and_keeps_cold_detail(
     )
 
 
+def test_crowded_turn_plan_budget_preserves_executable_vision_authoring(
+    tmp_path: Path,
+) -> None:
+    with _stable_budget_fixture_root(tmp_path / "turn-plan-vision") as stable_root:
+        project, runtime, registry_path, state_file = _write_fixture(
+            stable_root,
+            SCENARIOS[1],
+        )
+        command = _surface_commands(
+            project=project,
+            runtime=runtime,
+            registry_path=registry_path,
+            state_file=state_file,
+            output_format="json",
+        )["loopx_turn_plan"]
+        exit_code, text = _invoke_cli(command)
+
+    assert exit_code == 0, text
+    payload = json.loads(text)
+    writeback = payload["turn_envelope"]["replan_action_packet"][
+        "writeback_contract"
+    ]
+    assert "path_delta.evidence_refs" in writeback["required_fields"]
+    assert writeback["vision_authoring"]["schema_version"] == (
+        "goal_vision_replan_contract_v0"
+    )
+    # This fixed executable schema legitimately crosses the old 12k/320
+    # ceiling; retain bounded headroom without relaxing Todo-scale growth.
+    assert 12_000 < len(text) <= 14_500
+    assert len(text.splitlines()) <= 400
+
+
 def test_quota_cli_keeps_full_user_todo_diagnostics_on_explicit_cold_path(
     tmp_path: Path,
 ) -> None:
@@ -1413,10 +1461,33 @@ def test_collection_growth_and_bootstrap_duplication_are_explicit(tmp_path: Path
             crowded[spec.surface_id]["json"]["chars"]
             - small[spec.surface_id]["json"]["chars"]
         )
-        assert growth <= spec.max_json_growth_chars_per_unit * units, (
+        fixed_semantic_growth = spec.max_json_fixed_semantic_growth_chars
+        if fixed_semantic_growth:
+            assert spec.surface_id == "loopx_turn_plan"
+            small_packet = small[spec.surface_id]["json"]["payload"][
+                "turn_envelope"
+            ].get("replan_action_packet")
+            crowded_packet = crowded[spec.surface_id]["json"]["payload"][
+                "turn_envelope"
+            ]["replan_action_packet"]
+            small_writeback = (
+                small_packet.get("writeback_contract")
+                if isinstance(small_packet, dict)
+                else None
+            )
+            assert not isinstance(small_writeback, dict) or not isinstance(
+                small_writeback.get("vision_authoring"), dict
+            )
+            assert isinstance(
+                crowded_packet["writeback_contract"]["vision_authoring"], dict
+            )
+        assert growth <= (
+            spec.max_json_growth_chars_per_unit * units + fixed_semantic_growth
+        ), (
             spec.surface_id,
             growth,
             units,
+            fixed_semantic_growth,
         )
 
     start_payload = small["start_goal_guided"]["json"]["payload"]
diff --git a/tests/control_plane/test_cli_output_differential.py b/tests/control_plane/test_cli_output_differential.py
index b584c39d4b..8d930201f2 100644
--- a/tests/control_plane/test_cli_output_differential.py
+++ b/tests/control_plane/test_cli_output_differential.py
@@ -990,3 +990,31 @@ def test_public_multi_subagent_probe_reaches_v4_producer(tmp_path):
         "turn_envelope_action_dimensions_v4"
     ]
     assert any("agent_context" in path for path in rows[0]["json_shape_paths"])
+
+
+def test_measurement_only_probe_skips_ceiling_but_keeps_semantic_shape() -> None:
+    import runpy
+    from pathlib import Path
+
+    runner = runpy.run_path(
+        str(
+            Path(__file__).resolve().parents[2]
+            / "examples/control_plane/cli-output-probe-runner.py"
+        )
+    )
+    validate = runner["_assert_output_contract"]
+    validate(
+        output_format="json",
+        text='{"required": true}',
+        measurement={"payload": {"required": True}},
+        semantic_json_keys=("required",),
+        markdown_anchor=None,
+    )
+    with pytest.raises(AssertionError, match="lost semantic key"):
+        validate(
+            output_format="json",
+            text="{}",
+            measurement={"payload": {}},
+            semantic_json_keys=("required",),
+            markdown_anchor=None,
+        )
diff --git a/tests/control_plane/test_todo_consumer_scope.py b/tests/control_plane/test_todo_consumer_scope.py
new file mode 100644
index 0000000000..f8fbaa0a49
--- /dev/null
+++ b/tests/control_plane/test_todo_consumer_scope.py
@@ -0,0 +1,135 @@
+"""List and quota address the same Agent lane without granting execution."""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from canonical_authority_fixture import initialize_canonical_authority
+from loopx.control_plane.testing.canary_harness import write_fixture_registry, run_json_cli_result
+from loopx.control_plane.todos.goal_todo_projection import filtered_todo_summary
+from loopx.control_plane.todos.quota_summary import summarize_user_todos_for_quota
+from loopx.control_plane.todos.todo_summary import compact_todo_group
+
+
+@pytest.mark.parametrize("task_class,scope,visible", [
+    ("user_gate", {"claimed_by": "agent-b"}, False),
+    ("user_action", {"claimed_by": "agent-b"}, False),
+    ("user_gate", {"claimed_by": "agent-a"}, True),
+    ("user_action", {"claimed_by": "agent-a"}, True),
+    ("user_gate", {"claimed_by": "agent-b", "blocks_agent": "agent-a"}, True),
+    ("user_gate", {"claimed_by": "agent-a", "blocks_agent": "agent-b"}, False),
+    ("user_gate", {"claimed_by": "agent-b", "global_gate": True, "excluded_agents": ["agent-a"]}, True),
+    ("user_action", {"claimed_by": "agent-b", "bound_agent": "agent-a"}, True),
+    ("user_action", {"claimed_by": "agent-a", "bound_agent": "agent-b"}, False),
+    ("user_action", {}, True),
+    ("advancement_task", {"bound_agent": "agent-b"}, False),
+    ("user_gate", {}, True),
+])
+def test_list_and_quota_share_addressed_scope(task_class, scope, visible):
+    source = compact_todo_group([{"todo_id": "todo_scoped", "text": "Review the result",
+        "role": "user", "status": "open", "task_class": task_class, **scope}],
+        role="user", source_section="User Todo", item_limit=None)
+    selected = filtered_todo_summary(source, role="user", agent_id="agent-a")
+    quota = summarize_user_todos_for_quota(source, agent_identity={"agent_id": "agent-a"},
+        filter_user_gate_blocks_agent=True)
+    assert bool(selected["items"]) is visible
+    assert quota["open_count"] == int(visible)
+    assert selected["open_count"] == int(visible)
+    # Full Goal read remains diagnostic, including work addressed to other Agents.
+    assert filtered_todo_summary(source, role="user")["total_count"] == 1
+
+
+@pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"])
+def test_real_cli_scope_survives_missing_display_and_limits(tmp_path: Path, provider: str):
+    state, registry, runtime = tmp_path / "STATE.md", tmp_path / "registry.json", tmp_path / "runtime"
+    user_rows = [
+        ("todo_peer_gate", "user_gate", "claimed_by=agent-b"),
+        ("todo_peer_action", "user_action", "claimed_by=agent-b"),
+        ("todo_explicit_gate", "user_gate", "claimed_by=agent-b blocks_agent=agent-a"),
+        ("todo_explicit_action", "user_action", "claimed_by=agent-b bound_agent=agent-a"),
+        ("todo_global", "user_gate", "claimed_by=agent-b global_gate=true"),
+    ]
+    text = "---\nstatus: active\n---\n# Synthetic Goal\n\n## User Todo\n"
+    for todo_id, task_class, scope in user_rows:
+        text += f"- [ ] [P1] Review {todo_id}\n  <!-- loopx:todo todo_id={todo_id} status=open task_class={task_class} {scope} -->\n"
+    text += "\n## Agent Todo\n- [ ] [P1] Continue after the archived dependency\n"
+    text += "  <!-- loopx:todo todo_id=todo_successor status=open task_class=advancement_task claimed_by=agent-a resume_when=todo_done:todo_dependency -->\n"
+    text += "\n## Completed Work Archive\n- [x] Prior result\n"
+    text += "  <!-- loopx:todo todo_id=todo_dependency status=done task_class=advancement_task -->\n"
+    state.write_text(text)
+    write_fixture_registry(project=tmp_path, runtime_root=runtime, registry_path=registry,
+        goal_id="scope-goal", domain="consumer-scope", adapter_kind="generic_project_goal_v0",
+        state_file=str(state), registered_agents=["agent-a", "agent-b"], quota_allowed_slots=None)
+    if provider != "legacy":
+        from loopx.control_plane.coordination.runtime_shadow import build_runtime_shadow_source_snapshot
+        goal = json.loads(registry.read_text())["goals"][0]
+        projection, _ = build_runtime_shadow_source_snapshot(goal=goal, runtime_root=runtime,
+            state_path=state, registry_path=registry)
+        initialize_canonical_authority(runtime, "scope-goal", projection, state_path=state, provider=provider)
+        state.unlink()
+    before = state.read_bytes() if state.exists() else None
+    def read(*args):
+        code, result = run_json_cli_result("todo", "list", "--goal-id", "scope-goal", *args,
+            registry_path=registry, runtime_root=runtime)
+        assert code == 0, result
+        return result
+    unfiltered = read("--role", "user")
+    assert unfiltered["todo_count"] == 5
+    selected = read("--role", "user", "--agent-id", "agent-a")
+    assert {row["todo_id"] for row in selected["todos"]} == {"todo_explicit_gate", "todo_explicit_action", "todo_global"}
+    limited = read("--role", "user", "--agent-id", "agent-a", "--limit", "1", "--thin")
+    assert limited["todo_list_projection"]["matched_todo_count"] == 3
+    assert limited["returned_todo_count"] == 1
+    for todo_id in ["todo_peer_gate", "todo_peer_action"]:
+        assert read("--todo-id", todo_id, "--agent-id", "agent-a")["not_found"] is True
+    successor = read("--todo-id", "todo_successor", "--agent-id", "agent-a")
+    assert successor["todo"]["resume_ready"] is True
+    assert (state.read_bytes() if state.exists() else None) == before
+
+
+def test_filter_composes_with_existing_lane_call_and_rejects_downgraded_response(monkeypatch):
+    from loopx.control_plane import effect_runtime
+    source = compact_todo_group([{"todo_id": "todo_one", "text": "Review result", "role": "user",
+        "status": "open", "task_class": "user_action", "bound_agent": "agent-a"}],
+        role="user", source_section="User Todo", item_limit=None)
+    original = effect_runtime.effect_runtime_result
+    requests = []
+    def track(method, request, **kwargs):
+        if method == "todo.summary_lanes.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"
+    def downgrade(method, request, **kwargs):
+        result = original(method, request, **kwargs)
+        if method == "todo.summary_lanes.project":
+            result.pop("source_indices", None)
+        return result
+    monkeypatch.setattr(effect_runtime, "effect_runtime_result", downgrade)
+    with pytest.raises(ValueError, match="selection ordinals"):
+        filtered_todo_summary(source, role="user", agent_id="agent-a")
+
+
+@pytest.mark.parametrize("corruption", ["duplicate", "boolean", "outside_selection"])
+def test_typed_lane_response_cannot_alias_or_escape_selected_source(monkeypatch, corruption):
+    from loopx.control_plane import effect_runtime
+    source = compact_todo_group([
+        {"todo_id": "todo_owned", "text": "Review result", "role": "user", "status": "open",
+            "task_class": "user_action", "bound_agent": "agent-a"},
+        {"todo_id": "todo_other", "text": "Review peer result", "role": "user", "status": "open",
+            "task_class": "user_action", "bound_agent": "agent-b"}],
+        role="user", source_section="User Todo", item_limit=None)
+    outside = next(index for index, item in enumerate(source["items"]) if item["todo_id"] == "todo_other")
+    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]
+        return result
+    monkeypatch.setattr(effect_runtime, "effect_runtime_result", corrupt)
+    with pytest.raises(ValueError, match="source ordinal|escaped the selected source"):
+        filtered_todo_summary(source, role="user", agent_id="agent-a")
diff --git a/tests/control_plane_ts/authority_store_conformance.ts b/tests/control_plane_ts/authority_store_conformance.ts
index 74617074d1..6e15c62d3b 100644
--- a/tests/control_plane_ts/authority_store_conformance.ts
+++ b/tests/control_plane_ts/authority_store_conformance.ts
@@ -1,3 +1,4 @@
+import {registerTodoConsumerScopeConformance} from "./todo_consumer_scope_conformance.ts";
 import {registerUserCompletionFollowthroughConformance} from "./user_completion_followthrough_conformance.ts";
 import {registerSuccessionReadConformance} from "./succession_read_conformance.ts";
 import {registerUserCompletionUpdateConformance} from "./user_completion_update_conformance.ts";
@@ -247,6 +248,7 @@ export function registerAuthorityStoreConformance(
   registerAuthorityScanConformance(providerName, factory);
   registerOwnershipObservationConformance(providerName, factory);
   registerSuccessionReadConformance(providerName, factory);
+  registerTodoConsumerScopeConformance(providerName, factory);
   registerNativePlanningUpdateConformance(providerName, factory);
   registerUserCompletionUpdateConformance(providerName, factory);
   registerUserCompletionFollowthroughConformance(providerName, factory);
diff --git a/tests/control_plane_ts/production_scale_coordination_fixture.ts b/tests/control_plane_ts/production_scale_coordination_fixture.ts
index ea532c8eeb..70122a88fa 100644
--- a/tests/control_plane_ts/production_scale_coordination_fixture.ts
+++ b/tests/control_plane_ts/production_scale_coordination_fixture.ts
@@ -614,3 +614,20 @@ export function productionScaleSuccessionFixture(goalId: string, schema: Authori
   readModel.records_sha256 = canonicalAuthoritySha256(todos);
   return {projection, cases};
 }
+
+/** Retained User addressing in the complete graph, including legacy claims.
+ * Selection must not derive completeness from the short display population. */
+export function productionScaleConsumerScopeFixture(goalId: string, schema: AuthorityProjectionSchema = "native") {
+  const fixture = productionScaleSuccessionFixture(goalId, schema);
+  const extra = [
+    {todo_id: "todo_scope_peer_gate", task_class: "user_gate", claimed_by: "agent-b"},
+    {todo_id: "todo_scope_peer_action", task_class: "user_action", claimed_by: "agent-b"},
+    {todo_id: "todo_scope_explicit_gate", task_class: "user_gate", claimed_by: "agent-b", blocks_agent: "agent-a"},
+    {todo_id: "todo_scope_explicit_action", task_class: "user_action", claimed_by: "agent-b", bound_agent: "agent-a"},
+    {todo_id: "todo_scope_global", task_class: "user_gate", claimed_by: "agent-b", global_gate: true},
+  ].map((item, index) => ({role: "user", status: "open", done: false, archive_state: "active",
+    text: "Review a synthetic result", source_section: "User Todo", index: 2000 + index, ...item}));
+  return {...fixture, projection: authorityProjectionFixture(goalId,
+    [...fixture.projection.todos as Record<string, unknown>[], ...extra],
+    fixture.projection.leases as Record<string, unknown>[], schema, {handoff_mode: "legacy"})};
+}
diff --git a/tests/control_plane_ts/succession_read_conformance.ts b/tests/control_plane_ts/succession_read_conformance.ts
index 0daf18b45e..ae0dcda9ec 100644
--- a/tests/control_plane_ts/succession_read_conformance.ts
+++ b/tests/control_plane_ts/succession_read_conformance.ts
@@ -5,6 +5,8 @@ import type {AuthorityStoreConformanceFactory} from "./authority_store_conforman
 import {productionScaleSuccessionFixture} from "./production_scale_coordination_fixture.ts";
 import {validateCoordinationTodoReadModel} from "../../loopx/control_plane/coordination/coordination_projection.ts";
 
+const PYTHON = process.env.LOOPX_TEST_PYTHON ?? "python3";
+
 // Exercise the shipped Python consumer → typed policy, not a second test reducer.
 const CONSUMER = `
 import json, sys
@@ -29,7 +31,7 @@ export function registerSuccessionReadConformance(name: string, factory: Authori
       next_projection: projection, events: [], receipts: []})).status, "applied");
     const before = await store.loadAuthority();
     assert.equal(before.status, "loaded");
-    const child = spawnSync("python3", ["-c", CONSUMER], {encoding: "utf8", timeout: 90_000,
+    const child = spawnSync(PYTHON, ["-c", CONSUMER], {encoding: "utf8", timeout: 90_000,
       input: JSON.stringify({todos: before.head.todos, cases})});
     assert.equal(child.status, 0, child.stderr);
     const actual = JSON.parse(child.stdout);
diff --git a/tests/control_plane_ts/todo_consumer_scope.test.ts b/tests/control_plane_ts/todo_consumer_scope.test.ts
new file mode 100644
index 0000000000..01287c1666
--- /dev/null
+++ b/tests/control_plane_ts/todo_consumer_scope.test.ts
@@ -0,0 +1,57 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import {projectTodoSummaryLanes} from "../../loopx/control_plane/todos/summary_lanes.ts";
+import type {JsonObject} from "../../loopx/control_plane/effect_program.ts";
+
+const row = (todo_id: string, fields: JsonObject = {}): JsonObject => ({todo_id,
+  status: "open", done: false, task_class: "user_gate", 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, "", todo_id], claim: null, bound: null, blocks: null, global: false,
+  excluded: [], ...fields});
+const project = (rows: JsonObject[], selection: JsonObject = {}) => projectTodoSummaryLanes({
+  schema_version: "todo_summary_lanes_request_v1", rows, observed_at: 100,
+  selection: {role: "user", status: null, todo_id: null, agent_id: "agent-a", ...selection}});
+
+test("Agent read selection uses gate and action addressing, before lanes/counts", () => {
+  const rows = [row("todo_peer", {claim: "agent-b", claimed: true}),
+    row("todo_explicit", {claim: "agent-b", claimed: true, blocks: "agent-a"}),
+    row("todo_global", {claim: "agent-b", global: true, excluded: ["agent-a"]}),
+    row("todo_action_peer", {task_class: "user_action", claim: "agent-b"}),
+    row("todo_action_bound", {task_class: "user_action", claim: "agent-b", bound: "agent-a"}),
+    row("todo_legacy")];
+  const before = structuredClone(rows), result = project(rows);
+  assert.deepEqual(result.source_indices, [1, 2, 4, 5]);
+  assert.deepEqual((result.lanes as JsonObject).open_items, [1, 2, 4, 5]);
+  assert.equal((result.work_counts as JsonObject).open, 4);
+  assert.equal(result.full_selection, false);
+  assert.deepEqual(rows, before);
+  assert.deepEqual(project(rows, {agent_id: null}).source_indices, [0, 1, 2, 3, 4, 5]);
+});
+
+test("filters compose without renumbering source ordinals or losing full-source evaluation", () => {
+  const rows = [row("todo_done", {status: "done", done: true}),
+    row("todo_ready", {task_class: "advancement_task", claim: "agent-a", claimed: true,
+      has_resume: true, resume_ready: true, resume_evaluated: true}),
+    row("todo_blocked", {task_class: "blocker", status: "blocked"}),
+    row("todo_other", {task_class: "advancement_task", claim: "agent-b"}),
+    row("todo_excluded", {task_class: "advancement_task", excluded: ["agent-a"]})];
+  const result = project(rows, {role: "agent", status: "open", todo_id: "todo_ready"});
+  assert.deepEqual(result.source_indices, [1]);
+  assert.deepEqual((result.lanes as JsonObject).executable_items, [1]);
+  assert.equal((result.work_counts as JsonObject).advancement, 1);
+  assert.deepEqual(project(rows, {role: "agent", todo_id: "todo_other"}).source_indices, []);
+  assert.deepEqual(project(rows, {role: "agent", todo_id: "todo_excluded"}).source_indices, []);
+});
+
+test("unfiltered v1 preserves v0 lane algebra and rejects malformed selection", () => {
+  const rows = [row("todo_a"), row("todo_b", {status: "done", done: true})];
+  const v0 = projectTodoSummaryLanes({schema_version: "todo_summary_lanes_request_v0", rows, observed_at: 100});
+  const v1 = project(rows, {agent_id: null});
+  assert.deepEqual(v1.lanes, v0.lanes); assert.deepEqual(v1.work_counts, v0.work_counts);
+  assert.equal(v1.full_selection, true);
+  for (const selection of [{role: "other"}, {status: "finished"}, {agent_id: 1}, {todo_id: []}]) {
+    assert.throws(() => project(rows, selection));
+  }
+  assert.throws(() => project([row("todo_bad", {global: "true"})]));
+});
diff --git a/tests/control_plane_ts/todo_consumer_scope_conformance.ts b/tests/control_plane_ts/todo_consumer_scope_conformance.ts
new file mode 100644
index 0000000000..19e5754403
--- /dev/null
+++ b/tests/control_plane_ts/todo_consumer_scope_conformance.ts
@@ -0,0 +1,50 @@
+import assert from "node:assert/strict";
+import {spawnSync} from "node:child_process";
+import test from "node:test";
+import type {AuthorityStoreConformanceFactory} from "./authority_store_conformance.ts";
+import {productionScaleConsumerScopeFixture} from "./production_scale_coordination_fixture.ts";
+
+const PYTHON = process.env.LOOPX_TEST_PYTHON ?? "python3";
+
+// The actual Python read consumer still hosts rendering; policy runs in TS.
+const CONSUMER = `
+import json, sys
+from loopx.control_plane.coordination.local_authority import canonical_todo_summary_fields
+from loopx.control_plane.todos.goal_todo_projection import filtered_todo_summary
+from loopx.control_plane.todos.quota_summary import summarize_user_todos_for_quota
+p=json.load(sys.stdin)
+fields=canonical_todo_summary_fields(p['todos'])
+summary=fields['user_todos']
+selected=filtered_todo_summary(summary,role='user',agent_id='agent-a')
+limited=filtered_todo_summary(summary,role='user',agent_id='agent-a',item_limit=1)
+quota=summarize_user_todos_for_quota(summary,agent_identity={'agent_id':'agent-a'},filter_user_gate_blocks_agent=True)
+base=canonical_todo_summary_fields([row for row in p['todos'] if not row['todo_id'].startswith('todo_scope_')])['user_todos']
+base_quota=summarize_user_todos_for_quota(base,agent_identity={'agent_id':'agent-a'},filter_user_gate_blocks_agent=True)
+prefix=lambda values: sorted(row['todo_id'] for row in values if row['todo_id'].startswith('todo_scope_'))
+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']),
+ '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))
+`;
+export function registerTodoConsumerScopeConformance(name: string, factory: AuthorityStoreConformanceFactory): void {
+  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);
+    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");
+    if (before.status !== "loaded") return;
+    const child = spawnSync(PYTHON, ["-c", CONSUMER], {encoding: "utf8", timeout: 90_000,
+      input: JSON.stringify({todos: before.head.todos, cases})});
+    assert.equal(child.status, 0, child.stderr);
+    const result = JSON.parse(child.stdout);
+    const expected = ["todo_scope_explicit_action", "todo_scope_explicit_gate", "todo_scope_global"];
+    assert.deepEqual(result.selected, expected); assert.equal(result.quota_delta, 3);
+    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(await store.loadAuthority(), before, "read consumers must never mutate authority");
+  });
+}