refactor(planning): close canonical Todo planning consumer paths

docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md

@@ -2489,9 +2489,19 @@ The next complete stage packages are:
    and reads against actual callers. Status/attention now joins `todo list` in
    reading canonical Todo summaries after promotion, without requiring the
    Markdown file. Missing providers fail closed and empty canonical collections
-   never revive legacy Todos. This is consumer progress, not promotion proof:
-   Turn, quota, planning, standing decisions, leases and monitor writeback still
-   need their own parity inventory. Read authority does not grant writeback.
+   never revive legacy Todos. Refresh recommendation, repair/replan qualification,
+   completion-validation accountability, Todo-add replan binding and guided-start
+   frontier now share that canonical source. A refresh reads one snapshot and
+   passes it through its decisions rather than rereading a changing provider or
+   Markdown at each gate. Provider failure aborts; an empty snapshot is not a
+   fallback signal. This is consumer progress, not promotion proof: Turn/quota,
+   standing decisions, leases, monitor writeback, shared-goal alignment and
+   amendment revision bases still need their own parity inventory. Read authority
+   does not grant writeback. Source/display independence is tested with the
+   shared production-scale fixture and real FileAuthorityStore; these reads do
+   not establish freshness/CAS for a later business commit or change provider
+   defaults. Next Action narrative remains independent of Todo authority.
+
    Lifecycle admission and the preauthorized terminal fence now share the TS
    owner across legacy writers and native terminal transactions; the replaced
    Python rules are removed without changing provider defaults or promotion.

docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md

@@ -1975,8 +1975,15 @@ backend、实时双向同步或按命令拆开的权威；晋升后不支持的
    terminal/successor/archive 路径。按实际 caller 盘点剩余公开 mutation 和 read。
    status/attention 现在与 `todo list` 一样，在 promotion 后读 canonical Todo summary，
    不要求 Markdown 文件存在；provider 缺失 fail closed，canonical 空集合不能复活旧
-   Todo。这是 consumer 进展，不是 promotion 证明：Turn、quota、planning、standing
-   decision、lease、monitor writeback 仍需各自的 parity 清单。读权威不授予写回能力。
+   Todo。Refresh 推荐、repair/replan 验收、completion-validation 问责、Todo-add replan
+   绑定和 guided-start frontier 现已复用该 canonical 来源。一次 refresh 读取一份快照，
+   传给各项决策，不在每个门禁重新读取变化中的 provider 或 Markdown；provider 故障
+   直接中止，空快照不是 fallback 信号。这是 consumer 进展，不是 promotion 证明：
+   Turn/quota、standing decision、lease、monitor writeback、shared-goal alignment 与
+   amendment revision basis 仍需各自 parity 清单。读权威不授予写回能力。共用的复杂
+   fixture 和真实 FileAuthorityStore 验证 source/display 独立性，但不证明后续业务
+   commit 的 freshness/CAS，也不改变 provider 默认值。Next Action 正文仍独立于 Todo 权威。
+
    Lifecycle 准入及预授权 terminal fence 现由 legacy writer 与 native terminal
    transaction 共用 TS owner；删除对应 Python 规则，不改变 provider 默认或 promotion。
    这不是完整 native 字段编辑：在 update 的字段、ownership、validation 和 monitor/resume

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

@@ -225,10 +225,18 @@ than extending these adapters field by field.
    TS owner and delete the replaced decisions in the same slice. Reuse the
    canonical Todo summary for reads: `todo list` and status/attention must not
    select stale Markdown or event Todos after promotion, even when the display
-   is missing or the canonical collection is empty. Audit Turn, quota, planning,
-   Dashboard and standing-decision consumers separately; fixing one does not
-   qualify all consumers. Prove real-entrypoint parity and unavailable-provider
-   rejection, not just transport snapshots.
+   is missing or the canonical collection is empty. Refresh now loads one
+   unbounded canonical Todo snapshot for recommendation, repair/replan
+   qualification and completion-validation accountability; Todo-add's replan
+   binding and guided-start's existing frontier use the same source adapter.
+   Their existing decision reducers remain owners: no second planning store or
+   permission rule is introduced. Legacy callers retain their parser contracts.
+   Audit Turn/quota, Dashboard, standing decisions, shared-goal alignment and
+   amendment revision bases separately; this closes the named planning callers,
+   not every consumer. Prove real-entrypoint parity and unavailable-provider
+   rejection, not just transport snapshots. Independently authored Next Action
+   remains narrative, not a Todo import. Missing display does not authorize
+   reconstructing narrative or weaken an accountable completion fence.
 2. **Make the display a recoverable one-way projection.** Reuse the canonical
    journal/outbox and Todo-section renderer. Keep human narrative, source
    revision, idempotent delivery and actionable pending repair. A failed render

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

@@ -174,8 +174,14 @@ crossing 一起折叠进完整事务，不能沿着 adapter 逐字段继续加
    字段编辑、monitor、lease、event caller，把规则迁入既有 TS owner，并在同一切片
    删除被替代的 decision。读取复用 canonical Todo summary：promotion 后，`todo list`
    和 status/attention 不得选择陈旧 Markdown/event Todo；投影缺失、canonical 集合为空
-   也不例外。另行审计 Turn、quota、planning、Dashboard、standing-decision consumer；
-   修好一条不等于全部合格。通过真实入口验证 parity 和 provider 故障拒绝，不能只比传输快照。
+   也不例外。Refresh 现在只读一次无截断 canonical Todo 快照，供推荐、repair/replan
+   验收和 completion-validation 问责共同使用；Todo-add 的 replan 绑定和 guided-start
+   的既有 frontier 也复用同一来源适配器。既有 decision reducer 仍是规则 owner，不增加
+   第二份 planning store 或权限规则；旧模式保留原 parser 合同。Turn/quota、Dashboard、
+   standing decision、shared-goal alignment 与 amendment revision basis 另行审计，
+   不能将这些具名调用链的闭合等同于全部 consumer 合格。通过真实入口验证 parity 和
+   provider 故障拒绝。独立维护的 Next Action 仍是正文，不导入 Todo；展示缺失不授权
+   重建丢失正文，也不能削弱完成验收门禁。
 2. **把展示闭合为可恢复的单向投影。** 复用 canonical journal/outbox 与 Todo-section
    renderer，保留人工叙述、来源 revision、幂等交付和可操作的 pending repair。
    渲染失败不能撤销已提交事务，也不能授权 Markdown fallback；恢复投影不能重跑业务操作。

loopx/bootstrap_command_pack.py

@@ -1527,6 +1527,7 @@ def rerun_start_goal(selected_agent_id: str) -> str:
             project_connection,
             resolved_goal_id=str(command_pack.get("goal_id") or ""),
             effective_agent_id=str(command_pack.get("agent_id") or "") or None,
+            runtime_root_arg=runtime_root_arg,
         )
         if isinstance(project_connection, dict)
         and not isinstance(identity_selection_gate, dict)

loopx/cli_commands/todo.py

@@ -4,6 +4,7 @@
 from collections.abc import Callable, Sequence
 from pathlib import Path
 
+from ..control_plane.coordination.local_authority import read_canonical_todo_fields_if_promoted
 from ..control_plane.todos.contract import (
     TODO_TASK_CLASS_ADVANCEMENT,
     normalize_todo_continuation_policy,
@@ -148,11 +149,16 @@ def _validated_replan_successor_obligation(
         )
     registry = load_registry(registry_path)
     runtime_root = resolve_runtime_root(registry, runtime_root_arg)
-    _, _, state_text, _ = resolve_todo_state(
-        registry_path=registry_path,
-        goal_id=args.goal_id,
-        **_todo_path_args(args),
+    todo_fields = read_canonical_todo_fields_if_promoted(
+        runtime_root=runtime_root, goal_id=args.goal_id,
     )
+    state_text = ""
+    if todo_fields is None:
+        _, _, state_text, _ = resolve_todo_state(
+            registry_path=registry_path,
+            goal_id=args.goal_id,
+            **_todo_path_args(args),
+        )
     existing_runs, _ = load_index(
         runtime_root / "goals" / args.goal_id / "runs" / "index.jsonl"
     )
@@ -176,6 +182,7 @@ def _validated_replan_successor_obligation(
         None,
     )
     obligation, _ = qualify_replan_writeback(
+        todo_fields=todo_fields,
         newest_first_runs=newest_first_runs,
         state_text=state_text,
         agent_id=args.claimed_by,

loopx/control_plane/coordination/local_authority.py

@@ -13,7 +13,7 @@
 from uuid import uuid4
 
 from ...agent_registry import registered_agent_ids_from_registry
-from ...state_refresh import now_local
+from ..runtime.time import now_local_iso as now_local
 from ..effect_runtime import effect_runtime_result
 from .coordination_state_contract import (
     TODO_CANONICAL_READ_RECORD_SCHEMA_VERSION,
@@ -233,6 +233,23 @@ def read_canonical_todos_if_promoted(
     return payload
 
 
+def read_canonical_todo_fields_if_promoted(
+    *, runtime_root: Path, goal_id: str,
+    rollout_events: list[dict[str, Any]] | None = None,
+) -> dict[str, Any] | None:
+    """Read one unbounded planning snapshot; None alone permits legacy parsing.
+
+    Empty canonical state is authoritative. Provider failures propagate; this
+    read neither repairs Markdown nor grants mutation/promotion authority.
+    Callers pass the same fields to all decisions in one planning operation.
+    """
+    canonical = read_canonical_todos_if_promoted(runtime_root=runtime_root, goal_id=goal_id)
+    return (
+        canonical_todo_summary_fields(canonical["todos"], rollout_events=rollout_events)
+        if canonical is not None else None
+    )
+
+
 def canonical_todo_summary_fields(
     todos: list[dict[str, Any]],
     *,

loopx/control_plane/goals/start_goal_todo_delta.py

@@ -15,6 +15,8 @@
 from pathlib import Path
 from typing import Any
 
+from ..coordination.local_authority import read_canonical_todo_fields_if_promoted
+from ...paths import resolve_runtime_root
 from ...control_plane.todos.active_state_todo_parser import parse_active_state_todos
 from ...control_plane.todos.contract import (
     TODO_TASK_CLASS_ADVANCEMENT,
@@ -28,11 +30,12 @@
 _FRONTIER_PROJECTION_LIMIT = 1
 
 
 def existing_runnable_agent_frontier(
     inspection: Mapping[str, Any],
     *,
     resolved_goal_id: str,
     effective_agent_id: str | None,
+    runtime_root_arg: str | None = None,
 ) -> list[dict[str, Any]] | None:
     """Runnable advancement agent Todos already present in the goal's state.
 
@@ -41,9 +44,11 @@
     whose resume condition is satisfied (or absent) enter the frontier.
     Blocked, deferred, monitor, blocker, resume-blocked, or peer-claimed
     Todos never enter the frontier. Returns ``None`` whenever the frontier
-    cannot be proven (not connected, unknown goal, missing or unreadable
-    state file, or nothing runnable), so callers keep the unconditional
-    planning contract — fail-closed.
+    cannot be proven (not connected, unknown goal, missing legacy state, or
+    nothing runnable), so callers keep the unconditional planning contract.
+    After cutover only canonical records count; provider unavailability raises
+    instead of being mistaken for an empty frontier. A missing display is safe
+    to ignore, not permission to reconstruct its non-Todo narrative.
     """
     if inspection.get("connection_state") != "connected":
         return None
@@ -63,18 +68,22 @@
         Path(str(inspection.get("project") or "")),
         registry_goal.get("state_file"),
     )
-    if state_file is None or not state_file.is_file():
-        return None
-    try:
-        state_text = state_file.read_text(encoding="utf-8")
-    except OSError:
-        return None
-    parsed = parse_active_state_todos(
-        state_text,
-        goal=registry_goal,
-        state_path=state_file,
-        item_limit=None,
+    parsed = read_canonical_todo_fields_if_promoted(
+        runtime_root=resolve_runtime_root(
+            registry_payload or {}, runtime_root_arg, registry_path=registry_path,
+        ),
+        goal_id=resolved_goal_id,
     )
+    if parsed is None:
+        if state_file is None or not state_file.is_file():
+            return None
+        try:
+            state_text = state_file.read_text(encoding="utf-8")
+        except OSError:
+            return None
+        parsed = parse_active_state_todos(
+            state_text, goal=registry_goal, state_path=state_file, item_limit=None,
+        )
     agent_summary = parsed.get("agent_todos") if isinstance(parsed, dict) else None
     items = (
         agent_summary.get("items", []) if isinstance(agent_summary, dict) else []
@@ -86,19 +95,12 @@
         and todo_item_is_actionable_open(item)
         and item.get("task_class") == TODO_TASK_CLASS_ADVANCEMENT
     ]
-    if effective_agent_id:
-        runnable = [
-            item
-            for item in runnable
-            if not item.get("claimed_by")
-            or str(item.get("claimed_by")) == effective_agent_id
-        ]
-    else:
-        runnable = [
-            item
-            for item in runnable
-            if not item.get("claimed_by")
-        ]
+    runnable = [
+        item
+        for item in runnable
+        if not item.get("claimed_by")
+        or (effective_agent_id and str(item.get("claimed_by")) == effective_agent_id)
+    ]
     return runnable or None
 
 

loopx/control_plane/todos/completion_validation_accountability.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from typing import Any
+
 from .active_state_todo_parser import parse_active_state_todos
 from .completion_validation_projection import pending_completion_validation_todo
 
@@ -11,10 +13,12 @@ def require_accountable_completion_validation(
     *,
     todo_id: str | None,
     agent_id: str | None,
+    todo_fields: dict[str, Any] | None = None,
 ) -> None:
     """Reject accountable evidence while its exact validation Todo is open."""
 
-    summary = parse_active_state_todos(state_text, item_limit=None).get("agent_todos")
+    fields = todo_fields if todo_fields is not None else parse_active_state_todos(state_text, item_limit=None)
+    summary = fields.get("agent_todos")
     pending = pending_completion_validation_todo(
         summary,
         todo_id=todo_id,

loopx/control_plane/work_items/refresh_recommendation.py

@@ -5,11 +5,13 @@
 from typing import Any
 
 from ..agents.agent_lane_recommendation import build_agent_lane_next_action
+from ..coordination.local_authority import read_canonical_todo_fields_if_promoted
 from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result
 from ..todos.active_state_todo_parser import parse_active_state_todos
 from ..todos.contract import normalize_todo_id
 from ...feedback import validate_local_control_text
 from ...state_projection import active_state_next_action_entries
+from ...rollout_event_log import load_rollout_events, rollout_event_log_path
 
 REFRESH_RECOMMENDATION_REQUEST_SCHEMA_VERSION = "refresh_recommendation_request_v0"
 REFRESH_RECOMMENDATION_SCHEMA_VERSION = "refresh_recommendation_v0"
@@ -24,6 +26,31 @@
 RECOMMENDED_ACTION_SOURCE_DEFAULT = "default_refresh_action"
 
 
+def load_refresh_planning_source(
+    runtime_root: Path,
+    goal_id: str,
+    state_path: Path,
+    *,
+    require_display: bool,
+) -> tuple[str, list[dict[str, Any]], dict[str, Any] | None]:
+    """Read one shared planning snapshot without repairing its display.
+
+    Canonical Todo availability permits observation without Markdown, not an
+    edit of missing Next Action narrative. Provider failures propagate.
+    """
+    events = load_rollout_events(rollout_event_log_path(runtime_root, goal_id))
+    fields = read_canonical_todo_fields_if_promoted(
+        runtime_root=runtime_root, goal_id=goal_id, rollout_events=events,
+    )
+    try:
+        text = state_path.read_text(encoding="utf-8")
+    except FileNotFoundError:
+        if fields is None or require_display:
+            raise FileNotFoundError(f"state file does not exist: {state_path}") from None
+        text = ""
+    return text, events, fields
+
+
 def _first_valid_action(values: list[str]) -> str | None:
     for value in values:
         try:
@@ -41,9 +68,10 @@ def _agent_todo_summary(
     state_path: Path | None,
     settlement_todo_id: str | None,
     rollout_events: list[dict[str, Any]] | None,
+    todo_fields: dict[str, Any] | None,
 ) -> dict[str, Any] | None:
     preferred = {settlement_todo_id} if settlement_todo_id else None
-    parsed = parse_active_state_todos(
+    parsed = todo_fields if todo_fields is not None else parse_active_state_todos(
         state_text,
         goal=registry_goal,
         state_path=state_path,
@@ -95,6 +123,7 @@ def resolve_refresh_recommendation(
     registry_goal: dict[str, Any] | None = None,
     state_path: Path | None = None,
     rollout_events: list[dict[str, Any]] | None = None,
+    todo_fields: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     """Adapt canonical Todo facts into the TS-owned refresh read reducer."""
 
@@ -121,6 +150,7 @@ def resolve_refresh_recommendation(
             state_path=state_path,
             settlement_todo_id=settlement_todo_id,
             rollout_events=rollout_events,
+            todo_fields=todo_fields,
         )
         lane_candidate = build_agent_lane_next_action(
             agent_identity={"agent_id": agent_id} if agent_id else None,