perf(todos): limit replan reads to consumed views

docs/architecture/rfcs/monorepo-distribution-split-v0.md

@@ -3437,3 +3437,34 @@ boundaries. Workspace/meta-package publication, complete install/load/context
 closure, independent Standard packages, other host/platform adoption, matched
 model effectiveness and complete costs remain open. This synthetic task and
 recovery evidence establishes neither model benefit nor an installed-host upgrade.
+
+### 2026-11-12 — Replan consumes only required Todo views
+
+The replan writeback caller now requests its two role summaries and complete
+Agent terminal rows through the existing quota read transaction. The existing
+TypeScript display owner accepts an explicit subset; omitted selection retains
+all three ordinary quota views. User source-summary rows, unused Agent views
+and prepared Vision wait facts are not generated for this caller. Source
+references, complete supplied-source fallback and provider error diagnostics
+retain their existing owners. No storage, lease, scheduler or quota-debit rule
+changes.
+
+Against immutable base `0abd0dc27`, candidate code `46f844e9d` uses the same
+disposable mixed-state fixtures and six alternating pairs per workload on real
+File/default-SQLite. SQLite replan qualification medians at 320/3,500 extra
+Agent Todos are 93.2→83.3 ms and 612.3→572.0 ms; canonical-read plus
+qualification medians are 139.4→130.0 ms and 900.9→857.0 ms. Strict refresh
+dry-run rejection medians are 142.4→135.1 ms and 934.4→911.8 ms. At zero extra
+Todos that refresh median is 28.9→29.3 ms; retain this observation without a
+noise or causality claim. Up to 128 User rows are included. Normalized
+obligations and rejection details match; counting fewer RPCs alone is not the
+acceptance oracle. Seventy focused production-provider and transport checks
+pass, including complete terminal tails, stale/missing sources and ordinary
+display defaults.
+
+Preserve #8624's original negative SQLite qualification (83.0→91.2 ms and
+605.9→704.1 ms) at its original head and criteria. This new scoped result
+qualifies the consumed replan read cost; full CLI process, durable writeback,
+model and end-to-end outcome costs are not inferred from these timings. The
+complete minimal load/context closure, R1/M3, independent Standard packages,
+workspace/meta publication and installed-host adoption remain open.

docs/architecture/rfcs/monorepo-distribution-split-v0.zh-CN.md

@@ -2682,3 +2682,27 @@ package-smoke 工作流执行这套安装后验证。
 保留 D1-D7、原有里程碑、历史失败与验收边界。workspace/meta 发行、完整安装／
 加载／context 闭包、独立 Standard 包、其他宿主／平台采用、配对模型效果和完整
 成本仍开放。合成任务与恢复证据不证明模型收益，也不代表已升级当前宿主。
+
+### 2026-11-12 — Replan 只消费所需 Todo 视图
+
+Replan 写回调用方现在通过既有 quota 读取事务取得两个角色摘要及完整 Agent
+terminal 行。既有 TypeScript 显示 owner 接受显式视图子集；省略选择时仍保留普通
+quota 的全部三种视图。该调用方不再生成 User source-summary 行、未消费的 Agent
+视图及预建 Vision 等待事实。来源引用、完整内联来源回退及 provider 错误诊断仍由
+既有 owner 处理；存储、lease、scheduler 及 quota 扣额规则保持原语义。
+
+以不可变基线 `0abd0dc27` 对照候选代码 `46f844e9d`，复用同一隔离混合状态夹具，
+在真实 File／默认 SQLite 上每种负载执行六组交替配对。额外 320／3,500 条 Agent
+Todo 时，SQLite replan 资格检查中位数为 93.2→83.3 ms、612.3→572.0 ms；包含
+canonical 读取与资格检查的中位数为 139.4→130.0 ms、900.9→857.0 ms。严格 refresh
+dry-run 拒绝路径的中位数为 142.4→135.1 ms、934.4→911.8 ms。零条额外 Todo 时，
+该 refresh 中位数为 28.9→29.3 ms；保留这项观测，不推断噪声或因果。夹具包含
+最多 128 条 User 行。归一化后的义务及拒绝详情一致；减少 RPC 次数本身不能作为
+验收依据。70 项聚焦生产 provider 与传输检查通过，覆盖完整 terminal 尾部、
+过期／缺失来源及普通显示默认值。
+
+保留 #8624 原 head 及原标准下的 SQLite 负面资格结果（83.0→91.2 ms、
+605.9→704.1 ms）。这份新范围结果只证明所消费 replan 读取的成本资格；不能从
+这些计时推断完整 CLI 进程、持久写回、模型及端到端成果成本。完整最小加载／
+上下文闭包、R1/M3、独立 Standard 包、workspace／meta 发布及安装宿主采用仍未
+完成。

loopx/control_plane/effect_runtime.py

@@ -1611,12 +1611,14 @@ def effect_runtime_result(
     def validate_result(result: Any) -> Any:
         # A caller must explicitly request a bundle. A request scope or a small
         # source does not establish that it will consume additional views.
-        if set(params) == {"summary_display_views"} and (not isinstance(result, dict) or any(
-            not isinstance(result.get(view), list)
-            or any(not isinstance(row, dict) for row in result[view])
-            for view in ("summary", "planning", "terminal")
-        )):
-            raise RuntimeError("invalid TypeScript Todo source views")
+        if set(params) == {"summary_display_views"}:
+            requested = params["summary_display_views"].get("view_names", ("summary", "planning", "terminal"))
+            if not isinstance(result, dict) or any(
+                not isinstance(result.get(view), list)
+                or any(not isinstance(row, dict) for row in result[view])
+                for view in requested
+            ):
+                raise RuntimeError("invalid TypeScript Todo source views")
         return result
 
     key = None

loopx/control_plane/todos/context_projection.ts

@@ -27,15 +27,21 @@ export function projectTodoContextPage(input: JsonObject): JsonObject {
   if (input.summary_display_views !== undefined) {
     if (Object.keys(input).length !== 1) throw new EffectRuntimeRequestError("summary display views cannot be combined with other views");
     const source = requireJsonObject(input.summary_display_views, "summary display views");
-    if (Object.keys(source).some(key => key !== "source" && key !== "handoff_gates")) {
+    if (Object.keys(source).some(key => key !== "source" && key !== "handoff_gates" && key !== "view_names")) {
       throw new EffectRuntimeRequestError("invalid summary display views fields");
     }
     requireJsonObject(source.source, "summary source");
-    return {
-      summary: projectTodoSummaryDisplay(source),
-      planning: projectTodoSummaryDisplay({source: source.source, include_terminal: false}),
-      terminal: projectTodoSummaryDisplay({source: source.source, include_terminal: true}),
+    const views: Record<string, () => JsonObject[]> = {
+      summary: () => projectTodoSummaryDisplay(source),
+      planning: () => projectTodoSummaryDisplay({source: source.source, include_terminal: false}),
+      terminal: () => projectTodoSummaryDisplay({source: source.source, include_terminal: true}),
     };
+    const names = source.view_names === undefined ? Object.keys(views) : source.view_names;
+    if (!Array.isArray(names) || names.length === 0 || new Set(names).size !== names.length ||
+        names.some(name => typeof name !== "string" || !Object.hasOwn(views, name))) {
+      throw new EffectRuntimeRequestError("summary display views must name distinct supported views");
+    }
+    return Object.fromEntries(names.map(name => [name, views[name as string]()]));
   }
   if (input.summary_display !== undefined) {
     if (Object.keys(input).length !== 1) throw new EffectRuntimeRequestError("summary display cannot be combined with other views");

loopx/control_plane/todos/quota_selection.py

@@ -116,6 +116,7 @@ def project_quota_planning(
     resolve_capacity: bool = False, current_time: str | None = None,
     summary_kind: str | None = None,
     consume_display_views: bool = False, request_only: bool = False,
+    display_view_names: list[str] | None = None,
 ) -> dict[str, Any]:
     identity = agent_identity if isinstance(agent_identity, dict) else {}
     profile = identity.get("agent_profile")
@@ -195,6 +196,7 @@ def integer_text(raw: Any) -> str | None:
                    if not native_profile else {}),
                 "materialize_resume": reference is not None or request_only,
                 "consume_display_views": consume_display_views,
+                **({"display_view_names": display_view_names} if display_view_names is not None else {}),
             }} if summary_kind else {}),
             "frontier_deadline": value.get("frontier_deadline"),
             "route_items": build_todo_route_continuation_facts(value, handoff_gates=handoff_gates),

loopx/control_plane/todos/quota_selection.ts

@@ -501,8 +501,12 @@ export function projectTodoQuotaPlanning(value: unknown): JsonObject {
   if (request.summary_request !== undefined) {
     const wire = composeQuotaSummary(request, planning);
     const summary = requireJsonObject(request.summary_request, "summary request");
+    if (summary.display_view_names !== undefined && summary.consume_display_views !== true) {
+      throw new EffectRuntimeRequestError("display view names require consumed display views");
+    }
     if (summary.consume_display_views === true) {
-      const display = {source: summary.source, handoff_gates: (request.handoff_items as JsonObject[]).map(row => row.display)};
+      const display = {source: summary.source, handoff_gates: (request.handoff_items as JsonObject[]).map(row => row.display),
+        ...(summary.display_view_names === undefined ? {} : {view_names: summary.display_view_names})};
       wire.display_views = projectTodoContextPage({summary_display_views: display});
     }
     if (summary.consume_vision_wait_basis === true) {

loopx/control_plane/todos/quota_summary.py

@@ -508,6 +508,44 @@ def select_quota_todo_summaries(
             select_quota_todo_summary(agent_source, None, agent_identity=agent_identity))
 
 
+def select_quota_todo_summaries_with_terminal(
+    user_source: Any, agent_source: Any, *, agent_identity: dict[str, Any] | None,
+) -> tuple[dict[str, Any] | None, dict[str, Any] | None, list[dict[str, Any]]]:
+    """Read two summaries and complete Agent terminal rows in one transaction.
+
+    Reuse the typed quota and display owners while requesting only consumed
+    views. Ordinary quota retains its own full displays and prepared wait facts.
+    """
+    from .quota_selection import _decode_quota_summary_wire
+    from .goal_todo_projection import project_todo_read
+
+    sources = [(role, value) for role, value in (("user", user_source), ("agent", agent_source))
+               if isinstance(value, dict)]
+    requests = [project_quota_planning(value, all_open_items=[],
+        source_open_count=value.get("open_count"), agent_identity=agent_identity,
+        filter_user_gate_blocks_agent=role == "user", available_capabilities=None,
+        summary_kind="user", consume_display_views=role == "agent", request_only=True,
+        display_view_names=["terminal"] if role == "agent" else None)
+        for role, value in sources]
+    summaries: dict[str, dict[str, Any]] = {}
+    terminal: list[dict[str, Any]] = []
+    if requests:
+        result = project_todo_read({"kind": "quota_projection", "requests": requests})
+        wires = result.get("quota_wires")
+        if not isinstance(wires, list) or len(wires) != len(requests):
+            raise RuntimeError("invalid complete quota projection response")
+        for (role, source), wire in zip(sources, wires, strict=True):
+            summaries[role] = _decode_quota_summary_wire(wire, source)
+            if role == "agent":
+                views = wire.get("display_views")
+                if not isinstance(views, dict) or set(views) != {"terminal"}:
+                    raise RuntimeError("invalid consumed Agent terminal view")
+                terminal = views["terminal"]
+                if not isinstance(terminal, list) or any(not isinstance(row, dict) for row in terminal):
+                    raise RuntimeError("invalid Agent terminal source rows")
+    return summaries.get("user"), summaries.get("agent"), terminal
+
+
 def select_quota_todo_contexts(
     user_canonical: Any, user_asset: Any, agent_canonical: Any, agent_asset: Any,
     *, agent_identity: dict[str, Any] | None, available_capabilities: Any,

loopx/control_plane/work_items/semantic_replan_writeback.py

@@ -20,8 +20,7 @@
 )
 from ..todos.active_state_todo_parser import parse_active_state_todos
 from ..todos.quota_summary import (
-    select_planning_inventory_source_items,
-    select_quota_todo_summaries,
+    select_quota_todo_summaries_with_terminal,
 )
 from ..todos.succession_warning import todo_succession_gap_items
 from .autonomous_replan_ack import (
@@ -264,14 +263,9 @@ def qualify_replan_writeback(
     )
     raw_user_todos = todo_projection.get("user_todos")
     raw_agent_todos = todo_projection.get("agent_todos")
-    user_todos, agent_todos = select_quota_todo_summaries(
+    user_todos, agent_todos, agent_todo_source_items = select_quota_todo_summaries_with_terminal(
         raw_user_todos, raw_agent_todos, agent_identity=agent_identity,
     )
-    agent_todo_source_items = select_planning_inventory_source_items(
-        raw_agent_todos,
-        None,
-        include_terminal=True,
-    )
     agent_todo_completion_items = (
         [
             item