fix(quota): select authoritative Todo sources before projection

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

@@ -4,7 +4,7 @@
 - Supersedes / closes: none
 - Proposed by: LoopX maintainers
 - Date: 2026-08-15
-- Last revised: 2026-10-20
+- Last revised: 2026-10-21
 - Scope: an incremental, replacement-first migration of the LoopX control-plane
   core from Python to TypeScript without maintaining two semantic
   implementations
@@ -520,6 +520,21 @@ the read-only three-arm rehearsal as a separate promotion gate.
 - Checkpoint moved to the execution ledger: [Provider-neutral projection conformance checkpoint (2026-09-12)](ledger/typescript-control-plane-migration-v0/2026-09-12-provider-neutral-projection-conformance.md).
 - Checkpoint moved to the execution ledger: [Canonical Todo presentation checkpoint (2026-09-12)](ledger/typescript-control-plane-migration-v0/2026-09-12-canonical-todo-presentation.md).
 
+### Quota Todo source selection (2026-10-21)
+
+Quota source and planning-inventory reads select the preferred available summary
+before invoking the existing typed projection owners. Canonical/legacy precedence
+and an authoritative empty selection are preserved. Unselected rows no longer
+cross the runtime transport; malformed unselected rows cannot block a valid
+selected source. A selected-source projection failure still propagates without
+adopting the other source.
+
+The duplicated planning adapter and discarded projections are removed. Python
+retains compatibility source selection until its quota and planning callers move
+in-process. There is no new RPC, data cache or domain-rule implementation. This
+bounded caller change requires complete-command SQLite measurements and installed
+File/SQLite parity; fewer requests alone do not establish performance payoff.
+
 ### Long-goal persistence is part of the migration payoff
 
 The product target is at least ten elapsed days per goal, not a short-lived

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

@@ -4,7 +4,7 @@
 - 替代 / 关闭：无
 - Proposed by：LoopX maintainers
 - Date：2026-08-15
-- Last revised：2026-10-20
+- Last revised：2026-10-21
 - Scope：LoopX 控制面核心从 Python 到 TypeScript 的增量、replacement-first
   迁移；不长期维护两份语义实现
 - Tracking issue：[#3225](https://github.com/loopx-project/loopx/issues/3225)
@@ -395,6 +395,18 @@ promotion gate。
 - 检查点已移至执行账本：[Provider-neutral projection conformance 检查点（2026-09-12）](ledger/typescript-control-plane-migration-v0/2026-09-12-provider-neutral-projection-conformance.zh-CN.md)。
 - 检查点已移至执行账本：[Canonical Todo 展示检查点（2026-09-12）](ledger/typescript-control-plane-migration-v0/2026-09-12-canonical-todo-presentation.zh-CN.md)。
 
+### Quota Todo 来源选择（2026-10-21）
+
+Quota 来源读取与 planning-inventory 读取先选定优先且可用的 summary，
+再调用既有 typed 投影 owner。canonical/legacy 优先级与权威空选择保持不变。
+未选中的行不再跨 runtime transport；未选中行格式错误不会阻断有效的选中来源。
+选中来源的投影失败仍向上传播，不会采用另一来源。
+
+重复的 planning adapter 与被丢弃的投影已删除。Python 保留兼容来源选择，
+直到 quota 与 planning 调用者迁入同一进程。这里不新增 RPC、数据缓存或领域规则实现。
+这个有界调用者改动仍须验证完整命令的 SQLite 性能与已安装 File/SQLite 等价性；
+仅减少请求次数不能证明性能收益。
+
 ### 长程持久化也是迁移收益的一部分
 
 产品目标是单个 goal 至少持续十个自然日。shared-authority RFC 的

loopx/control_plane/todos/quota_summary.py

@@ -759,37 +759,24 @@ def select_quota_todo_summary(
     return summary
 
 
+def _select_todo_source(
+    canonical_value: Any,
+    project_asset_value: Any,
+) -> dict[str, Any] | None:
+    sources = (canonical_value, project_asset_value)
+    if not is_canonical_attention_todo_summary(canonical_value):
+        sources = tuple(reversed(sources))
+    return next((source for source in sources if isinstance(source, dict)), None)
+
+
 def select_quota_todo_source_items(
     canonical_value: Any,
     project_asset_value: Any,
 ) -> list[dict[str, Any]]:
     """Return the authoritative pre-hot-path todo source for semantic checks."""
 
-    canonical_items = (
-        todo_summary_source_items(canonical_value)
-        if isinstance(canonical_value, dict)
-        else None
-    )
-    project_asset_items = (
-        todo_summary_source_items(project_asset_value)
-        if isinstance(project_asset_value, dict)
-        else None
-    )
-    if is_canonical_attention_todo_summary(canonical_value):
-        return canonical_items if canonical_items is not None else project_asset_items or []
-    return project_asset_items if project_asset_items is not None else canonical_items or []
-
-
-def _planning_inventory_source_items(value: Any, *, include_terminal: bool = False) -> list[dict[str, Any]] | None:
-    """Return canonical Todo rows before presentation-lane expansion.
-
-    Planning consumers share domain-state rows instead of rebuilding a larger,
-    order-sensitive union from presentation lanes.
-    """
-
-    if not isinstance(value, dict):
-        return None
-    return todo_planning_source_items(value, include_terminal=include_terminal)
+    source = _select_todo_source(canonical_value, project_asset_value)
+    return todo_summary_source_items(source) if source is not None else []
 
 
 def select_planning_inventory_source_items(
@@ -800,11 +787,8 @@ def select_planning_inventory_source_items(
 ) -> list[dict[str, Any]]:
     """Select the canonical non-terminal rows shared by planning read models."""
 
-    canonical_items = _planning_inventory_source_items(canonical_value, include_terminal=include_terminal)
-    project_asset_items = _planning_inventory_source_items(project_asset_value, include_terminal=include_terminal)
-    if is_canonical_attention_todo_summary(canonical_value):
-        return canonical_items if canonical_items is not None else project_asset_items or []
-    return project_asset_items if project_asset_items is not None else canonical_items or []
+    source = _select_todo_source(canonical_value, project_asset_value)
+    return todo_planning_source_items(source, include_terminal=include_terminal) if source is not None else []
 
 
 def _task_orchestration_authority_items(

tests/control_plane/test_quota_summary_source_precedence.py

@@ -1,8 +1,10 @@
 from copy import deepcopy
+import json
 
 import pytest
 
 from loopx.control_plane.todos import quota_selection, quota_summary
+from loopx.control_plane import effect_runtime
 
 
 def source(todo_id, *, canonical=False, empty=False):
@@ -74,3 +76,81 @@ def unavailable(*_args, **_kwargs):
         quota_summary.select_quota_todo_summary(
             source("todo_primary", canonical=canonical), source("todo_fallback"),
         )
+
+
+@pytest.mark.parametrize("selector", [
+    quota_summary.select_quota_todo_source_items,
+    quota_summary.select_planning_inventory_source_items,
+])
+@pytest.mark.parametrize("canonical,project,expected,excluded", [
+    (source("todo_canonical", canonical=True), source("todo_project"), "todo_canonical", "todo_project"),
+    (source("todo_legacy"), source("todo_project"), "todo_project", "todo_legacy"),
+    (source("todo_empty", canonical=True, empty=True), source("todo_project"), None, "todo_project"),
+    (source("todo_legacy"), {}, None, "todo_legacy"),
+    (None, source("todo_project"), "todo_project", None),
+    (source("todo_legacy"), None, "todo_legacy", None),
+    (None, None, None, None),
+])
+def test_source_reads_preserve_precedence_without_sending_discarded_rows(
+    monkeypatch, selector, canonical, project, expected, excluded,
+):
+    before = deepcopy((canonical, project))
+    requests = []
+    original = effect_runtime._request_with_info
+
+    def record(*args, **kwargs):
+        requests.append(json.dumps(kwargs["params"]))
+        return original(*args, **kwargs)
+
+    monkeypatch.setattr(effect_runtime, "_request_with_info", record)
+    rows = selector(canonical, project)
+
+    assert [row["todo_id"] for row in rows] == ([expected] if expected else [])
+    if excluded:
+        assert all(excluded not in request for request in requests)
+    assert (canonical, project) == before
+
+
+@pytest.mark.parametrize("selector", [
+    quota_summary.select_quota_todo_source_items,
+    quota_summary.select_planning_inventory_source_items,
+])
+def test_selected_source_transport_failure_does_not_adopt_other_tasks(monkeypatch, selector):
+    def unavailable(*_args, **_kwargs):
+        raise RuntimeError("selected source transport unavailable")
+
+    monkeypatch.setattr(effect_runtime, "_request_with_info", unavailable)
+    with pytest.raises(RuntimeError, match="TypeScript Effect runtime request failed"):
+        selector(source("todo_primary", canonical=True), source("todo_other"))
+
+
+def test_terminal_planning_keeps_the_selected_source_causality():
+    canonical = source("todo_terminal", canonical=True)
+    canonical["items"][0].update(status="done", done=True, evidence="verified outcome")
+    project = source("todo_other")
+
+    assert quota_summary.select_planning_inventory_source_items(canonical, project) == []
+    terminal = quota_summary.select_planning_inventory_source_items(
+        canonical, project, include_terminal=True,
+    )
+    assert [row["todo_id"] for row in terminal] == ["todo_terminal"]
+    assert terminal[0]["evidence"] == "verified outcome"
+
+
+@pytest.mark.parametrize("malformed_selected", [False, True])
+def test_malformed_handoff_display_row_only_blocks_its_selected_source(malformed_selected):
+    canonical = source("todo_canonical", canonical=True)
+    project = source("todo_discarded")
+    malformed = canonical if malformed_selected else project
+    malformed["items"][0].update(
+        index="invalid-display-index", excluded_agents=["agent-b"],
+        unblocks_todo_id="todo_prior",
+    )
+
+    if malformed_selected:
+        with pytest.raises(ValueError):
+            quota_summary.select_quota_todo_source_items(canonical, project)
+    else:
+        rows = quota_summary.select_quota_todo_source_items(canonical, project)
+
+        assert [row["todo_id"] for row in rows] == ["todo_canonical"]