fix(status): use canonical authority for promoted Todo health

docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-28-retirement-cadence.md

@@ -197,10 +197,50 @@ ten-day soak. No active authority, release default or legacy-writer deletion
 decision changes. B still needs sustained workload/platform/capacity evidence;
 C still needs consumer/onboarding and supported upgrade acceptance.
 
-The next B slice is the remaining whole-command cold path: profile history
-artifact lookup, active-contract validation and public-boundary scanning on
-the same retained inputs before selecting the owning repair. Separate provider
-head-read time from caller work; preserve freshness, full decision inputs and
-corruption rejection. Re-run installed CLI consumers after integration. Do not
-count this read optimization as closing A/C or use a fixed remaining-PR estimate;
-retire a writer only with its last supported caller and recovery acceptance.
+### Contract health follows Todo authority
+
+#5222 is merged and locally adopted after backup, CLI/App/service upgrade and
+actual page readback. Default quota output is about 93 KB versus 1.37 MB with
+full detail, with equal Todo counts; 13 previous-delivery static resources match
+byte for byte. This is adoption evidence, not a new formal release, provider
+default switch or completed D2 soak.
+
+An isolated public CLI counterexample found that the Todo list reads canonical
+state while contract health still parses Markdown Todos. Adding only a stale
+User Todo without task_class to the display copy makes a healthy File or SQLite
+Goal fail status with exit code 1. The repair routes promoted contract checks
+through the existing TS canonical snapshot/record validator and shared User Todo
+class/scope rules and supported Todo metadata health; Python transports bounded
+semantic fields and adapts the diagnostic. Agent routing, claim/exclusion
+conflicts, removed policies and legacy status errors remain unhealthy. Structural validity alone does not make an open User Todo healthy.
+Missing providers and corrupt read models remain Goal-scoped errors,
+with no Markdown fallback. Unpromoted Goals retain legacy checks; invalid UTF-8 yields a structured read
+error while still rejecting the command. Narrative,
+registry, history and public-boundary checks remain. This does not introduce or
+replace Todo authoring validation, nor reauthorize completed/deferred history.
+Real File/SQLite controls cover both persisted record shapes, absent display,
+invalid active class/scope and Agent metadata, valid implied historical bindings,
+legal executor exclusions and completed/archived records without a class. Narrative text stays outside the diagnostic RPC; large
+collections are transported in bounded batches without changing message limits.
+
+A paired isolated contract-only measurement uses the 1,109-Todo current
+projection from retained history and an approximately 7 MB display file. Three
+warm samples for the initial structural-only repair fell from 0.52–0.58 seconds
+to 0.11–0.12 seconds for File and 0.14–0.16 seconds for SQLite. These timings
+precede the active User Todo semantic correction and do not qualify its cost.
+The experiment reinitializes the current projection;
+it is not full history replay, whole-status latency or cross-platform capacity
+qualification. Private inputs remain outside Git.
+
+Scale characterization with 4,101 synthetic Agent Todos still hits the existing
+`todo.succession.project` RPC response budget in whole `status` on both base and
+repair for File/SQLite. The repaired contract API can read that collection;
+this does not qualify the remaining whole-command payload boundary.
+
+The next B work remains history artifact lookup and remaining public payload/
+cold-path costs, preserving file-change freshness, full decision inputs and
+corruption rejection. Contract checks and attention still read canonical state
+separately; this repair adds no cross-request cache and claims no command-wide
+consistent snapshot. Recheck installed consumers after integration; A/C and D2
+retain their own open acceptance. Retire each writer only after its last
+supported caller and recovery acceptance are qualified.

docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-28-retirement-cadence.zh-CN.md

@@ -154,8 +154,36 @@ Quota 观察复用既有 should-run 摘要：捕获的单 Goal 行序列化由 1
 authority、发布默认值或旧 writer 删除决定。B 仍缺持续负载／平台／容量证据；C 仍需
 consumer／新建入口及受支持升级验收。
 
-B 的下一段是剩余整命令冷路径：在相同保留输入上分别分析历史 artifact 查找、
-active-contract 验证和公共边界扫描，再选择所属 owner 修复。区分 provider head
-读取与调用方工作，保留 freshness、完整决策输入和损坏拒绝；集成后重跑安装态 CLI
-消费者。不能把读取优化计为 A/C 完成，也不继续给固定的剩余 PR 数；只有最后受支持
-调用方退出且恢复验收通过，才能删除对应 writer。
+### 合同健康检查的权威归属
+
+#5222 已合并并完成本机备份、CLI／App／服务升级及实际页面读回。默认 quota
+响应约 93 KB，显式全明细约 1.37 MB，Todo 计数相同；上一份交付的 13 个静态资源
+字节一致。这是采用证据，不代表新一轮正式 release、provider 默认切换或 D2 完成。
+
+后续公共 CLI 的隔离反例表明：Todo 列表已读 canonical provider，但合同健康检查
+仍解析旧 Markdown Todo。仅在展示副本增加一条缺少 task_class 的旧 User Todo，
+File 和 SQLite 的正常 Goal 均被判为不健康，status 退出码变成 1。
+修复让晋升后的合同检查复用既有 TS canonical 快照／记录校验和 User Todo class／scope
+规则及既有 Todo 元数据健康约束；Python 只分批传输必要语义字段并适配诊断。
+Agent 路由、认领／排除冲突、废弃策略与旧格式非法状态仍判为不健康。结构有效不等于未完成 User Todo 健康。
+provider 缺失或读模型损坏仍报 Goal 范围的错误，不回退 Markdown。未晋升 Goal
+保留旧格式检查；非法 UTF-8 改为结构化读取错误，命令仍拒绝。叙述、registry、
+历史及公共边界检查不因此取消。此处不新增或替代
+Todo 写入时的业务校验，也不重审完成／deferred 历史的授权。真实 File／SQLite
+对照覆盖两种持久记录格式、缺失展示副本、非法活跃 class／scope 与 Agent 元数据、合法历史隐式绑定、
+合法执行者排除及缺 class 的完成／归档记录。正文不进入诊断 RPC，大集合使用有界分批，不放宽消息上限。
+
+用保留历史所得的 1,109 个 Todo 当前 projection 和约 7 MB 展示文件，在隔离存储中
+配对测量初版仅检查结构的合同修复。三个热样本由 0.52–0.58 秒降为 File 的
+0.11–0.12 秒、SQLite 的 0.14–0.16 秒；这些数据早于活跃 User Todo 语义修正，
+不用于证明修正版本的成本。此实验重新初始化当前 projection，不是完整历史重放，也不是
+整个 status 延迟或跨平台容量验收。私有输入不入库。
+
+4,101 个合成 Agent Todo 的规模对照中，File／SQLite 的基线与修复版完整 `status`
+仍触及既有 `todo.succession.project` RPC 响应预算；修复后的合同 API 能读取该集合，
+不代表剩余整命令包体边界已完成验收。
+
+B 下一步仍是历史 artifact 查找和剩余公共包体／冷路径，保留文件变化 freshness、
+完整决策输入及损坏拒绝。合同检查与 attention 仍各自读取 canonical 快照；本次没有
+引入跨请求缓存或声称命令级一致快照。集成后继续核对安装态消费者，A/C 与 D2
+维持各自未完成项；只有最后受支持调用方退出且恢复验收通过，才能删除对应 writer。

loopx/contract.py

@@ -10,6 +10,10 @@
 from typing import Any
 
 from .agent_registry import registered_agent_ids_for_goal
+from .control_plane.coordination.local_authority import (
+    LocalCoordinationAuthorityUnavailable,
+    read_canonical_todos_if_promoted,
+)
 from .control_plane.goals.contract_health import (
     contract_error_diagnostic,
     contract_error_views,
@@ -26,6 +30,7 @@
 )
 from .control_plane.runtime.file_text_reads import iter_utf8_file_reads
 from .control_plane.todos.active_state_editing import COMPLETED_WORK_ARCHIVE_HEADING
+from .control_plane.todos.authoring_scope import todo_contract_diagnostics
 from .history import (
     RunHistoryAudit,
     build_run_history_audit,
@@ -430,9 +435,10 @@ def _index_duplicate_warning(
     return f"{safe_goal_id}: duplicate index rows raw={raw} unique={unique}{detail}; {action}"
 
 
-def _active_state_todo_contract_diagnostics(
+def _todo_contract_diagnostics(
     registry: dict[str, Any],
     *,
+    runtime_root: Path,
     goal_id_filter: str | None = None,
     activation_state_filter: GoalActivationState | str | None = None,
 ) -> tuple[list[dict[str, Any]], int]:
@@ -458,6 +464,36 @@ def add_error(code: str, message: str) -> None:
                 )
             )
 
+        # The durable fence selects the authority for diagnostics as well as
+        # Todo display. Reuse the TS read-model/record validator: the Markdown
+        # copy cannot invalidate or rescue a promoted collection.
+        try:
+            canonical = read_canonical_todos_if_promoted(
+                runtime_root=runtime_root, goal_id=goal_id,
+            )
+        except LocalCoordinationAuthorityUnavailable as exc:
+            add_error(exc.code, f"{goal_id}: canonical Todo contract unavailable: {exc}")
+            continue
+        if canonical is not None:
+            # Structural validity does not replace the shared Todo metadata
+            # and non-terminal User class/scope rules. Evaluate provider rows without reading display.
+            try:
+                canonical_diagnostics = todo_contract_diagnostics(
+                    todos=canonical["todos"],
+                    registered_agents=registered_agent_ids_for_goal(goal),
+                    terminal_statuses=TERMINAL_TODO_STATUSES,
+                )
+            except RuntimeError as exc:
+                add_error(
+                    "canonical_todo_contract_diagnostics_unavailable",
+                    f"{goal_id}: canonical Todo contract diagnostics unavailable: {exc}",
+                )
+                continue
+            checked += canonical_diagnostics["checked"]
+            for row in canonical_diagnostics["diagnostics"]:
+                add_error(row["code"], f"{goal_id}: canonical todo {row['todo_id']} {row['detail']}")
+            continue
+
         registered_agents = registered_agent_ids_for_goal(goal)
         repo_text = str(goal.get("repo") or "").strip()
         if not repo_text:
@@ -467,7 +503,7 @@ def add_error(code: str, message: str) -> None:
             continue
         try:
             lines = state_file.read_text(encoding="utf-8").splitlines()
-        except OSError as exc:
+        except (OSError, UnicodeError) as exc:
             add_error(
                 "active_state_read_failed",
                 f"{goal_id}: cannot read active state for todo contract check: {exc}",
@@ -753,7 +789,7 @@ def _active_state_projection_gap_warnings(
             continue
         try:
             state_text = state_file.read_text(encoding="utf-8")
-        except OSError:
+        except (OSError, UnicodeError):
             continue
         projection_gap = state_projection_gap_warning(state_text)
         if not projection_gap:
@@ -1025,9 +1061,15 @@ def add_global_error(code: str, message: str) -> None:
 
     if registry is None:
         registry = load_registry(registry_path)
+    runtime_root = resolve_runtime_root(
+        registry,
+        runtime_root_override,
+        registry_path=registry_path,
+    )
     todo_contract_diagnostics, checked_user_gates = (
-        _active_state_todo_contract_diagnostics(
+        _todo_contract_diagnostics(
             registry,
+            runtime_root=runtime_root,
             goal_id_filter=goal_id_filter,
             activation_state_filter=activation_state_filter,
         )
@@ -1043,11 +1085,6 @@ def add_global_error(code: str, message: str) -> None:
         )
     )
 
-    runtime_root = resolve_runtime_root(
-        registry,
-        runtime_root_override,
-        registry_path=registry_path,
-    )
     if runtime_root == DEFAULT_RUNTIME_ROOT or runtime_root.exists():
         checks.append(f"runtime root resolved: {runtime_root}")
     else:

loopx/control_plane/effect_runtime_handlers.ts

@@ -107,7 +107,7 @@ import { transitionTodoNextAction } from "./todos/next_action.ts";
 import { planTodoFieldUpdate } from "./todos/field_update.ts";
 import { planPublicTodoUpdate } from "./todos/public_update.ts";
 import { planMonitorMetadata } from "./todos/monitor_metadata.ts";
-import { planTodoAuthoringScope } from "./todos/authoring_scope.ts";
+import { evaluateTodoContractDiagnostics, planTodoAuthoringScope } from "./todos/authoring_scope.ts";
 import {
   evaluateTodoResumeConditions,
   normalizeTodoResumeWhen,
@@ -480,6 +480,7 @@ export function createEffectRuntimeHandlers(
     ["coordination.source.project", withCoordinationSourceTransfer("coordination.source.project", projectCoordinationSource)],
     ["todo.monitor_metadata.plan", planMonitorMetadata],
     ["todo.authoring_scope.plan", planTodoAuthoringScope],
+    ["todo.contract_diagnostics.evaluate", evaluateTodoContractDiagnostics],
     [
       "todo.claim.decide",
       (params) => evaluateCoordinationTodoClaimDecision(

loopx/control_plane/todos/authoring_scope.py

@@ -43,3 +43,31 @@ def require_user_todo_task_class(
     plan_todo_authoring_scope(command="class", role=role, intent={
         "task_class": task_class, "blocks_agent": blocks_agent, "global_gate": global_gate,
     }, registered_agents=[], goal_id="")
+
+
+def todo_contract_diagnostics(
+    *, todos: list[dict[str, Any]], registered_agents: list[str],
+    terminal_statuses: set[str] | frozenset[str],
+) -> dict[str, Any]:
+    """Read-only canonical Todo diagnostics; the shared TS owner keeps the rule."""
+    # Transport only bounded semantic facts, never narrative text.
+    fields = ("todo_id", "role", "status", "task_class", "blocks_agent",
+              "global_gate", "bound_agent", "goal_bound", "claimed_by",
+              "excluded_agents", "removed_continuation_policy", "archive_state")
+    rows = [{field: todo.get(field) for field in fields} for todo in todos]
+    diagnostics: list[dict[str, Any]] = []
+    checked = 0
+    for offset in range(0, len(rows), 512):
+        result = effect_runtime_result("todo.contract_diagnostics.evaluate", {
+            "schema_version": "todo_contract_diagnostics_request_v0",
+            "todos": rows[offset:offset + 512], "registered_agents": registered_agents,
+            "terminal_statuses": sorted(terminal_statuses),
+        })
+        if (not isinstance(result, dict)
+                or result.get("schema_version") != "todo_contract_diagnostics_result_v0"
+                or not isinstance(result.get("diagnostics"), list)
+                or not isinstance(result.get("checked"), int)):
+            raise RuntimeError("TypeScript Todo contract diagnostics result shape mismatch")
+        checked += result["checked"]
+        diagnostics.extend(result["diagnostics"])
+    return {"checked": checked, "diagnostics": diagnostics}