diff --git a/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-28-retirement-cadence.md b/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-28-retirement-cadence.md
index d68049fe09..f475cdc148 100644
--- a/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-28-retirement-cadence.md
+++ b/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.
diff --git a/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-28-retirement-cadence.zh-CN.md b/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-28-retirement-cadence.zh-CN.md
index 7a1538b9fd..d04cd7811b 100644
--- a/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-28-retirement-cadence.zh-CN.md
+++ b/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。
diff --git a/loopx/contract.py b/loopx/contract.py
index beacc9bd31..77ca55e3e5 100644
--- a/loopx/contract.py
+++ b/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:
diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index 10b65d6484..7e0db6ec04 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/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(
diff --git a/loopx/control_plane/todos/authoring_scope.py b/loopx/control_plane/todos/authoring_scope.py
index 3faa08dc0b..01c01e7b5e 100644
--- a/loopx/control_plane/todos/authoring_scope.py
+++ b/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}
diff --git a/loopx/control_plane/todos/authoring_scope.ts b/loopx/control_plane/todos/authoring_scope.ts
index 4753bb95a9..17dfcb2777 100644
--- a/loopx/control_plane/todos/authoring_scope.ts
+++ b/loopx/control_plane/todos/authoring_scope.ts
@@ -3,6 +3,8 @@
 import type { JsonObject } from "../effect_program.ts";
 import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts";
 import { requireJsonObject } from "../runtime_decode.ts";
+import { AuthorityStoreProtocolError } from "../coordination/authority_store_codec.ts";
+import { removedTodoContinuationPolicy } from "./legacy_continuation_policy_migration.ts";
 import { normalizeRegisteredTodoAgents, normalizeTodoAgent, stripPythonWhitespace } from "../coordination/todo_agents.ts";
 import {
   normalizeTodoResumeWhen,
@@ -12,8 +14,11 @@ import {
 
 export const TODO_AUTHORING_SCOPE_REQUEST_SCHEMA = "todo_authoring_scope_request_v0";
 export const TODO_AUTHORING_SCOPE_RESULT_SCHEMA = "todo_authoring_scope_result_v0";
+export const TODO_CONTRACT_DIAGNOSTICS_REQUEST_SCHEMA = "todo_contract_diagnostics_request_v0";
+export const TODO_CONTRACT_DIAGNOSTICS_RESULT_SCHEMA = "todo_contract_diagnostics_result_v0";
 export const USER_TODO_TASK_CLASSES: ReadonlySet<string> = new Set(["user_action", "user_gate"]);
 export const AGENT_TODO_TASK_CLASSES: ReadonlySet<string> = new Set(["advancement_task", "continuous_monitor", "blocker"]);
+const TODO_STATUSES: ReadonlySet<string> = new Set(["open", "done", "blocked", "deferred"]);
 export const TODO_OWNERSHIP_INTENT_FIELDS = ["claimed_by", "clear_claim", "excluded_agents"] as const;
 
 /** Normalize explicit execution-owner intent before binding its replay identity.
@@ -44,15 +49,26 @@ function string(value: unknown, field: string): string | null {
   return value;
 }
 
-function requireTaskClass(role: string, taskClass: string | null, blocks: unknown, global: unknown): void {
+type TodoClassViolation = "task_class_missing" | "user_action_scope_invalid" | "task_class_role_mismatch";
+
+/** Read-only class rule shared by authoring and canonical diagnostics.
+ * It reports the violation instead of repairing the row or applying defaults. */
+function todoClassViolation(role: string, taskClass: string | null, blocks: unknown, global: unknown): TodoClassViolation | null {
   if (role !== "user") {
-    if (USER_TODO_TASK_CLASSES.has(taskClass ?? "")) fail("user_action and user_gate task_class are only valid for --role user");
-    return;
+    return USER_TODO_TASK_CLASSES.has(taskClass ?? "") ? "task_class_role_mismatch" : null;
   }
   const normalized = stripPythonWhitespace(taskClass ?? "");
-  if (!USER_TODO_TASK_CLASSES.has(normalized)) fail("user todo requires explicit --task-class user_gate or user_action; " +
+  if (!USER_TODO_TASK_CLASSES.has(normalized)) return "task_class_missing";
+  if (normalized === "user_action" && (blocks || global)) return "user_action_scope_invalid";
+  return null;
+}
+
+function requireTaskClass(role: string, taskClass: string | null, blocks: unknown, global: unknown): void {
+  const violation = todoClassViolation(role, taskClass, blocks, global);
+  if (violation === "task_class_role_mismatch") fail("user_action and user_gate task_class are only valid for --role user");
+  if (violation === "task_class_missing") fail("user todo requires explicit --task-class user_gate or user_action; " +
     "use user_gate for blocking owner/controller decisions and user_action for non-blocking user-visible todos");
-  if (normalized === "user_action" && (blocks || global)) fail("user_action is non-blocking and cannot set blocks_agent or global_gate; " +
+  if (violation === "user_action_scope_invalid") fail("user_action is non-blocking and cannot set blocks_agent or global_gate; " +
     "use --task-class user_gate for blocking decisions");
 }
 
@@ -66,6 +82,149 @@ interface Scope {
 export type UserTodoScopeConflict = "binding_conflict" | "gate_scope_conflict" |
   "global_binding_conflict" | "agent_binding_conflict" | "gate_scope_missing" | "binding_missing";
 
+export type TodoContractDiagnosticCode = "user_todo_task_class_missing" | "user_action_blocking_scope_invalid" |
+  "user_gate_scope_conflict" | "user_todo_response_scope_conflict" | "goal_user_gate_agent_binding_invalid" |
+  "agent_user_gate_goal_binding_invalid" | "agent_user_gate_response_binding_mismatch" |
+  "user_todo_bound_agent_unregistered" | "multi_agent_user_todo_missing_response_scope" |
+  "user_gate_blocks_unregistered_agent" | "multi_agent_user_gate_missing_scope" |
+  "todo_status_invalid" | "todo_excluded_agents_invalid" | "agent_todo_removed_continuation_policy" |
+  "agent_todo_blocks_agent_invalid" | "todo_claimed_by_excluded_agent" |
+  "todo_executor_exclusion_scope_invalid" | "todo_excludes_unregistered_agent";
+
+/** Bounded, target-neutral explanations. They never quote row text. */
+const TODO_CONTRACT_DIAGNOSTIC_DETAILS: Record<TodoContractDiagnosticCode, string> = {
+  todo_status_invalid: "todo status must be open, done, blocked, or deferred",
+  todo_excluded_agents_invalid: "todo excluded_agents must contain public-safe agent ids",
+  agent_todo_removed_continuation_policy: "agent todo uses a removed continuation policy; repair it explicitly before claiming",
+  agent_todo_blocks_agent_invalid: "blocks_agent is reserved for user gates; agent executor constraints use excluded_agents",
+  todo_claimed_by_excluded_agent: "todo cannot claim an agent that is also excluded",
+  todo_executor_exclusion_scope_invalid: "executor exclusions are only valid for agent todos",
+  todo_excludes_unregistered_agent: "agent todo excludes an agent that is not registered for this goal",
+  user_todo_task_class_missing: "open user todo requires task_class=user_gate or task_class=user_action",
+  user_action_blocking_scope_invalid: "open user_action todo is non-blocking and cannot set blocks_agent or global_gate",
+  user_gate_scope_conflict: "open user_gate todo cannot set both blocks_agent and global_gate",
+  user_todo_response_scope_conflict: "open user todo cannot set both bound_agent and goal_bound",
+  goal_user_gate_agent_binding_invalid: "goal-wide user_gate todo cannot bind its continuation to one agent",
+  agent_user_gate_goal_binding_invalid: "agent-scoped user_gate todo cannot set goal_bound",
+  agent_user_gate_response_binding_mismatch: "agent-scoped user_gate todo must bind to the same agent named by blocks_agent",
+  user_todo_bound_agent_unregistered: "open user todo binds an agent that is not registered for this goal",
+  multi_agent_user_todo_missing_response_scope: "open user todo in a multi-agent goal requires bound_agent or goal_bound",
+  user_gate_blocks_unregistered_agent: "open user_gate todo blocks an agent that is not registered for this goal",
+  multi_agent_user_gate_missing_scope: "open user_gate todo in a multi-agent goal requires blocks_agent or global_gate",
+};
+
+function optionalText(value: unknown): string | null {
+  return typeof value === "string" ? stripPythonWhitespace(value) || null : null;
+}
+
+type TodoOwnershipViolation = "claim_excluded" | "exclusion_role";
+
+function todoOwnershipViolations(role: string, claim: unknown, exclusions: unknown): TodoOwnershipViolation[] {
+  if (!Array.isArray(exclusions)) return [];
+  const violations: TodoOwnershipViolation[] = [];
+  if (claim && exclusions.includes(claim)) violations.push("claim_excluded");
+  if (role !== "agent" && exclusions.length) violations.push("exclusion_role");
+  return violations;
+}
+
+function optionalAgent(value: unknown): string | null {
+  if (!optionalText(value)) return null;
+  try { return normalizeTodoAgent(value, "Todo agent"); }
+  catch (error) {
+    if (error instanceof AuthorityStoreProtocolError) return null;
+    throw error;
+  }
+}
+
+/** Preserve the supported metadata health rules, independently of authorizing a write. */
+function todoMetadataDiagnostics(row: JsonObject, status: string, terminal: ReadonlySet<string>,
+  agents: readonly string[]): TodoContractDiagnosticCode[] {
+  const codes: TodoContractDiagnosticCode[] = [];
+  if (!TODO_STATUSES.has(status)) codes.push("todo_status_invalid");
+  let exclusions: string[] = [];
+  try {
+    const raw = typeof row.excluded_agents === "string" ? row.excluded_agents.split(",") : row.excluded_agents;
+    exclusions = normalizeTodoOwnershipIntent({excluded_agents: raw}).excluded_agents as string[] ?? [];
+  } catch (error) {
+    if (!(error instanceof AuthorityStoreProtocolError || error instanceof EffectRuntimeRequestError)) throw error;
+    codes.push("todo_excluded_agents_invalid");
+  }
+  if (row.role === "agent" && removedTodoContinuationPolicy(row.removed_continuation_policy)) {
+    codes.push("agent_todo_removed_continuation_policy");
+  }
+  if (row.role === "agent" && optionalAgent(row.blocks_agent)) codes.push("agent_todo_blocks_agent_invalid");
+  const ownership = todoOwnershipViolations(String(row.role), optionalAgent(row.claimed_by), exclusions);
+  if (ownership.includes("claim_excluded")) codes.push("todo_claimed_by_excluded_agent");
+  const archivedTerminal = row.archive_state === "archive" && terminal.has(status);
+  if (ownership.includes("exclusion_role") && !archivedTerminal) codes.push("todo_executor_exclusion_scope_invalid");
+  if (row.role === "agent" && agents.length && exclusions.some(agent => !agents.includes(agent))) {
+    codes.push("todo_excludes_unregistered_agent");
+  }
+  return codes;
+}
+
+/** Check read-model scope with the same class/scope rules as authoring.
+ * Historical rows retain their existing implied continuation binding. */
+function userTodoContractDiagnostic(row: JsonObject, registeredAgents: readonly string[]): TodoContractDiagnosticCode | null {
+  const taskClass = optionalText(row.task_class);
+  const blocks = optionalAgent(row.blocks_agent);
+  const global = row.global_gate === true;
+  const bound = optionalAgent(row.bound_agent);
+  const goalBound = row.goal_bound === true;
+  const classViolation = todoClassViolation("user", taskClass, blocks, global);
+  if (classViolation === "task_class_missing") return "user_todo_task_class_missing";
+  if (classViolation === "user_action_scope_invalid") return "user_action_blocking_scope_invalid";
+  const effectiveBound = bound ?? (taskClass === "user_gate" ? blocks : null) ?? optionalAgent(row.claimed_by);
+  const effectiveGoalBound = goalBound || (taskClass === "user_gate" && global);
+  const conflict = userTodoScopeConflict(taskClass, {
+    bound_agent: effectiveBound, goal_bound: effectiveGoalBound,
+    blocks_agent: blocks, global_gate: global,
+  }, registeredAgents.length);
+  // Retain existing diagnostic precedence and codes for persisted scope conflicts.
+  if (conflict === "gate_scope_conflict" || (taskClass === "user_gate" && blocks && global)) return "user_gate_scope_conflict";
+  if (bound && goalBound) return "user_todo_response_scope_conflict";
+  if (conflict === "global_binding_conflict" || (conflict === "binding_conflict" && global)) return "goal_user_gate_agent_binding_invalid";
+  if (conflict === "agent_binding_conflict" || (conflict === "binding_conflict" && blocks)) {
+    return goalBound ? "agent_user_gate_goal_binding_invalid" : "agent_user_gate_response_binding_mismatch";
+  }
+  if (bound && registeredAgents.length > 0 && !registeredAgents.includes(bound)) return "user_todo_bound_agent_unregistered";
+  if (conflict === "binding_missing" || (conflict === "gate_scope_missing" && !effectiveBound && !effectiveGoalBound)) {
+    return "multi_agent_user_todo_missing_response_scope";
+  }
+  if (taskClass === "user_gate" && blocks && registeredAgents.length > 0
+    && !registeredAgents.includes(blocks)) return "user_gate_blocks_unregistered_agent";
+  if (conflict === "gate_scope_missing") return "multi_agent_user_gate_missing_scope";
+  return null;
+}
+
+/** Evaluate canonical read-model rows without reading Markdown or writing state. */
+export function evaluateTodoContractDiagnostics(value: unknown): JsonObject {
+  const request = requireJsonObject(value, "Todo contract diagnostics request");
+  if (request.schema_version !== TODO_CONTRACT_DIAGNOSTICS_REQUEST_SCHEMA) {
+    fail("Todo contract diagnostics schema mismatch");
+  }
+  if (!Array.isArray(request.todos)) fail("todos must be an array");
+  if (!Array.isArray(request.registered_agents)) fail("registered_agents must be an array");
+  if (!Array.isArray(request.terminal_statuses)) fail("terminal_statuses must be an array");
+  const agents = normalizeRegisteredTodoAgents(request.registered_agents);
+  const terminal = new Set(request.terminal_statuses.map(
+    entry => stripPythonWhitespace(String(entry)).toLowerCase()));
+  const diagnostics: JsonObject[] = [];
+  let checked = 0;
+  for (const raw of request.todos) {
+    const row = requireJsonObject(raw, "Todo contract row");
+    const status = stripPythonWhitespace(String(row.status ?? "")).toLowerCase();
+    const codes = todoMetadataDiagnostics(row, status, terminal, agents);
+    if (row.role === "user" && !terminal.has(status)) {
+      checked += 1;
+      const code = userTodoContractDiagnostic(row, agents);
+      if (code) codes.push(code);
+    }
+    for (const code of codes) diagnostics.push({todo_id: optionalText(row.todo_id), code, detail: TODO_CONTRACT_DIAGNOSTIC_DETAILS[code]});
+  }
+  return {schema_version: TODO_CONTRACT_DIAGNOSTICS_RESULT_SCHEMA, checked, diagnostics};
+}
+
 /** Check a fully resolved scope without inventing any authoring defaults.
  * Both public edits and materialized terminal successors consume this rule. */
 export function userTodoScopeConflict(taskClass: string | null, scope: Scope,
@@ -175,19 +334,18 @@ export function planTodoAuthoringScope(value: unknown): JsonObject {
   if (!Array.isArray(registeredAgents)) fail("registered_agents must be an array");
   const agents = normalizeRegisteredTodoAgents(registeredAgents);
   const status = stripPythonWhitespace(string(intent.status, "status") ?? "").toLowerCase() || string(todo.status, "status") || "open";
-  if (!["open", "done", "blocked", "deferred"].includes(status)) fail("todo status must be one of: open, done, blocked, deferred");
+  if (!TODO_STATUSES.has(status)) fail("todo status must be one of: open, done, blocked, deferred");
   if (command === "create" && status === "done") fail("todo add cannot create completed work; add it open and use `loopx todo complete`");
   if (command === "update" && role === "agent" && intent.status && status === "done") fail("agent todo completion must use complete_goal_todo " +
     "(CLI: `loopx todo complete`) so completion policy, successor, and no-follow-up contracts are enforced");
   const scope = planScope(command ?? "", role, taskClass, todo, intent, agents, string(request.goal_id, "goal_id") ?? "");
   const exclusions = intent.excluded_agents ?? todo.excluded_agents;
-  if (TODO_OWNERSHIP_INTENT_FIELDS.some(field => intent[field] != null && intent[field] !== false)) {
-    const claim = intent.clear_claim ? null : intent.claimed_by || todo.claimed_by;
-    if (claim && Array.isArray(exclusions) && exclusions.includes(claim)) {
-      fail("claimed_by cannot also appear in excluded_agents; clear or transfer the claim in the same update");
-    }
+  const ownership = todoOwnershipViolations(role, intent.clear_claim ? null : intent.claimed_by || todo.claimed_by, exclusions);
+  if (TODO_OWNERSHIP_INTENT_FIELDS.some(field => intent[field] != null && intent[field] !== false)
+    && ownership.includes("claim_excluded")) {
+    fail("claimed_by cannot also appear in excluded_agents; clear or transfer the claim in the same update");
   }
-  if (role !== "agent" && Array.isArray(exclusions) && exclusions.length) fail("excluded_agents is only valid for agent todos; clear exclusions before moving this todo to a user role");
+  if (ownership.includes("exclusion_role")) fail("excluded_agents is only valid for agent todos; clear exclusions before moving this todo to a user role");
   // Completed history remains repairable; it does not create an active gate.
   if (status !== "done") {
     requireTaskClass(role, taskClass, scope.blocks_agent, scope.global_gate);
diff --git a/loopx/control_plane/todos/legacy_continuation_policy_migration.ts b/loopx/control_plane/todos/legacy_continuation_policy_migration.ts
index 5f3ca117b2..9b408dbf38 100644
--- a/loopx/control_plane/todos/legacy_continuation_policy_migration.ts
+++ b/loopx/control_plane/todos/legacy_continuation_policy_migration.ts
@@ -13,15 +13,19 @@ function existingAgent(value: unknown): string | null {
   }
 }
 
+/** Recognize the retained legacy policy markers without authorizing a repair. */
+export function removedTodoContinuationPolicy(value: unknown): "primary_review" | "review_handoff" | null {
+  const policy = stripPythonWhitespace(String(value ?? "")).toLowerCase();
+  return policy === "primary_review" || policy === "review_handoff" ? policy : null;
+}
+
 export function validateLegacyContinuationPolicyRepair(
   block: JsonObject,
   intent: JsonObject,
   todoId: string,
 ): void {
-  const removed = stripPythonWhitespace(
-    String(block.removed_continuation_policy ?? ""),
-  ).toLowerCase();
-  if (removed !== "primary_review" && removed !== "review_handoff") return;
+  const removed = removedTodoContinuationPolicy(block.removed_continuation_policy);
+  if (removed === null) return;
   const prefix = `todo_id '${todoId}' uses removed continuation_policy=${removed}; `;
   if (intent.claim_only) {
     throw new EffectRuntimeRequestError(prefix + "repair it before claiming");
diff --git a/tests/control_plane/test_canonical_status_todos.py b/tests/control_plane/test_canonical_status_todos.py
index db5d5e419a..eeff448e66 100644
--- a/tests/control_plane/test_canonical_status_todos.py
+++ b/tests/control_plane/test_canonical_status_todos.py
@@ -8,8 +8,10 @@
 from pathlib import Path
 
 import pytest
-from canonical_authority_fixture import initialize_canonical_authority
+from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime
 
+from loopx.contract import check_contract
+from loopx.control_plane.effect_runtime import restart_effect_runtime
 from loopx.control_plane.coordination.local_authority import LocalCoordinationAuthorityUnavailable
 from loopx.control_plane.coordination.coordination_state_contract import (
     TODO_DOMAIN_READ_RECORD_SCHEMA_VERSION,
@@ -20,8 +22,41 @@
 from loopx.status import active_state_todo_fields
 
 
+@pytest.fixture(autouse=True)
+def isolated_runtime(tmp_path, monkeypatch):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    yield
+    restart_effect_runtime()
+
+
+@pytest.fixture(params=["file", "sqlite"])
+def provider(request):
+    return request.param
+
+
 @pytest.fixture(params=["legacy", "native"])
-def promoted_goal(tmp_path: Path, request):
+def record_format(request):
+    return request.param
+
+
+def _projection(goal_id, records, record_format):
+    projection = build_todo_runtime_shadow_projection(goal_id=goal_id, todos=records)
+    if record_format == "native":
+        for record in projection["todos"]:
+            record["schema_version"] = "todo_domain_record_v0"
+            record.pop("index")
+            record.pop("source_section")
+        projection["todo_read_model"] = {
+            "schema_version": TODO_DOMAIN_READ_RECORD_SCHEMA_VERSION,
+            "contract_fields": list(TODO_DOMAIN_RECORD_FIELDS),
+            "todo_count": len(projection["todos"]),
+            "records_sha256": hashlib.sha256(canonical_bytes(projection["todos"])).hexdigest(),
+        }
+    return projection
+
+
+@pytest.fixture
+def promoted_goal(tmp_path: Path, record_format, provider):
     state = tmp_path / "ACTIVE_GOAL_STATE.md"
     state.write_text(
         "# Goal\n\n## Next Action\n\nKeep the human narrative.\n\n"
@@ -54,19 +89,8 @@ def promoted_goal(tmp_path: Path, request):
         },
         "decision_outcome": "reject",
     }]
-    projection = build_todo_runtime_shadow_projection(goal_id=goal["id"], todos=records)
-    if request.param == "native":
-        for record in projection["todos"]:
-            record["schema_version"] = "todo_domain_record_v0"
-            record.pop("index")
-            record.pop("source_section")
-        projection["todo_read_model"] = {
-            "schema_version": TODO_DOMAIN_READ_RECORD_SCHEMA_VERSION,
-            "contract_fields": list(TODO_DOMAIN_RECORD_FIELDS),
-            "todo_count": len(projection["todos"]),
-            "records_sha256": hashlib.sha256(canonical_bytes(projection["todos"])).hexdigest(),
-        }
-    initialize_canonical_authority(runtime, goal["id"], projection, state_path=state)
+    projection = _projection(goal["id"], records, record_format)
+    initialize_canonical_authority(runtime, goal["id"], projection, state_path=state, provider=provider)
     return goal, runtime, state
 
 
@@ -98,9 +122,9 @@ def test_status_uses_provider_and_allows_native_monitor_writeback_without_displa
             assert fields["active_state_next_action"] == "Keep the human narrative."
 
 
-def test_unavailable_provider_does_not_restore_stale_display_tasks(promoted_goal):
+def test_unavailable_provider_does_not_restore_stale_display_tasks(promoted_goal, provider):
     goal, runtime, state = promoted_goal
-    (runtime / "authority" / "file-v0").rename(runtime / "unavailable-provider")
+    (runtime / "authority" / f"{provider}-v0").rename(runtime / "unavailable-provider")
     with pytest.raises(LocalCoordinationAuthorityUnavailable):
         active_state_todo_fields(goal, runtime_root=runtime)
     assert "todo_stale" in state.read_text()
@@ -135,9 +159,19 @@ def test_unpromoted_status_retains_markdown_without_starting_authority(tmp_path,
     assert fields["agent_todos"]["items"][0]["text"] == "Legacy work"
 
 
-def test_public_status_cli_reads_canonical_attention_without_markdown(promoted_goal, tmp_path):
+@pytest.mark.parametrize("display", ["missing", "invalid_utf8", "invalid_user_todo"])
+def test_public_status_cli_uses_canonical_contract_and_attention(promoted_goal, tmp_path, display):
     goal, runtime, state = promoted_goal
-    state.unlink()
+    if display == "missing":
+        state.unlink()
+    elif display == "invalid_utf8":
+        state.write_bytes(b"\xff")
+    else:
+        state.write_text(
+            "# Goal\n\n## User Todo\n\n- [ ] Obsolete display-only gate\n"
+            "  <!-- loopx:todo todo_id=todo_stale status=open -->\n",
+        )
+    original = state.read_bytes() if state.exists() else None
     registry = tmp_path / "registry.json"
     registry.write_text(json.dumps({
         "schema_version": 1, "common_runtime_root": str(runtime), "goals": [goal],
@@ -147,8 +181,145 @@ def test_public_status_cli_reads_canonical_attention_without_markdown(promoted_g
          "--format", "json", "status", "--goal-id", goal["id"]],
         capture_output=True, text=True, timeout=60,
     )
-    assert result.returncode == 0, result.stderr
     payload = json.loads(result.stdout)
+    assert result.returncode == 0, payload["contract"]
+    assert payload["contract"]["ok"] is True
     assert "todo_canonical" in json.dumps(payload["attention_queue"])
     assert "todo_stale" not in json.dumps(payload["attention_queue"])
+    assert (state.read_bytes() if state.exists() else None) == original
+
+
+def test_contract_reports_promoted_failure_without_legacy_rescue(promoted_goal, provider, tmp_path):
+    goal, runtime, state = promoted_goal
+    (runtime / "authority" / f"{provider}-v0").rename(runtime / "unavailable-provider")
+    registry = tmp_path / "registry.json"
+    registry.write_text(json.dumps({"schema_version": 1, "goals": [goal]}))
+    result = check_contract(
+        registry_path=registry, runtime_root_override=str(runtime), scan_roots=[], limit=3,
+        goal_id_filter=goal["id"], include_public_boundary_scan=False,
+    )
+    assert result["ok"] is False
+    assert result["goal_errors"][goal["id"]]
+    assert any("canonical Todo contract unavailable" in row["message"]
+               for row in result["error_diagnostics"])
+
+
+def test_contract_rejects_corrupt_canonical_read_model_despite_valid_display(tmp_path, provider):
+    state = tmp_path / "state.md"
+    state.write_text("# Goal\n\n## Agent Todo\n\n")
+    runtime = tmp_path / "runtime"
+    goal = {"id": "goal-a", "repo": str(tmp_path), "state_file": str(state),
+            "domain": "software", "adapter": {"kind": "read_only_project_map_v0"}}
+    projection = build_todo_runtime_shadow_projection(goal_id="goal-a", todos=[])
+    projection["todo_read_model"]["records_sha256"] = "0" * 64
+    initialize_canonical_authority(runtime, "goal-a", projection, state_path=state, provider=provider)
+    registry = tmp_path / "registry.json"
+    registry.write_text(json.dumps({"schema_version": 1, "goals": [goal]}))
+    result = check_contract(
+        registry_path=registry, runtime_root_override=str(runtime), scan_roots=[], limit=3,
+        goal_id_filter="goal-a", include_public_boundary_scan=False,
+    )
+    assert result["ok"] is False
+    assert any("read-model digest mismatch" in row["message"]
+               for row in result["error_diagnostics"])
+
+
+@pytest.mark.parametrize(("fields", "agents", "expected_code"), [
+    ({}, ["agent-a"], "user_todo_task_class_missing"),
+    ({"task_class": "user_action", "global_gate": True}, ["agent-a"], "user_action_blocking_scope_invalid"),
+    ({"task_class": "user_gate", "global_gate": True, "blocks_agent": "agent-a"},
+     ["agent-a"], "user_gate_scope_conflict"),
+    ({"task_class": " user_gate ", "global_gate": True, "blocks_agent": "agent-a"},
+     ["agent-a"], "user_gate_scope_conflict"),
+    ({"task_class": "user_action", "bound_agent": "agent-a", "goal_bound": True},
+     ["agent-a"], "user_todo_response_scope_conflict"),
+    ({"task_class": "user_gate", "global_gate": True, "bound_agent": "agent-a"},
+     ["agent-a"], "goal_user_gate_agent_binding_invalid"),
+    ({"task_class": "user_gate", "blocks_agent": "agent-a", "goal_bound": True},
+     ["agent-a"], "agent_user_gate_goal_binding_invalid"),
+    ({"task_class": "user_gate", "blocks_agent": "agent-a", "bound_agent": "agent-b"},
+     ["agent-a", "agent-b"], "agent_user_gate_response_binding_mismatch"),
+    ({"task_class": "user_action", "bound_agent": "agent-other"}, ["agent-a"], "user_todo_bound_agent_unregistered"),
+    ({"task_class": "user_gate", "blocks_agent": "agent-other"},
+     ["agent-a"], "user_gate_blocks_unregistered_agent"),
+    ({"task_class": "user_gate", "goal_bound": True}, ["agent-a", "agent-b"], "multi_agent_user_gate_missing_scope"),
+    ({"task_class": "user_action"}, ["agent-a", "agent-b"], "multi_agent_user_todo_missing_response_scope"),
+    ({"role": "agent", "blocks_agent": "agent-a"}, ["agent-a"], "agent_todo_blocks_agent_invalid"),
+    ({"role": "agent", "claimed_by": "agent-a", "excluded_agents": ["agent-a"]},
+     ["agent-a"], "todo_claimed_by_excluded_agent"),
+    ({"role": "agent", "excluded_agents": ["agent-other"]}, ["agent-a"], "todo_excludes_unregistered_agent"),
+    ({"role": "agent", "removed_continuation_policy": "review_handoff"},
+     ["agent-a"], "agent_todo_removed_continuation_policy"),
+    ({"role": "agent", "excluded_agents": ["!"]}, ["agent-a"], "todo_excluded_agents_invalid"),
+    ({"task_class": "user_action", "excluded_agents": ["agent-a"]},
+     ["agent-a"], "todo_executor_exclusion_scope_invalid"),
+    ({"role": "agent", "claimed_by": "agent-a", "excluded_agents": ["agent-b"]},
+     ["agent-a", "agent-b"], None),
+    ({"status": "done", "done": True, "archive_state": "archive", "excluded_agents": ["agent-a"]},
+     ["agent-a"], None),
+    ({"status": "done", "done": True, "excluded_agents": ["agent-a"]},
+     ["agent-a"], "todo_executor_exclusion_scope_invalid"),
+    ({"task_class": "user_action"}, ["agent-a"], None),
+    ({"task_class": "user_gate", "global_gate": True}, ["agent-a", "agent-b"], None),
+    ({"task_class": "user_gate", "blocks_agent": "agent-a"}, ["agent-a", "agent-b"], None),
+    ({"status": "done", "done": True}, ["agent-a", "agent-b"], None),
+    ({"status": "deferred", "done": True, "resume_when": "capacity_available:network"}, ["agent-a", "agent-b"], None),
+])
+def test_canonical_todo_semantics_without_display(tmp_path, provider, record_format, fields, agents, expected_code):
+    state = tmp_path / "state.md"
+    state.write_text("# Goal\n\n## User Todo\n\n")
+    runtime = tmp_path / "runtime"
+    goal = {"id": "goal-a", "repo": str(tmp_path), "state_file": str(state),
+            "domain": "software", "adapter": {"kind": "read_only_project_map_v0"},
+            "coordination": {"registered_agents": agents}}
+    records = [{"schema_version": "todo_item_v0", "todo_id": "todo_checked", "index": 1,
+                "role": "user", "status": "open", "done": False, "text": "Canonical work",
+                "priority": "P1", "archive_state": "active", "source_section": "User Todo", **fields}]
+    initialize_canonical_authority(runtime, goal["id"], _projection(goal["id"], records, record_format),
+                                   state_path=state, provider=provider)
+    state.unlink()  # Canonical semantics must not need the display to exist.
+    registry = tmp_path / "registry.json"
+    registry.write_text(json.dumps({"schema_version": 1, "common_runtime_root": str(runtime), "goals": [goal]}))
+    result = check_contract(registry_path=registry, runtime_root_override=str(runtime), scan_roots=[], limit=3,
+                            goal_id_filter=goal["id"], include_public_boundary_scan=False)
+    assert result["ok"] is (expected_code is None), result["error_diagnostics"]
+    assert [row["code"] for row in result["error_diagnostics"]] == ([expected_code] if expected_code else [])
+    assert all(row["goal_id"] == goal["id"] for row in result["error_diagnostics"])
+    process = subprocess.run([sys.executable, "-m", "loopx.cli", "--registry", str(registry),
+                              "--format", "json", "status", "--goal-id", goal["id"]],
+                             capture_output=True, text=True, timeout=60)
+    assert process.returncode == (1 if expected_code else 0), process.stdout + process.stderr
+    assert json.loads(process.stdout)["contract"]["ok"] is (expected_code is None)
     assert not state.exists()
+
+
+@pytest.mark.parametrize("fields", [{"role": "agent", "status": "bogus"},
+                                  {"role": "user", "task_class": "user_action", "status": "bogus"}])
+def test_legacy_canonical_status_is_validated_without_display(tmp_path, provider, fields):
+    test_canonical_todo_semantics_without_display(
+        tmp_path, provider, "legacy", fields, ["agent-a"], "todo_status_invalid",
+    )
+
+
+def test_large_canonical_user_narratives_do_not_enter_diagnostics_rpc(tmp_path, provider, record_format):
+    state = tmp_path / "state.md"
+    state.write_text("# Goal\n\n## User Todo\n\n")
+    runtime = tmp_path / "runtime"
+    goal = {"id": "goal-a", "repo": str(tmp_path), "state_file": str(state),
+            "domain": "software", "adapter": {"kind": "read_only_project_map_v0"}}
+    records = [{"schema_version": "todo_item_v0", "todo_id": f"todo_user_{index:04d}", "index": index + 1,
+                "role": "user", "status": "open", "done": False, "task_class": "user_action",
+                "text": "Synthetic user narrative. " * 100, "priority": "P1", "archive_state": "active",
+                "source_section": "User Todo"} for index in range(1100)]
+    assert len(json.dumps(records).encode()) > 2 * 1024 * 1024
+    initialize_canonical_authority(runtime, goal["id"], _projection(goal["id"], records, record_format),
+                                   state_path=state, provider=provider)
+    registry = tmp_path / "registry.json"
+    registry.write_text(json.dumps({"schema_version": 1, "common_runtime_root": str(runtime), "goals": [goal]}))
+    result = subprocess.run([sys.executable, "-m", "loopx.cli", "--registry", str(registry),
+                             "--format", "json", "status", "--goal-id", goal["id"]],
+                            capture_output=True, text=True, timeout=60)
+    assert result.returncode == 0, result.stdout + result.stderr
+    payload = json.loads(result.stdout)
+    assert payload["contract"]["ok"] is True
+    assert state.read_text() == "# Goal\n\n## User Todo\n\n"
diff --git a/tests/control_plane_ts/todo_authoring_scope.test.ts b/tests/control_plane_ts/todo_authoring_scope.test.ts
index e914fc9396..4b112dcddd 100644
--- a/tests/control_plane_ts/todo_authoring_scope.test.ts
+++ b/tests/control_plane_ts/todo_authoring_scope.test.ts
@@ -1,7 +1,9 @@
 import assert from "node:assert/strict";
 import test from "node:test";
+import { validateLegacyContinuationPolicyRepair } from "../../loopx/control_plane/todos/legacy_continuation_policy_migration.ts";
 import type { JsonObject } from "../../loopx/control_plane/effect_program.ts";
 import { planTodoAuthoringScope, TODO_AUTHORING_SCOPE_REQUEST_SCHEMA,
+  evaluateTodoContractDiagnostics, TODO_CONTRACT_DIAGNOSTICS_REQUEST_SCHEMA,
   userTodoScopeConflict } from "../../loopx/control_plane/todos/authoring_scope.ts";
 
 function plan(intent: JsonObject, overrides: JsonObject = {}): JsonObject {
@@ -106,3 +108,57 @@ test("malformed intent cannot turn a truthy string or an unknown field into scop
     assert.throws(() => plan({task_class: "user_gate", actor_agent_id: "agent-a", ...intent}), /boolean|does not own/);
   }
 });
+
+test("canonical diagnostics keep metadata and non-terminal user rules without repairing rows", () => {
+  const evaluate = (todos: JsonObject[], agents = ["agent-a"], terminal = ["done", "deferred", "archived"]) =>
+    evaluateTodoContractDiagnostics({schema_version: TODO_CONTRACT_DIAGNOSTICS_REQUEST_SCHEMA,
+      todos, registered_agents: agents, terminal_statuses: terminal});
+  const open = (extra: JsonObject): JsonObject => ({schema_version: "todo_item_v0", todo_id: "todo_x",
+    role: "user", status: "open", done: false, text: "raw text that must not surface", ...extra});
+  const codes = (todos: JsonObject[], agents = ["agent-a"], terminal = ["done", "deferred", "archived"]) =>
+    (evaluate(todos, agents, terminal).diagnostics as JsonObject[]).map(row => row.code);
+  assert.deepEqual(codes([open({})]), ["user_todo_task_class_missing"]);
+  assert.deepEqual(codes([open({done: true})]), ["user_todo_task_class_missing"]);
+  assert.deepEqual(codes([open({task_class: "user_action", global_gate: true})]), ["user_action_blocking_scope_invalid"]);
+  assert.deepEqual(codes([open({task_class: "user_gate", global_gate: true, blocks_agent: "agent-a"})]),
+    ["user_gate_scope_conflict"]);
+  assert.deepEqual(codes([open({task_class: "user_gate", goal_bound: true, bound_agent: "agent-a"})]),
+    ["user_todo_response_scope_conflict"]);
+  assert.deepEqual(codes([open({task_class: "user_gate", global_gate: true, bound_agent: "agent-a"})]),
+    ["goal_user_gate_agent_binding_invalid"]);
+  assert.deepEqual(codes([open({task_class: "user_gate", blocks_agent: "agent-a", goal_bound: true})]),
+    ["agent_user_gate_goal_binding_invalid"]);
+  assert.deepEqual(codes([open({task_class: "user_gate", goal_bound: true})], ["agent-a", "agent-b"]),
+    ["multi_agent_user_gate_missing_scope"]);
+  assert.deepEqual(codes([open({task_class: "user_action"})], ["agent-a", "agent-b"]),
+    ["multi_agent_user_todo_missing_response_scope"]);
+  const agent = (extra: JsonObject): JsonObject => ({...open({}), role: "agent", ...extra});
+  assert.deepEqual(codes([agent({blocks_agent: "agent-a"})]), ["agent_todo_blocks_agent_invalid"]);
+  assert.deepEqual(codes([agent({claimed_by: "agent-a", excluded_agents: ["agent-a"]})]), ["todo_claimed_by_excluded_agent"]);
+  assert.deepEqual(codes([agent({excluded_agents: ["agent-other"]})]), ["todo_excludes_unregistered_agent"]);
+  assert.deepEqual(codes([agent({removed_continuation_policy: "review_handoff"})]), ["agent_todo_removed_continuation_policy"]);
+  assert.deepEqual(codes([agent({excluded_agents: ["!"]})]), ["todo_excluded_agents_invalid"]);
+  assert.deepEqual(codes([agent({status: "bogus"})]), ["todo_status_invalid"]);
+  assert.deepEqual(codes([open({task_class: "user_action", excluded_agents: ["agent-a"]})]), ["todo_executor_exclusion_scope_invalid"]);
+  assert.deepEqual(codes([open({status: "done", archive_state: "archive", excluded_agents: ["agent-a"]})]), []);
+  assert.deepEqual(codes([agent({claimed_by: "agent-a", excluded_agents: ["agent-b"]})], ["agent-a", "agent-b"]), []);
+  // Controls: valid open user work, completed history without a class, and agent rows stay healthy.
+  assert.deepEqual(codes([open({task_class: "user_action"})]), []);
+  assert.deepEqual(codes([{...open({}), status: "done", done: true}]), []);
+  assert.deepEqual(codes([{schema_version: "todo_item_v0", todo_id: "todo_a", role: "agent",
+    status: "open", done: false}]), []);
+  const observed = evaluate([open({}), {...open({}), status: "done", done: true}]);
+  assert.equal(observed.checked, 1);
+  assert.equal(JSON.stringify(observed).includes("raw text"), false);
+});
+
+test("shared removed-policy classification preserves explicit migration authority", () => {
+  for (const policy of ["primary_review", " review_handoff "]) {
+    const source = {removed_continuation_policy: policy};
+    assert.throws(() => validateLegacyContinuationPolicyRepair(source, {claim_only: true}, "todo_x"), /repair it before claiming/);
+    assert.throws(() => validateLegacyContinuationPolicyRepair(source, {}, "todo_x"), /repair it explicitly/);
+    assert.doesNotThrow(() => validateLegacyContinuationPolicyRepair(source,
+      {continuation_policy: "independent_handoff", excluded_agents: ["agent-a"]}, "todo_x"));
+  }
+  assert.doesNotThrow(() => validateLegacyContinuationPolicyRepair({}, {claim_only: true}, "todo_x"));
+});