From 2f930acef0477717a60a2c28fe71260a4749ab17 Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Sun, 13 Sep 2026 07:03:38 +0800
Subject: [PATCH 1/3] refactor(coordination): own handoff mode in canonical
 transactions

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 loopx/cli_commands/handoff_mode.py            |  16 ++-
 .../coordination/authority_core.py            |  28 ++--
 .../coordination/handoff_mode_policy.ts       |  22 +++
 .../coordination/handoff_mode_runtime.ts      |  29 ++++
 .../coordination/handoff_mode_transaction.ts  |  96 +++++++++++++
 .../coordination/local_authority_runtime.ts   |   6 +-
 .../coordination/todo_lifecycle_decision.ts   |   3 +-
 .../control_plane/effect_runtime_handlers.ts  |   4 +
 loopx/control_plane/todos/handoff_mode.py     |  76 ++++++----
 .../todos/provider_handoff_mode.py            |  39 +++++
 .../test_canonical_handoff_mode.py            |  98 +++++++++++++
 .../test_shadow_fence_caller_parity_e2e.py    |   5 +
 .../authority_store_conformance.ts            |   2 +
 .../handoff_mode_conformance.ts               | 135 ++++++++++++++++++
 .../legacy_writer_fence_caller_parity_v0.json |  21 ++-
 15 files changed, 518 insertions(+), 62 deletions(-)
 create mode 100644 loopx/control_plane/coordination/handoff_mode_policy.ts
 create mode 100644 loopx/control_plane/coordination/handoff_mode_runtime.ts
 create mode 100644 loopx/control_plane/coordination/handoff_mode_transaction.ts
 create mode 100644 loopx/control_plane/todos/provider_handoff_mode.py
 create mode 100644 tests/control_plane/test_canonical_handoff_mode.py
 create mode 100644 tests/control_plane_ts/handoff_mode_conformance.ts

diff --git a/loopx/cli_commands/handoff_mode.py b/loopx/cli_commands/handoff_mode.py
index dca0050022..790f14fd54 100644
--- a/loopx/cli_commands/handoff_mode.py
+++ b/loopx/cli_commands/handoff_mode.py
@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from ..control_plane.coordination.local_authority import LocalCoordinationAuthorityUnavailable
+
 from ..control_plane.coordination.legacy_writer_fence import LegacyCoordinationWriterFenced
 from ..control_plane.coordination.shadow_management import ShadowManagementError
 
@@ -69,7 +71,7 @@ def register_handoff_mode_command(
     parser = subparsers.add_parser(
         "handoff-mode",
         help=(
-            "Show or set the per-goal handoff_mode front-matter field that "
+            "Show or set the authoritative per-goal handoff_mode that "
             "selects which ownership authority governs todo handoffs."
         ),
     )
@@ -94,6 +96,8 @@ def register_handoff_mode_command(
             "for ownership changes and makes the completion fence mandatory."
         ),
     )
+    parser.add_argument("--operation-id", help="Stable canonical set intent id for recovery after a lost response.")
+    parser.add_argument("--dry-run", action="store_true", help="Validate set without committing the mode or a receipt.")
     parser.add_argument("--project", help="Project root. Defaults to the registry goal repo.")
     parser.add_argument("--state-file", help="Active goal state path. Defaults to the registry goal state_file.")
 
@@ -114,11 +118,12 @@ def handle_handoff_mode_command(
     }
     try:
         if args.handoff_mode_command == "show":
-            if args.mode:
-                raise ValueError("handoff-mode show does not accept --mode")
+            if args.mode or args.operation_id or args.dry_run:
+                raise ValueError("handoff-mode show does not accept set options")
             payload = show_goal_handoff_mode(
                 registry_path=registry_path,
                 goal_id=args.goal_id,
+                runtime_root_arg=runtime_root_arg,
                 **path_args,
             )
         else:
@@ -129,17 +134,18 @@ def handle_handoff_mode_command(
                 goal_id=args.goal_id,
                 mode=args.mode,
                 runtime_root_arg=runtime_root_arg,
+                operation_id=args.operation_id, dry_run=args.dry_run,
                 **path_args,
             )
-    except (HandoffModeError, LegacyCoordinationWriterFenced, ShadowManagementError) as exc:
+    except (HandoffModeError, LegacyCoordinationWriterFenced, ShadowManagementError, LocalCoordinationAuthorityUnavailable) as exc:
         payload = {
+            **exc.payload,
             "ok": False,
             "schema_version": "goal_handoff_mode_v0",
             "action": getattr(args, "handoff_mode_command", None),
             "goal_id": args.goal_id,
             "error": str(exc),
             "error_code": exc.code,
-            **exc.payload,
         }
     except LockAcquireTimeoutError as exc:
         payload = {
diff --git a/loopx/control_plane/coordination/authority_core.py b/loopx/control_plane/coordination/authority_core.py
index 9a1e586e4c..664f90e60d 100644
--- a/loopx/control_plane/coordination/authority_core.py
+++ b/loopx/control_plane/coordination/authority_core.py
@@ -784,20 +784,20 @@ def _decide_handoff_transition(
     snapshot: CoordinationSnapshot,
     command: HandoffModeTransitionCommand,
 ) -> TransitionPlan:
-    if snapshot.handoff_mode is command.requested_mode:
-        return _result(
-            DecisionOutcome.NO_CHANGE,
-            "handoff_mode_unchanged",
-            next_snapshot=snapshot,
-            idempotent=True,
-        )
-    if snapshot.active_claimed_todo_ids or snapshot.active_lease_todo_ids:
-        return _result(DecisionOutcome.REJECTED, "handoff_mode_not_quiescent")
-    return _result(
-        DecisionOutcome.APPLY,
-        "handoff_mode_transition",
-        next_snapshot=replace(snapshot, handoff_mode=command.requested_mode),
-    )
+    result = effect_runtime_result("coordination.handoff_mode.plan", {
+        "schema_version": "loopx_handoff_mode_plan_request_v0",
+        "previous_mode": snapshot.handoff_mode.value,
+        "requested_mode": command.requested_mode.value,
+        "active_claimed_todo_ids": list(snapshot.active_claimed_todo_ids),
+        "active_lease_todo_ids": list(snapshot.active_lease_todo_ids),
+    })
+    if not isinstance(result, dict) or result.get("schema_version") != "loopx_handoff_mode_plan_result_v0":
+        raise RuntimeError("TypeScript handoff mode plan shape mismatch")
+    outcome = DecisionOutcome(result["outcome"])
+    return _result(outcome, str(result["code"]),
+        next_snapshot=(None if outcome is DecisionOutcome.REJECTED else
+                       replace(snapshot, handoff_mode=command.requested_mode)),
+        idempotent=result["idempotent"])
 
 
 def decide(
diff --git a/loopx/control_plane/coordination/handoff_mode_policy.ts b/loopx/control_plane/coordination/handoff_mode_policy.ts
new file mode 100644
index 0000000000..dd978bd61b
--- /dev/null
+++ b/loopx/control_plane/coordination/handoff_mode_policy.ts
@@ -0,0 +1,22 @@
+/** One quiescence rule for the legacy adapter and canonical mode transaction. */
+import type {JsonObject} from "../effect_program.ts";
+import {requireJsonObject, requireStringArray, requireStringLiteral} from "../runtime_decode.ts";
+
+export const HANDOFF_MODES = ["legacy", "soft_claim", "hard_lease"] as const;
+export type HandoffMode = typeof HANDOFF_MODES[number];
+export const HANDOFF_MODE_PLAN_SCHEMA = "loopx_handoff_mode_plan_request_v0";
+
+export function planHandoffMode(value: unknown): JsonObject {
+  const input = requireJsonObject(value, "handoff mode plan");
+  if (input.schema_version !== HANDOFF_MODE_PLAN_SCHEMA) throw new Error("handoff mode plan schema mismatch");
+  const previous = requireStringLiteral(input.previous_mode, HANDOFF_MODES, "previous_mode");
+  const requested = requireStringLiteral(input.requested_mode, HANDOFF_MODES, "requested_mode");
+  const claims = requireStringArray(input.active_claimed_todo_ids, "active_claimed_todo_ids");
+  const leases = requireStringArray(input.active_lease_todo_ids, "active_lease_todo_ids");
+  const unchanged = previous === requested;
+  const rejected = !unchanged && (claims.length > 0 || leases.length > 0);
+  return {schema_version: "loopx_handoff_mode_plan_result_v0",
+    outcome: unchanged ? "no_change" : rejected ? "rejected" : "apply",
+    code: unchanged ? "handoff_mode_unchanged" : rejected ? "handoff_mode_not_quiescent" : "handoff_mode_transition",
+    idempotent: unchanged, previous_mode: previous, handoff_mode: requested};
+}
diff --git a/loopx/control_plane/coordination/handoff_mode_runtime.ts b/loopx/control_plane/coordination/handoff_mode_runtime.ts
new file mode 100644
index 0000000000..c7c82bc58a
--- /dev/null
+++ b/loopx/control_plane/coordination/handoff_mode_runtime.ts
@@ -0,0 +1,29 @@
+/** Local transport binds the existing provider and writer fence to the mode transaction. */
+import type {JsonObject} from "../effect_program.ts";
+import {requireJsonObject} from "../runtime_decode.ts";
+import {requireAuthorityStoreId} from "./authority_store_codec.ts";
+import {openLocalAuthorityStore, localAuthorityOpenFailure} from "./local_authority_provider.ts";
+import {runtimeRoot, sourceAuthorityFor, withCanonicalWriter} from "./local_authority_runtime.ts";
+import {ShadowManagementError} from "./shadow_management.ts";
+import {executeHandoffModeSet, HANDOFF_MODE_SET_SCHEMA} from "./handoff_mode_transaction.ts";
+
+export async function setLocalHandoffMode(value: unknown): Promise<JsonObject> {
+  const evidence = {source_authority: "file_v0", decision_read_from_provider: true, legacy_fallback_used: false};
+  try {
+    const input = requireJsonObject(value, "handoff mode set request");
+    if (input.schema_version !== HANDOFF_MODE_SET_SCHEMA) throw new Error("handoff mode request schema mismatch");
+    const root = runtimeRoot(input.runtime_root);
+    const goalId = requireAuthorityStoreId(input.goal_id, "goal id");
+    return await withCanonicalWriter(root, goalId, input.dry_run === true, async () => {
+      const store = await openLocalAuthorityStore(root, goalId);
+      evidence.source_authority = sourceAuthorityFor(store);
+      return {...await executeHandoffModeSet(store, {goal_id: goalId,
+        operation_id: input.operation_id as string, requested_mode: input.requested_mode as string,
+        observed_at: input.observed_at as string, dry_run: input.dry_run as boolean}), ...evidence};
+    });
+  } catch (error) {
+    return {schema_version: "loopx_coordination_handoff_mode_set_result_v0", status: "failed", changed: false,
+      reason_code: error instanceof ShadowManagementError ? error.reason_code : "handoff_mode_unavailable",
+      reason: error instanceof Error ? error.message : String(error), ...evidence, ...localAuthorityOpenFailure(error)};
+  }
+}
diff --git a/loopx/control_plane/coordination/handoff_mode_transaction.ts b/loopx/control_plane/coordination/handoff_mode_transaction.ts
new file mode 100644
index 0000000000..371ca23845
--- /dev/null
+++ b/loopx/control_plane/coordination/handoff_mode_transaction.ts
@@ -0,0 +1,96 @@
+/** Provider-neutral mode transition: quiescence and mode share one CAS snapshot. */
+import type {JsonObject} from "../effect_program.ts";
+import type {AuthorityStore, AuthorityStoreReceiptResult} from "./authority_store.ts";
+import {canonicalAuthorityObject, canonicalAuthoritySha256, requireAuthorityStoreId} from "./authority_store_codec.ts";
+import {indexCoordinationProjection, validateCoordinationTodoReadModel} from "./coordination_projection.ts";
+import {HANDOFF_MODES, HANDOFF_MODE_PLAN_SCHEMA, planHandoffMode} from "./handoff_mode_policy.ts";
+import {requireBoolean, requireStringLiteral} from "../runtime_decode.ts";
+import {parseIsoTimestamp} from "../runtime_timestamp.ts";
+import {leaseIsActive, TASK_LEASE_SCHEMA_VERSION} from "../work_items/task_lease_acquire.ts";
+
+export const HANDOFF_MODE_SET_SCHEMA = "loopx_coordination_handoff_mode_set_request_v0";
+const RESULT_SCHEMA = "loopx_coordination_handoff_mode_set_result_v0";
+const RECEIPT_SCHEMA = "loopx_coordination_handoff_mode_receipt_v0";
+export interface HandoffModeSetInput {
+  goal_id: string;
+  operation_id: string;
+  requested_mode: string;
+  observed_at: string;
+  dry_run: boolean;
+}
+
+function failure(reason_code: string, reason: string): JsonObject {
+  return {schema_version: RESULT_SCHEMA, status: "failed", changed: false, reason_code, reason};
+}
+
+function replay(receipt: AuthorityStoreReceiptResult, input: HandoffModeSetInput, hash: string,
+  status: "applied" | "replayed" | "recovered"): JsonObject | null {
+  if (receipt.status === "missing") return null;
+  if (receipt.status !== "found") return {schema_version: RESULT_SCHEMA, ...receipt, changed: false};
+  const record = receipt.receipts[0];
+  if (receipt.receipts.length !== 1 || record?.schema_version !== RECEIPT_SCHEMA ||
+    record.goal_id !== input.goal_id || record.operation_id !== input.operation_id || record.request_sha256 !== hash) {
+    return failure("coordination_operation_identity_mismatch", "operation id names another handoff mode intent");
+  }
+  const decision = canonicalAuthorityObject(record.decision, "handoff mode decision receipt");
+  return {schema_version: RESULT_SCHEMA, ...decision, status,
+    changed: status !== "replayed" && decision.changed === true,
+    provider_revision: receipt.provider_revision, cursor: receipt.cursor};
+}
+
+export async function executeHandoffModeSet(store: AuthorityStore, raw: HandoffModeSetInput): Promise<JsonObject> {
+  let input: HandoffModeSetInput;
+  let now: Date;
+  try {
+    input = {...raw, goal_id: requireAuthorityStoreId(raw.goal_id, "goal id"),
+      operation_id: requireAuthorityStoreId(raw.operation_id, "operation id"),
+      requested_mode: requireStringLiteral(raw.requested_mode, HANDOFF_MODES, "requested_mode"),
+      dry_run: requireBoolean(raw.dry_run, "dry_run")};
+    const parsed = typeof input.observed_at === "string" ? parseIsoTimestamp(input.observed_at) : null;
+    if (!parsed) throw new Error("observed_at must be a valid ISO timestamp");
+    now = parsed;
+  } catch (error) { return failure("invalid_handoff_mode_request", String(error)); }
+  // Retry time is observation context, not a new intent. Preview never consumes an operation id.
+  const hash = canonicalAuthoritySha256({goal_id: input.goal_id, requested_mode: input.requested_mode});
+  if (!input.dry_run) {
+    const previous = replay(await store.readReceipt(input.operation_id), input, hash, "replayed");
+    if (previous) return previous;
+  }
+  const loaded = await store.loadAuthority();
+  if (loaded.status !== "loaded") return {schema_version: RESULT_SCHEMA, ...loaded, changed: false};
+  let decision: JsonObject;
+  try {
+    const head = loaded.head;
+    const indexed = indexCoordinationProjection(head, input.goal_id);
+    validateCoordinationTodoReadModel(head, input.goal_id);
+    const previous = requireStringLiteral(head.handoff_mode ?? "legacy", HANDOFF_MODES, "canonical handoff_mode");
+    const claimed = [...indexed.todos.values()].filter(todo => todo.archive_state === "active" &&
+      todo.done !== true && typeof todo.claimed_by === "string" && todo.claimed_by.trim()).map(todo => ({
+      todo_id: todo.todo_id, claimed_by: todo.claimed_by, status: todo.status}));
+    const leases = [...indexed.leases.values()].filter(lease => {
+      if (lease.schema_version !== TASK_LEASE_SCHEMA_VERSION) throw new Error("canonical lease schema mismatch");
+      return leaseIsActive(lease, now);
+    }).map(lease => ({todo_id: lease.todo_id, owner: lease.owner, expires_at: lease.expires_at}));
+    const plan = planHandoffMode({schema_version: HANDOFF_MODE_PLAN_SCHEMA, previous_mode: previous,
+      requested_mode: input.requested_mode, active_claimed_todo_ids: claimed.map(todo => todo.todo_id),
+      active_lease_todo_ids: leases.map(lease => lease.todo_id)});
+    decision = {goal_id: input.goal_id, operation_id: input.operation_id, previous_mode: previous,
+      previous_mode_valid: true, handoff_mode: input.requested_mode, changed: plan.outcome === "apply"};
+    if (plan.outcome === "rejected") return {...failure(String(plan.code),
+      "handoff_mode can only change without unfinished claimed Todos or time-active leases"),
+      ...decision, claimed_todos: claimed, active_leases: leases, provider_revision: loaded.provider_revision};
+  } catch (error) { return failure("invalid_handoff_mode_authority", String(error)); }
+  if (input.dry_run) return {schema_version: RESULT_SCHEMA, ...decision, status: "planned",
+    provider_revision: loaded.provider_revision};
+  // Seal even an unchanged accepted intent: retry after another mode switch must not reapply it.
+  const commit = await store.commitAuthority({operation_id: input.operation_id,
+    expected_provider_revision: loaded.provider_revision,
+    next_projection: {...loaded.head, handoff_mode: input.requested_mode},
+    events: decision.changed ? [{schema_version: "loopx_handoff_mode_changed_v0", ...decision}] : [],
+    receipts: [{schema_version: RECEIPT_SCHEMA, goal_id: input.goal_id, operation_id: input.operation_id,
+      request_sha256: hash, decision}]});
+  return replay(await store.readReceipt(input.operation_id), input, hash,
+    commit.status === "applied" ? "applied" : "recovered") ?? (commit.status === "applied"
+    ? failure("coordination_commit_readback_mismatch", "applied mode transaction lacks its durable receipt")
+    : {schema_version: RESULT_SCHEMA, ...commit, changed: false});
+}
diff --git a/loopx/control_plane/coordination/local_authority_runtime.ts b/loopx/control_plane/coordination/local_authority_runtime.ts
index 7ba94c105e..0e95a7a3d8 100644
--- a/loopx/control_plane/coordination/local_authority_runtime.ts
+++ b/loopx/control_plane/coordination/local_authority_runtime.ts
@@ -98,11 +98,11 @@ export {
 } from "./coordination_state_contract.generated.ts";
 export { LEGACY_COORDINATION_WRITER_FENCE_SCHEMA } from "./legacy_writer_fence.ts";
 
-function sourceAuthorityFor(store: AuthorityStore): "sqlite_v0" | "file_v0" {
+export function sourceAuthorityFor(store: AuthorityStore): "sqlite_v0" | "file_v0" {
   return store instanceof SqliteAuthorityStore ? "sqlite_v0" : "file_v0";
 }
 
-async function withCanonicalWriter<T>(root: string, goalId: string, dryRun: boolean, write: () => Promise<T>): Promise<T> {
+export async function withCanonicalWriter<T>(root: string, goalId: string, dryRun: boolean, write: () => Promise<T>): Promise<T> {
   if (dryRun) return await write();
   return await withFileMutationLock(shadowMaintenanceLockPath(root, goalId), async () => {
     await requireShadowPrimaryWriteAllowed(root, goalId);
@@ -147,7 +147,7 @@ interface LocalAuthorityRuntimeDependencies {
   createCanonicalStore?: (directory: string, goalId: string) => AuthorityStore;
 }
 
-function runtimeRoot(value: unknown): string {
+export function runtimeRoot(value: unknown): string {
   if (typeof value !== "string" || value.trim() !== value || !isAbsolute(value)) {
     throw new Error("runtime_root must be an absolute path");
   }
diff --git a/loopx/control_plane/coordination/todo_lifecycle_decision.ts b/loopx/control_plane/coordination/todo_lifecycle_decision.ts
index 22984e0ff2..f361347433 100644
--- a/loopx/control_plane/coordination/todo_lifecycle_decision.ts
+++ b/loopx/control_plane/coordination/todo_lifecycle_decision.ts
@@ -25,13 +25,12 @@ export const COORDINATION_TERMINAL_FENCE_REQUEST_SCHEMA =
   "loopx_coordination_terminal_fence_request_v0";
 export const COORDINATION_TERMINAL_FENCE_RESULT_SCHEMA =
   "loopx_coordination_terminal_fence_result_v0";
-const HANDOFF_MODES = ["legacy", "soft_claim", "hard_lease"] as const;
+import {HANDOFF_MODES, type HandoffMode} from "./handoff_mode_policy.ts";
 const OUTCOMES = ["approve", "reject", "cancel"] as const;
 const AUTHORITY_ACTIONS = ["complete", "reassign", "supersede", "update"] as const;
 const EXECUTOR_RECLAIM_ACTION = "reclaim";
 
 type LifecycleCommand = typeof COMMANDS[number] | typeof MUTATION_COMMANDS[number];
-type HandoffMode = typeof HANDOFF_MODES[number];
 type DecisionOutcome = typeof OUTCOMES[number];
 
 interface DecisionScope extends JsonObject {
diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index c18a5fec09..7a1347fb3a 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -1,3 +1,5 @@
+import {planHandoffMode} from "./coordination/handoff_mode_policy.ts";
+import {setLocalHandoffMode} from "./coordination/handoff_mode_runtime.ts";
 import {evaluateTaskLeaseOwnerEligibility} from "./work_items/task_lease_eligibility.ts";
 import { evaluateSubagentContext, describeSubagentContext } from "./subagent_context.ts";
 import {
@@ -473,6 +475,8 @@ export function createEffectRuntimeHandlers(
     ["coordination.local_authority.todo_create", createLocalCoordinationTodo],
     ["coordination.local_authority.todo_update", updateLocalCoordinationTodo],
     ["coordination.local_authority.monitor_poll", pollLocalCoordinationMonitor],
+    ["coordination.handoff_mode.plan", planHandoffMode],
+    ["coordination.local_authority.handoff_mode_set", setLocalHandoffMode],
     ["coordination.local_authority.todo_terminal", terminalLifecycleLocalCoordinationTodo],
     ["coordination.local_authority.todo_archive", archiveLocalCoordinationTodos],
     ["coordination.local_authority.todo_archive_ack", acknowledgeLocalCoordinationTodoArchive],
diff --git a/loopx/control_plane/todos/handoff_mode.py b/loopx/control_plane/todos/handoff_mode.py
index fec1706424..08affa4c03 100644
--- a/loopx/control_plane/todos/handoff_mode.py
+++ b/loopx/control_plane/todos/handoff_mode.py
@@ -1,12 +1,13 @@
 """Per-goal handoff mode: which ownership authority governs todo handoffs.
 
-The goal's ACTIVE_GOAL_STATE.md YAML front-matter may declare ``handoff_mode``:
+Before promotion the active-state frontmatter declares ``handoff_mode``; afterward
+the selected canonical provider owns it. Public show/set route by that authority:
 
 * absent / ``legacy`` (default): today's dual soft-claim + hard-lease
   behavior, byte-for-byte. The known soft-claim/hard-lease split brain stays
   open in this mode by design; it is surfaced additively, never silently
   repaired.
-* ``soft_claim``: the markdown claim is the only ownership record. Task-lease
+* ``soft_claim``: the Todo claim is the only ownership record. Task-lease
   acquire/renew/transfer are typed-rejected; release and inspect stay allowed
   for cleanup and observability of legacy leftovers.
 * ``hard_lease``: ownership changes on an existing todo require the acting
@@ -17,13 +18,11 @@
   ``coordination.todo_lifecycle_authority`` override is the one audited door
   through the gate.
 
-The mode lives in the front-matter (not the registry) because the markdown
-file is the artifact that travels across endpoints; lease JSON and registry
-are host-local. Pre-NoKV the file syncs last-writer-wins, so two hosts can
-briefly disagree about the mode; that window is documented, not engineered
-around here.
+Canonical mode, complete Todo/lease quiescence, CAS and replay share one TypeScript
+transaction. Stale or missing Markdown and local lease files are not fallback
+sources. The legacy mode below remains a frontmatter compatibility contract.
 
-The v0 transition scan is materialized-state only: it reads open claims from
+The unpromoted v0 transition scan is materialized-state only: it reads open claims from
 the locked ``ACTIVE_GOAL_STATE.md`` text plus time-active local lease files. It
 does not merge the event projection, so a claim that exists only in the event
 log can be missed. A successful switch is therefore not a proof that every
@@ -119,13 +118,8 @@ def goal_handoff_mode_for_goal(
     project: Path | None = None,
     state_file: Path | None = None,
 ) -> str:
-    _project, resolved_state_file = _resolve_state(
-        registry_path=registry_path,
-        goal_id=goal_id,
-        project=project,
-        state_file=state_file,
-    )
-    return goal_handoff_mode(resolved_state_file.read_text(encoding="utf-8"))
+    return str(show_goal_handoff_mode(registry_path=registry_path, goal_id=goal_id,
+        project=project, state_file=state_file)["handoff_mode"])
 
 
 def enter_todo_ownership_handoff_gate(
@@ -255,7 +249,17 @@ def show_goal_handoff_mode(
     goal_id: str,
     project: Path | None = None,
     state_file: Path | None = None,
+    runtime_root_arg: str | None = None,
 ) -> dict[str, Any]:
+    from ..work_items.task_lease import runtime_root_from_registry
+    from .provider_handoff_mode import read_canonical_handoff_mode
+
+    canonical = read_canonical_handoff_mode(
+        runtime_root=runtime_root_from_registry(registry_path, runtime_root_arg), goal_id=goal_id)
+    if canonical is not None:
+        return {"ok": True, "schema_version": HANDOFF_MODE_SCHEMA_VERSION, "action": "show",
+                "goal_id": goal_id, **canonical,
+                "handoff_mode": normalize_handoff_mode(canonical["handoff_mode"]), "source": "canonical_provider"}
     _project, resolved_state_file = _resolve_state(
         registry_path=registry_path,
         goal_id=goal_id,
@@ -314,14 +318,16 @@ def _quiescence_offenders(
         runtime_root_from_registry,
         task_lease_dir,
     )
-    from .active_state_todo_parser import parse_active_state_todos
+    from .active_state_todo_parser import parse_todo_source
+    from .todo_summary import structured_todo_item, todo_projection_sort_key
 
+    # Quiescence needs normalized ownership facts, not status/resume/capability display.
+    # Retain the public offender order without evaluating unrelated projection rules.
     claimed: list[dict[str, Any]] = []
-    todos = parse_active_state_todos(state_text, item_limit=None)
-    for role in ("user_todos", "agent_todos"):
-        summary = todos.get(role)
-        items = summary.get("items") if isinstance(summary, dict) else []
-        for item in items or []:
+    todos, _, sections = parse_todo_source(state_text)
+    for role in ("user", "agent"):
+        items = [structured_todo_item(item, role=role, source_section=sections[role]) for item in todos[role]]
+        for item in sorted(items, key=todo_projection_sort_key):
             if not isinstance(item, dict) or item.get("done") is True:
                 continue
             owner = normalize_todo_claimed_by(item.get("claimed_by"))
@@ -401,8 +407,13 @@ def set_goal_handoff_mode(
     project: Path | None = None,
     state_file: Path | None = None,
     runtime_root_arg: str | None = None,
+    operation_id: str | None = None,
+    dry_run: bool = False,
 ) -> dict[str, Any]:
-    """Set the goal handoff mode; requires a quiescent goal for transitions.
+    """Set the authoritative goal mode; changed modes require quiescence.
+
+    Promoted Goals use one provider transaction; the remaining text below
+    describes the unpromoted compatibility writer.
 
     In v0, quiescence means no open todo materialized in the locked active-state
     Markdown carries a claimed_by owner and no time-active lease file exists
@@ -434,18 +445,25 @@ def set_goal_handoff_mode(
             "handoff-mode set requires an explicit --mode value",
             code="invalid_handoff_mode",
         )
+    from .provider_handoff_mode import set_canonical_handoff_mode
+
+    runtime_root = runtime_root_from_registry(registry_path, runtime_root_arg)
+    canonical = set_canonical_handoff_mode(runtime_root=runtime_root, goal_id=goal_id,
+        mode=requested, operation_id=operation_id, dry_run=dry_run)
+    if canonical is not None:
+        return canonical
+    if operation_id is not None:
+        raise HandoffModeError("--operation-id requires canonical authority", code="handoff_mode_operation_id_unsupported")
     _project, resolved_state_file = _resolve_state(
         registry_path=registry_path,
         goal_id=goal_id,
         project=project,
         state_file=state_file,
     )
-    # One effective runtime root for the lease lock, the quiescence scan, and
-    # the post-commit observation of this call.
-    runtime_root = runtime_root_from_registry(registry_path, runtime_root_arg)
+    # The already resolved root also governs the legacy lock, scan and capture.
     with legacy_todo_write_transaction(
         registry_path, goal_id, resolved_state_file, None, "handoff_mode_set",
-        False, runtime_root=runtime_root,
+        dry_run, runtime_root=runtime_root,
     ):
         original = resolved_state_file.read_text(encoding="utf-8")
         previous, previous_mode_fields = _previous_handoff_mode_fields(
@@ -462,8 +480,10 @@ def set_goal_handoff_mode(
         }
         if previous == requested:
             payload["changed"] = False
+            if dry_run:
+                payload["dry_run"] = True
             return payload
-        capture = begin_todo_runtime_shadow_capture(
+        capture = None if dry_run else begin_todo_runtime_shadow_capture(
             registry_path=registry_path, runtime_root=runtime_root, goal_id=goal_id,
             state_path=resolved_state_file, write_class="handoff_mode_set",
             original_text=original,
@@ -529,6 +549,8 @@ def set_goal_handoff_mode(
                         "active_leases": leases,
                     },
                 )
+            if dry_run:
+                return {**payload, "dry_run": True, "changed": True}
             lines = original.splitlines()
             _write_handoff_mode_frontmatter(lines, requested)
             new_text = "\n".join(lines) + ("\n" if original.endswith("\n") else "")
diff --git a/loopx/control_plane/todos/provider_handoff_mode.py b/loopx/control_plane/todos/provider_handoff_mode.py
new file mode 100644
index 0000000000..93a4c9bd05
--- /dev/null
+++ b/loopx/control_plane/todos/provider_handoff_mode.py
@@ -0,0 +1,39 @@
+"""Transport and readback for the canonical mode owner; no local authority fallback."""
+from pathlib import Path
+from typing import Any
+from uuid import uuid4
+
+from ..coordination.local_authority import (
+    LOCAL_AUTHORITY_SOURCES, LocalCoordinationAuthorityUnavailable,
+    local_authority_is_promoted, read_canonical_todos_if_promoted,
+)
+from ..effect_runtime import effect_runtime_result
+from ..runtime.time import now_local_iso
+
+
+def read_canonical_handoff_mode(*, runtime_root: Path, goal_id: str) -> dict[str, Any] | None:
+    source = read_canonical_todos_if_promoted(runtime_root=runtime_root, goal_id=goal_id, include_leases=True)
+    if source is None:
+        return None
+    return {key: source[key] for key in ("handoff_mode", "source_authority", "provider_revision",
+                                        "decision_read_from_provider", "legacy_fallback_used")}
+
+
+def set_canonical_handoff_mode(*, runtime_root: Path, goal_id: str, mode: str,
+                               operation_id: str | None, dry_run: bool) -> dict[str, Any] | None:
+    if not local_authority_is_promoted(runtime_root=runtime_root, goal_id=goal_id):
+        return None
+    result = effect_runtime_result("coordination.local_authority.handoff_mode_set", {
+        "schema_version": "loopx_coordination_handoff_mode_set_request_v0",
+        "runtime_root": str(runtime_root.expanduser().resolve()), "goal_id": goal_id,
+        "operation_id": (operation_id if operation_id is not None else f"handoff-mode:{goal_id}:{uuid4().hex}"),
+        "requested_mode": mode, "observed_at": now_local_iso(), "dry_run": dry_run,
+    })
+    if (not isinstance(result, dict) or result.get("status") not in {"applied", "replayed", "recovered", "planned"}
+        or result.get("source_authority") not in LOCAL_AUTHORITY_SOURCES
+        or result.get("decision_read_from_provider") is not True or result.get("legacy_fallback_used") is not False):
+        payload = result if isinstance(result, dict) else {}
+        raise LocalCoordinationAuthorityUnavailable(str(payload.get("reason") or "canonical mode unavailable"),
+            code=str(payload.get("reason_code") or "handoff_mode_unavailable"), payload=payload)
+    return {**result, "ok": True, "schema_version": "goal_handoff_mode_v0", "action": "set",
+            "source": "canonical_provider", "dry_run": dry_run}
diff --git a/tests/control_plane/test_canonical_handoff_mode.py b/tests/control_plane/test_canonical_handoff_mode.py
new file mode 100644
index 0000000000..7a69397027
--- /dev/null
+++ b/tests/control_plane/test_canonical_handoff_mode.py
@@ -0,0 +1,98 @@
+"""Public mode commands must use the selected canonical snapshot after promotion."""
+import json
+import subprocess
+import sys
+
+import pytest
+
+from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime
+from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
+
+
+@pytest.fixture(params=["file", "sqlite"])
+def canonical_mode(tmp_path, monkeypatch, request):
+    if request.param == "sqlite":
+        isolate_sqlite_runtime(tmp_path, monkeypatch)
+    state = tmp_path / "state.md"
+    state.write_text("---\nhandoff_mode: hard_lease\n---\n\n## Agent Todo\n")
+    runtime = tmp_path / "runtime"
+    registry = tmp_path / "registry.json"
+    registry.write_text(json.dumps({"common_runtime_root": str(runtime), "goals": [
+        {"id": "mode-goal", "repo": str(tmp_path), "state_file": state.name}]}))
+    projection = build_todo_runtime_shadow_projection(goal_id="mode-goal", handoff_mode="soft_claim", todos=[])
+    initialize_canonical_authority(runtime, "mode-goal", projection, state_path=state, provider=request.param)
+    def cli(*args):
+        result = subprocess.run([sys.executable, "-m", "loopx.cli", "--registry", str(registry),
+            "--format", "json", "handoff-mode", *args, "--goal-id", "mode-goal"],
+            capture_output=True, text=True, timeout=90)
+        return result.returncode, json.loads(result.stdout)
+    return state, runtime, cli
+
+
+def test_canonical_show_ignores_stale_or_missing_display(canonical_mode):
+    state, _, cli = canonical_mode
+    before = state.read_bytes()
+    code, result = cli("show")
+    assert code == 0 and result["handoff_mode"] == "soft_claim", result
+    assert result["source"] == "canonical_provider"
+    assert state.read_bytes() == before
+    state.unlink()
+    code, result = cli("show")
+    assert code == 0 and result["handoff_mode"] == "soft_claim", result
+    assert not state.exists()
+
+
+def test_public_set_preview_replay_and_later_mode_preserve_display(canonical_mode):
+    state, runtime, cli = canonical_mode
+    from loopx.control_plane.coordination.local_authority import read_canonical_todos_if_promoted
+    def read():
+        return read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="mode-goal", include_leases=True)
+    original = state.read_bytes()
+    before = read()
+    code, preview = cli("set", "--mode", "legacy", "--dry-run", "--operation-id", "mode-intent")
+    assert code == 0 and preview["status"] == "planned", preview
+    assert read() == before
+    code, changed = cli("set", "--mode", "legacy", "--operation-id", "mode-intent")
+    assert code == 0 and changed["changed"] is True, changed
+    after = read()
+    assert after["handoff_mode"] == "legacy"
+    assert after["todos"] == before["todos"] and after["leases"] == before["leases"]
+    assert cli("set", "--mode", "hard_lease", "--operation-id", "later-mode")[0] == 0
+    later = read()
+    code, replay = cli("set", "--mode", "legacy", "--operation-id", "mode-intent")
+    assert code == 0 and replay["status"] == "replayed" and replay["changed"] is False, replay
+    assert read() == later
+    code, mismatch = cli("set", "--mode", "soft_claim", "--operation-id", "mode-intent")
+    assert code == 1 and mismatch["error_code"] == "coordination_operation_identity_mismatch", mismatch
+    assert state.read_bytes() == original
+    state.unlink()
+    assert cli("set", "--mode", "soft_claim", "--operation-id", "without-display")[0] == 0
+    assert cli("show")[1]["handoff_mode"] == "soft_claim"
+    assert not state.exists()
+
+
+def test_canonical_mode_does_not_resurrect_legacy_lease(canonical_mode):
+    state, runtime, cli = canonical_mode
+    from loopx.control_plane.work_items.task_lease import task_lease_dir
+    lease_dir = task_lease_dir(runtime_root=runtime, goal_id="mode-goal")
+    lease_dir.mkdir(parents=True, exist_ok=True)
+    stale = lease_dir / "todo_legacy.json"
+    stale.write_text(json.dumps({"schema_version": "task_lease_v0", "todo_id": "todo_legacy",
+        "status": "active", "expires_at": "2099-01-01T00:00:00Z", "owner": "agent-a"}))
+    before = stale.read_bytes(), state.read_bytes()
+    code, result = cli("set", "--mode", "hard_lease")
+    assert code == 0, result
+    assert (stale.read_bytes(), state.read_bytes()) == before
+
+
+def test_provider_failure_is_not_a_legacy_fallback(canonical_mode, monkeypatch):
+    state, runtime, _ = canonical_mode
+    from loopx.control_plane.todos import provider_handoff_mode
+    from loopx.control_plane.coordination.local_authority import LocalCoordinationAuthorityUnavailable
+    monkeypatch.setattr(provider_handoff_mode, "effect_runtime_result", lambda *_args: {
+        "status": "unavailable", "reason_code": "synthetic_provider_down", "reason": "Unavailable"})
+    before = state.read_bytes()
+    with pytest.raises(LocalCoordinationAuthorityUnavailable, match="Unavailable"):
+        provider_handoff_mode.set_canonical_handoff_mode(runtime_root=runtime, goal_id="mode-goal",
+            mode="hard_lease", operation_id="unavailable", dry_run=False)
+    assert state.read_bytes() == before
diff --git a/tests/control_plane/test_shadow_fence_caller_parity_e2e.py b/tests/control_plane/test_shadow_fence_caller_parity_e2e.py
index 75a335e3cc..6de744494b 100644
--- a/tests/control_plane/test_shadow_fence_caller_parity_e2e.py
+++ b/tests/control_plane/test_shadow_fence_caller_parity_e2e.py
@@ -238,6 +238,11 @@ def test_fence_caller_parity(workspaces: Callable[[str], Workspace], row: dict)
         assert {key: observed["envelope"].get(key) for key in row["expect"]} == row["expect"], observed
     else:
         assert observed["envelope"] == row["expect"], observed
+    if row["caller"] == "handoff_mode_set":
+        # This command now crosses the canonical boundary. Its dynamic revision,
+        # operation ID and lease expiry are not a literal legacy-writer envelope.
+        assert observed["envelope"].get("claimed_todos") or observed["envelope"].get("active_leases"), observed
+        assert observed["envelope"].get("provider_revision"), observed
     assert observed["effect"] == row["effect"], observed
     assert observed["outbox_added"] == row.get("outbox_added", []), observed
 
diff --git a/tests/control_plane_ts/authority_store_conformance.ts b/tests/control_plane_ts/authority_store_conformance.ts
index 88670a9378..053d825401 100644
--- a/tests/control_plane_ts/authority_store_conformance.ts
+++ b/tests/control_plane_ts/authority_store_conformance.ts
@@ -1,5 +1,6 @@
 import {registerAuthorityScanConformance} from "./authority_scan_conformance.ts";
 import {executeCoordinationTodoArchiveCompleted} from "../../loopx/control_plane/coordination/todo_archive.ts";
+import {registerHandoffModeConformance} from "./handoff_mode_conformance.ts";
 import assert from "node:assert/strict";
 import { createHash } from "node:crypto";
 import test from "node:test";
@@ -209,6 +210,7 @@ export function registerAuthorityStoreConformance(
   registerAuthorityScanConformance(providerName, factory);
   registerNativePlanningUpdateConformance(providerName, factory);
   registerCoordinationReceiptConformance(providerName, factory);
+  registerHandoffModeConformance(providerName, factory);
   for (const native of [false, true]) test(`${providerName} conformance: standing revocation survives canonical ordering and archive (${native ? "native" : "legacy"})`, async (t) => {
     const {store} = await factory(t);
     const goal = "goal-standing";
diff --git a/tests/control_plane_ts/handoff_mode_conformance.ts b/tests/control_plane_ts/handoff_mode_conformance.ts
new file mode 100644
index 0000000000..7e3e88587a
--- /dev/null
+++ b/tests/control_plane_ts/handoff_mode_conformance.ts
@@ -0,0 +1,135 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import type {JsonObject} from "../../loopx/control_plane/effect_program.ts";
+import type {AuthorityStore} from "../../loopx/control_plane/coordination/authority_store.ts";
+import {canonicalAuthoritySha256} from "../../loopx/control_plane/coordination/authority_store_codec.ts";
+import {executeHandoffModeSet} from "../../loopx/control_plane/coordination/handoff_mode_transaction.ts";
+import type {AuthorityStoreConformanceFactory} from "./authority_store_conformance.ts";
+import {productionScaleCoordinationFixture} from "./production_scale_coordination_fixture.ts";
+
+const request = {goal_id: "mode-goal", operation_id: "mode-change", requested_mode: "soft_claim",
+  observed_at: "2026-09-13T00:00:00Z", dry_run: false};
+async function head(store: AuthorityStore) {
+  const loaded = await store.loadAuthority();
+  assert.equal(loaded.status, "loaded");
+  if (loaded.status !== "loaded") throw new Error("missing fixture");
+  return loaded;
+}
+function quiescent() {
+  const projection = productionScaleCoordinationFixture(request.goal_id).projection;
+  for (const todo of projection.todos as JsonObject[]) delete todo.claimed_by;
+  for (const lease of projection.leases as JsonObject[]) lease.status = "released";
+  (projection.todo_read_model as JsonObject).records_sha256 = canonicalAuthoritySha256(projection.todos);
+  return projection;
+}
+async function seed(store: AuthorityStore, projection = quiescent()) {
+  assert.equal((await store.commitAuthority({operation_id: "seed", expected_provider_revision: null,
+    next_projection: projection, events: [], receipts: []})).status, "applied");
+}
+function intercept(store: AuthorityStore, commit: AuthorityStore["commitAuthority"]): AuthorityStore {
+  return {storeIdentity: () => store.storeIdentity(), loadAuthority: () => store.loadAuthority(),
+    readReceipt: id => store.readReceipt(id), scanCommitted: (cursor, limit) => store.scanCommitted(cursor, limit),
+    commitAuthority: commit};
+}
+
+export function registerHandoffModeConformance(provider: string, factory: AuthorityStoreConformanceFactory) {
+  test(`${provider}: mode transition preserves the large projection and seals replay, including no-op`, async t => {
+    const {store, contender} = await factory(t);
+    await seed(store);
+    const before = await head(store);
+    const preview = await executeHandoffModeSet(store, {...request, dry_run: true});
+    assert.equal(preview.status, "planned");
+    assert.deepEqual(await head(store), before);
+    assert.equal((await store.readReceipt(request.operation_id)).status, "missing");
+    const result = await executeHandoffModeSet(store, request);
+    assert.equal(result.status, "applied", JSON.stringify(result));
+    const after = await head(contender);
+    assert.equal((after.head.todos as JsonObject[]).length, 464);
+    assert.deepEqual(after.head, {...before.head, handoff_mode: "soft_claim"} as JsonObject);
+    const noop = {...request, operation_id: "unchanged-mode"};
+    const unchanged = await executeHandoffModeSet(store, noop);
+    assert.equal(unchanged.changed, false);
+    assert.equal((await store.readReceipt(noop.operation_id)).status, "found");
+    assert.equal((await executeHandoffModeSet(store, {...request, operation_id: "later-mode", requested_mode: "legacy"})).status, "applied");
+    const later = await head(store);
+    for (const retry of [request, noop]) {
+      const replay = await executeHandoffModeSet(contender, {...retry, observed_at: "2028-01-01T00:00:00Z"});
+      assert.equal(replay.status, "replayed");
+      assert.equal(replay.changed, false);
+      assert.deepEqual(await head(store), later);
+    }
+    assert.equal((await executeHandoffModeSet(store, {...request, requested_mode: "hard_lease"})).reason_code,
+      "coordination_operation_identity_mismatch");
+  });
+
+  for (const kind of ["claim", "lease", "invalid_expiry", "unknown_lease_schema"] as const) {
+    test(`${provider}: mode rejects ${kind} anywhere in the full authority without writing`, async t => {
+      const {store} = await factory(t);
+      const projection = quiescent();
+      if (kind === "claim") {
+        const todo = [...projection.todos as JsonObject[]].reverse().find(todo => todo.done !== true)!;
+        todo.claimed_by = "agent-b";
+        (projection.todo_read_model as JsonObject).records_sha256 = canonicalAuthoritySha256(projection.todos);
+      } else {
+        const lease = (projection.leases as JsonObject[]).at(-1)!;
+        lease.status = "active";
+        lease.expires_at = kind === "invalid_expiry" ? "invalid" : "2027-01-01T00:00:00Z";
+        if (kind === "unknown_lease_schema") lease.schema_version = "unknown";
+      }
+      await seed(store, projection);
+      const before = await head(store);
+      const result = await executeHandoffModeSet(store, request);
+      assert.equal(result.status, "failed", JSON.stringify(result));
+      assert.equal(result.reason_code, kind === "claim" || kind === "lease"
+        ? "handoff_mode_not_quiescent" : "invalid_handoff_mode_authority");
+      assert.deepEqual(await head(store), before);
+      assert.equal((await store.readReceipt(request.operation_id)).status, "missing");
+    });
+  }
+
+  test(`${provider}: expiry at observation time is quiescent and invalid modes never repair canonical state`, async t => {
+    const {store} = await factory(t);
+    const projection = quiescent();
+    const lease = (projection.leases as JsonObject[])[0]!;
+    lease.status = "active"; lease.expires_at = request.observed_at;
+    await seed(store, projection);
+    const before = await head(store);
+    for (const invalid of [{requested_mode: "banana"}, {observed_at: "not-a-time"}]) {
+      assert.equal((await executeHandoffModeSet(store, {...request, ...invalid})).status, "failed");
+      assert.deepEqual(await head(store), before);
+    }
+    assert.equal((await executeHandoffModeSet(store, request)).status, "applied");
+  });
+
+  test(`${provider}: concurrent claim invalidates quiescence; no commit follows a stale snapshot`, async t => {
+    const {store, contender} = await factory(t);
+    await seed(store);
+    const raced = intercept(store, async commit => {
+      const current = await head(contender);
+      const next = structuredClone(current.head);
+      const todo = (next.todos as JsonObject[]).find(todo => todo.done !== true)!;
+      todo.claimed_by = "agent-b";
+      (next.todo_read_model as JsonObject).records_sha256 = canonicalAuthoritySha256(next.todos);
+      assert.equal((await contender.commitAuthority({operation_id: "concurrent-claim", expected_provider_revision: current.provider_revision,
+        next_projection: next, events: [], receipts: []})).status, "applied");
+      return store.commitAuthority(commit);
+    });
+    assert.equal((await executeHandoffModeSet(raced, request)).status, "conflict");
+    assert.equal((await head(store)).head.handoff_mode, "hard_lease");
+    assert.equal((await store.readReceipt(request.operation_id)).status, "missing");
+    assert.equal((await executeHandoffModeSet(store, request)).reason_code, "handoff_mode_not_quiescent");
+  });
+
+  test(`${provider}: lost commit response recovers its receipt without replaying the change`, async t => {
+    const {store} = await factory(t);
+    await seed(store);
+    const lost = intercept(store, async commit => {
+      assert.equal((await store.commitAuthority(commit)).status, "applied");
+      return {status: "ambiguous", reason_code: "response_lost", reason: "synthetic lost response"};
+    });
+    assert.equal((await executeHandoffModeSet(lost, request)).status, "recovered");
+    const after = await head(store);
+    assert.equal((await executeHandoffModeSet(store, request)).status, "replayed");
+    assert.deepEqual(await head(store), after);
+  });
+}
diff --git a/tests/fixtures/control_plane/legacy_writer_fence_caller_parity_v0.json b/tests/fixtures/control_plane/legacy_writer_fence_caller_parity_v0.json
index 0edc8527c7..db8950644b 100644
--- a/tests/fixtures/control_plane/legacy_writer_fence_caller_parity_v0.json
+++ b/tests/fixtures/control_plane/legacy_writer_fence_caller_parity_v0.json
@@ -1476,7 +1476,7 @@
       }
     },
     {
-      "id": "cli-handoff_mode_set-engaged",
+      "id": "cli-handoff_mode_set-canonical-quiescence",
       "surface": "cli",
       "workspace": "w1",
       "caller": "handoff_mode_set",
@@ -1487,15 +1487,13 @@
         "schema_version": "goal_handoff_mode_v0",
         "action": "set",
         "goal_id": "observable",
-        "error": "legacy coordination writer is fenced; use the promoted canonical authority (file_v0) for goal observable; fence caller-fixture; the primary record was not changed",
-        "error_code": "legacy_coordination_writer_fenced",
-        "write_check": {
-          "schema_version": "loopx_legacy_coordination_write_check_result_v0",
-          "status": "blocked",
-          "reason_code": "legacy_coordination_writer_fenced",
-          "authority_mode": "file_v0",
-          "fence_id": "caller-fixture"
-        }
+        "error_code": "handoff_mode_not_quiescent",
+        "status": "failed",
+        "changed": false,
+        "handoff_mode": "soft_claim",
+        "source_authority": "file_v0",
+        "decision_read_from_provider": true,
+        "legacy_fallback_used": false
       },
       "effect": {
         "added": [],
@@ -1515,7 +1513,8 @@
           "error_code": "legacy_coordination_writer_fenced",
           "schema_version": "loopx_legacy_coordination_write_check_result_v0"
         }
-      }
+      },
+      "match": "subset"
     },
     {
       "id": "cli-todo_archive_completed_preview-engaged",

From abb489786560c5680d07fe6255b86087615bbf88 Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Sun, 13 Sep 2026 07:03:44 +0800
Subject: [PATCH 2/3] docs(coordination): document canonical mode operation and
 migration scope

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 ...shared-goal-authority-state-provider-v0.md |  9 ++-
 ...-goal-authority-state-provider-v0.zh-CN.md |  7 +-
 .../typescript-control-plane-migration-v0.md  |  1 +
 ...script-control-plane-migration-v0.zh-CN.md |  1 +
 docs/reference/handoff-mode.md                | 81 +++++++++++++++++++
 5 files changed, 92 insertions(+), 7 deletions(-)
 create mode 100644 docs/reference/handoff-mode.md

diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
index 1acac3c526..e004accccd 100644
--- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
+++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
@@ -1544,10 +1544,10 @@ projects normalized snapshots, invokes those decisions, and reconstructs the
 provider-neutral `TransitionPlan`. The local lease-file transaction and the
 coordination executor therefore consume the same lease decisions; locking,
 source revalidation, file persistence, provider CAS, and receipt construction
-remain in their respective execution layers. Todo, terminal-fence, and
-handoff-mode decisions stay in the Python core until their own reviewed
-TypeScript cutovers; local holder/fence-close lock mechanics remain execution
-effects rather than provider contracts.
+remain in their respective execution layers. The initial extraction retained
+Todo, terminal-fence and handoff-mode decisions in Python; subsequent cutovers
+move them to their typed owners. Handoff quiescence now lives in
+`handoff_mode_policy.ts`. Local holder/fence-close locks remain execution effects.
 
 Keep three layers distinct as the provider work proceeds:
 
@@ -2894,6 +2894,7 @@ an equal-byte retry syncs file and directory before reporting `current`. Narrati
 canonical records stay intact. This converges the retained Python presentation/legacy
 input adapter; it adds no RPC or business state machine and does not change TS authority
 transactions, provider defaults, SQLite D2 or D3 promotion requirements.
+Canonical handoff-mode show/set no longer depend on Markdown frontmatter or local lease files. One TS transaction binds quiescence, mode and durable operation replay to the same revision, including sealed no-op intents. This adds a provider-neutral command boundary, not a provider default or whole-Goal cutover; frontmatter remains outside the Todo-section renderer. See [operation and recovery](../../reference/handoff-mode.md).
 
 T2 now commits a lease-free native Monitor observation and its independent
 successors in one canonical CAS/receipt; the route planner alone still grants
diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md
index ead4e87fd0..9f00ef062d 100644
--- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md
+++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md
@@ -1231,9 +1231,9 @@ transfer、release 则由 `task_lease_lifecycle_decision.ts` 的纯 seam 持有
 `authority_core` 只负责投影 normalized snapshot、调用这些 decision，再重建
 provider-neutral `TransitionPlan`。因此，本地 lease-file transaction 与 coordination
 executor 消费同一份 lease decision；加锁、source 重验、文件持久化、provider CAS 与
-receipt 构造仍分别属于各自 execution layer。Todo、terminal-fence 与 handoff-mode
-决策继续留在 Python core，直到各自经过 review 的 TypeScript cutover；本地 holder /
-fence-close 锁机制属于 execution effect，而不是 provider contract。
+receipt 构造仍分别属于各自 execution layer。初次抽取保留的 Todo、terminal-fence
+与 handoff-mode Python 决策由后续切片移入 typed owner；handoff 空闲判断现归属
+`handoff_mode_policy.ts`。本地 holder／fence-close 锁仍属于 execution effect。
 
 后续 provider 工作必须始终分开三层：
 
@@ -2288,6 +2288,7 @@ fenced 示例被当成真实任务、归档 end marker 后叙述进入历史、
 文件／目录同步，之后才报告 `current`。区域外正文和 canonical record 不被改写。
 这是永久 Python 展示／legacy 输入适配层的收敛：TS authority transaction、provider
 默认值、SQLite D2 与 D3 promotion 合同不变，不增加 RPC 或另一份业务状态机。
+Canonical handoff-mode show/set 不再依赖 Markdown frontmatter 或本地 lease；一笔 TS 事务把空闲检查、mode 与耐久操作回执绑定到同一 revision，包括未改值请求的回执。该命令边界不切换默认 provider、不晋升整 Goal；frontmatter 仍不属于 Todo-section renderer。操作与恢复见 [handoff-mode](../../reference/handoff-mode.md)。
 
 能力缺口 consumer 在 legacy/canonical 输入上共用 TS requirement/resolution owner，
 包括 quota 的 Monitor 能力分流。删除 Python missing-set 与 owner/repair 决策 builder，
diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
index 1e4568d154..a5fadfbdf7 100644
--- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
+++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
@@ -639,6 +639,7 @@ an equal-byte retry syncs file and directory before reporting `current`. Narrati
 canonical records stay intact. This converges the retained Python presentation/legacy
 input adapter; it adds no RPC or business state machine and does not change TS authority
 transactions, provider defaults, SQLite D2 or D3 promotion requirements.
+Handoff mode now shares a typed quiescence policy between the legacy adapter and one provider-neutral CAS/receipt transaction. Promoted show/set consume canonical mode and complete Todo/lease facts; the old Python transition decision is removed. Legacy state/lease locks remain until their last writer retires. See [handoff-mode operation and replay](../../reference/handoff-mode.md).
 
 Task-graph topology now shares `work_items/planning_relations.ts` with inventory
 and horizon. One pure TS request owns relationship discovery, deterministic
diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
index 11c58394f9..0ce2a496eb 100644
--- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
+++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
@@ -497,6 +497,7 @@ fenced 示例被当成真实任务、归档 end marker 后叙述进入历史、
 文件／目录同步，之后才报告 `current`。区域外正文和 canonical record 不被改写。
 这是永久 Python 展示／legacy 输入适配层的收敛：TS authority transaction、provider
 默认值、SQLite D2 与 D3 promotion 合同不变，不增加 RPC 或另一份业务状态机。
+Handoff mode 的 legacy adapter 与原生 CAS／receipt 事务现共用 TS 空闲判断；晋升后的 show/set 使用 canonical mode 和完整 Todo／lease 快照，删除 Python 切换决策。旧 state／lease 锁仍服务未晋升 writer，不能提前删除。操作与回放合同见 [handoff-mode](../../reference/handoff-mode.md)。
 
 Task graph topology 与 inventory/horizon 共用 `work_items/planning_relations.ts`。
 一轮纯 TS 请求拥有关系发现、稳定有界遍历、边去重与缺失/截断完整度；删除
diff --git a/docs/reference/handoff-mode.md b/docs/reference/handoff-mode.md
new file mode 100644
index 0000000000..c4e88847ce
--- /dev/null
+++ b/docs/reference/handoff-mode.md
@@ -0,0 +1,81 @@
+# Goal handoff mode
+
+`handoff-mode` chooses the ownership rule used by existing Todo/lease operations:
+`legacy` retains the claim/lease compatibility model, `soft_claim` uses the Todo
+claim, and `hard_lease` requires the existing lease fences. It is not an Agent
+capability grant, provider selector, or Goal promotion command.
+
+## Read and change
+
+```bash
+loopx handoff-mode show --goal-id example-goal --format json
+loopx handoff-mode set --goal-id example-goal --mode soft_claim --dry-run --format json
+loopx handoff-mode set --goal-id example-goal --mode soft_claim --format json
+```
+
+Before promotion, these commands use the existing frontmatter writer and its
+state/lease locks. After promotion, they use the selected canonical provider;
+`show` returns `source=canonical_provider` and its `provider_revision`, even if
+Markdown is stale or missing. `--runtime-root` applies to both show and set.
+Provider errors fail closed. A leftover local lease file cannot override an
+empty canonical lease collection.
+
+A mode change requires no unfinished claimed active Todo and no time-active
+lease. The canonical transaction checks the complete Todo/lease snapshot,
+including records outside display limits. An expiry equal to the observation
+time is expired; an invalid active lease timestamp or unknown lease schema
+cannot prove quiescence. Concurrent mutations invalidate the CAS snapshot and
+return a conflict without switching the mode. Todos, lease records and their
+read-model digests are preserved by the mode change.
+
+The unpromoted scan retains its older materialized-state scope: it does not
+claim to include event-only Todos. Its quiescence decision and the canonical
+transaction now share one typed policy. No default mode changes.
+
+## Recover a canonical request
+
+Choose an operation ID before a canonical set if a lost response must be retried:
+
+```bash
+loopx handoff-mode set --goal-id example-goal --mode soft_claim --operation-id mode-change-1 --format json
+# Repeat this exact intent to recover its original receipt.
+loopx handoff-mode set --goal-id example-goal --mode soft_claim --operation-id mode-change-1 --format json
+loopx handoff-mode show --goal-id example-goal --format json
+```
+
+The ID binds the goal and requested mode. Reuse with a different mode is rejected.
+A retry's clock may advance; it still recovers the original result. Even an
+accepted unchanged canonical set seals a receipt and advances provider revision,
+while returning `changed=false`. If another mode was selected afterward, replay
+returns the original decision without restoring it. Use `show` for current mode.
+Preview writes neither a mode nor an operation receipt. `--operation-id` requires
+canonical authority; the legacy writer does not promise durable operation replay.
+
+Select a previous mode with a **new** operation ID to change it back, subject to
+the same quiescence check. Do not disable the writer fence or restore old Markdown
+to roll back a canonical change. The existing Todo-section renderer does not
+project frontmatter: canonical mode is read through `handoff-mode show`, not a
+possibly old frontmatter value. This command does not qualify a provider profile,
+complete D1–D3, deploy PostgreSQL, or authorize active-Goal migration.
+
+## 中文
+
+`handoff-mode` 选择 Todo 的 claim／lease 所有权规则，不授予 capability、不选择
+provider，也不执行 Goal 晋升。上面的命令分别用于读取、预览和切换。
+
+晋升前保留 frontmatter 与本地锁兼容路径；晋升后从 canonical provider 读取，
+Markdown 缺失／陈旧和遗留本地 lease 不再影响判断。`show` 返回来源及 revision；
+provider 失败明确报错，不回退旧文件。现有 Todo-section 投影不包含 frontmatter，
+因此当前 mode 应通过 `show` 查询。
+
+切换要求完整快照内不存在未完成的已认领活动 Todo、不存在有效 lease。过期时间
+恰好等于观察时间视为已过期；非法有效期或未知 lease schema 不能作为空闲证据。
+并发修改使 CAS 冲突，不能在旧检查结果上继续切换。原 Todo、lease 和摘要不变。
+未晋升路径仍仅扫描物化状态，不宣称覆盖 event-only Todo；两条路径共用 TS 切换规则。
+
+需支持丢响应恢复时，在首次 canonical set 前指定 `--operation-id`，重试沿用同一
+目标 mode 和 ID。不同 mode 复用 ID 会被拒绝；即使最初 mode 未变，也记录耐久回执。
+若后来已切到其他 mode，旧请求重放只返回原回执，不把 mode 改回去；用 `show` 读当前值。
+预览不写入；旧 writer 不支持该幂等 ID。需要切回时，用新 ID 请求原 mode，仍须满足
+空闲门禁，不能通过关闭 fence 或恢复旧 Markdown 回滚。本功能不解除 provider
+默认值、长程资格化、PostgreSQL 部署或 D1–D3 的剩余条件。

From 9567086ebbf446c2429a729e4e6d53b4160146c1 Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Sun, 13 Sep 2026 13:55:18 +0800
Subject: [PATCH 3/3] fix(coordination): preserve legacy fence on provider
 outage

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 .../coordination/handoff_mode_transaction.ts  | 11 +++---
 loopx/control_plane/todos/handoff_mode.py     | 34 +++++++++++++++++--
 .../todos/provider_handoff_mode.py            |  9 ++++-
 .../test_shadow_writer_boundaries.py          | 33 ++++++++++++++++++
 .../handoff_mode_conformance.ts               |  5 +--
 5 files changed, 82 insertions(+), 10 deletions(-)

diff --git a/loopx/control_plane/coordination/handoff_mode_transaction.ts b/loopx/control_plane/coordination/handoff_mode_transaction.ts
index 371ca23845..d5b7feb15f 100644
--- a/loopx/control_plane/coordination/handoff_mode_transaction.ts
+++ b/loopx/control_plane/coordination/handoff_mode_transaction.ts
@@ -19,8 +19,9 @@ export interface HandoffModeSetInput {
   dry_run: boolean;
 }
 
-function failure(reason_code: string, reason: string): JsonObject {
-  return {schema_version: RESULT_SCHEMA, status: "failed", changed: false, reason_code, reason};
+function failure(reason_code: string, reason: string, failureKind?: "decision_rejection"): JsonObject {
+  return {schema_version: RESULT_SCHEMA, status: "failed", changed: false, reason_code, reason,
+    ...(failureKind ? {failure_kind: failureKind} : {})};
 }
 
 function replay(receipt: AuthorityStoreReceiptResult, input: HandoffModeSetInput, hash: string,
@@ -30,7 +31,8 @@ function replay(receipt: AuthorityStoreReceiptResult, input: HandoffModeSetInput
   const record = receipt.receipts[0];
   if (receipt.receipts.length !== 1 || record?.schema_version !== RECEIPT_SCHEMA ||
     record.goal_id !== input.goal_id || record.operation_id !== input.operation_id || record.request_sha256 !== hash) {
-    return failure("coordination_operation_identity_mismatch", "operation id names another handoff mode intent");
+    return failure("coordination_operation_identity_mismatch", "operation id names another handoff mode intent",
+      "decision_rejection");
   }
   const decision = canonicalAuthorityObject(record.decision, "handoff mode decision receipt");
   return {schema_version: RESULT_SCHEMA, ...decision, status,
@@ -77,7 +79,8 @@ export async function executeHandoffModeSet(store: AuthorityStore, raw: HandoffM
     decision = {goal_id: input.goal_id, operation_id: input.operation_id, previous_mode: previous,
       previous_mode_valid: true, handoff_mode: input.requested_mode, changed: plan.outcome === "apply"};
     if (plan.outcome === "rejected") return {...failure(String(plan.code),
-      "handoff_mode can only change without unfinished claimed Todos or time-active leases"),
+      "handoff_mode can only change without unfinished claimed Todos or time-active leases",
+      "decision_rejection"),
       ...decision, claimed_todos: claimed, active_leases: leases, provider_revision: loaded.provider_revision};
   } catch (error) { return failure("invalid_handoff_mode_authority", String(error)); }
   if (input.dry_run) return {schema_version: RESULT_SCHEMA, ...decision, status: "planned",
diff --git a/loopx/control_plane/todos/handoff_mode.py b/loopx/control_plane/todos/handoff_mode.py
index 08affa4c03..3bf3f9b71a 100644
--- a/loopx/control_plane/todos/handoff_mode.py
+++ b/loopx/control_plane/todos/handoff_mode.py
@@ -432,7 +432,15 @@ def set_goal_handoff_mode(
         runtime_root_from_registry,
         task_lease_lock_path,
     )
-    from ..coordination.legacy_writer_fence import legacy_todo_write_transaction
+    from ..coordination.legacy_writer_fence import (
+        LegacyCoordinationWriterFenced,
+        legacy_todo_write_transaction,
+        require_legacy_coordination_write_allowed,
+    )
+    from ..coordination.local_authority import (
+        LocalCoordinationAuthorityRejection,
+        LocalCoordinationAuthorityUnavailable,
+    )
     from ..coordination.runtime_shadow_writer_adapter import (
         write_captured_todo_state,
         begin_todo_runtime_shadow_capture,
@@ -448,8 +456,28 @@ def set_goal_handoff_mode(
     from .provider_handoff_mode import set_canonical_handoff_mode
 
     runtime_root = runtime_root_from_registry(registry_path, runtime_root_arg)
-    canonical = set_canonical_handoff_mode(runtime_root=runtime_root, goal_id=goal_id,
-        mode=requested, operation_id=operation_id, dry_run=dry_run)
+    try:
+        canonical = set_canonical_handoff_mode(
+            runtime_root=runtime_root,
+            goal_id=goal_id,
+            mode=requested,
+            operation_id=operation_id,
+            dry_run=dry_run,
+        )
+    except LocalCoordinationAuthorityRejection:
+        raise
+    except LocalCoordinationAuthorityUnavailable:
+        # A present legacy fence is the admission boundary for this caller.
+        # Re-check it when canonical dispatch is unavailable so an outage cannot
+        # turn a fenced legacy writer into an attempted Markdown mutation.
+        try:
+            require_legacy_coordination_write_allowed(
+                runtime_root=runtime_root,
+                goal_id=goal_id,
+            )
+        except LegacyCoordinationWriterFenced:
+            raise
+        raise
     if canonical is not None:
         return canonical
     if operation_id is not None:
diff --git a/loopx/control_plane/todos/provider_handoff_mode.py b/loopx/control_plane/todos/provider_handoff_mode.py
index 93a4c9bd05..1ae1c2b0fa 100644
--- a/loopx/control_plane/todos/provider_handoff_mode.py
+++ b/loopx/control_plane/todos/provider_handoff_mode.py
@@ -4,7 +4,8 @@
 from uuid import uuid4
 
 from ..coordination.local_authority import (
-    LOCAL_AUTHORITY_SOURCES, LocalCoordinationAuthorityUnavailable,
+    LOCAL_AUTHORITY_SOURCES, LocalCoordinationAuthorityRejection,
+    LocalCoordinationAuthorityUnavailable,
     local_authority_is_promoted, read_canonical_todos_if_promoted,
 )
 from ..effect_runtime import effect_runtime_result
@@ -33,6 +34,12 @@ def set_canonical_handoff_mode(*, runtime_root: Path, goal_id: str, mode: str,
         or result.get("source_authority") not in LOCAL_AUTHORITY_SOURCES
         or result.get("decision_read_from_provider") is not True or result.get("legacy_fallback_used") is not False):
         payload = result if isinstance(result, dict) else {}
+        if payload.get("status") == "failed" and payload.get("failure_kind") == "decision_rejection":
+            raise LocalCoordinationAuthorityRejection(
+                str(payload.get("reason") or "canonical handoff mode request was rejected"),
+                code=str(payload.get("reason_code") or "handoff_mode_rejected"),
+                payload=payload,
+            )
         raise LocalCoordinationAuthorityUnavailable(str(payload.get("reason") or "canonical mode unavailable"),
             code=str(payload.get("reason_code") or "handoff_mode_unavailable"), payload=payload)
     return {**result, "ok": True, "schema_version": "goal_handoff_mode_v0", "action": "set",
diff --git a/tests/control_plane/test_shadow_writer_boundaries.py b/tests/control_plane/test_shadow_writer_boundaries.py
index 3540eb1c76..294da6027d 100644
--- a/tests/control_plane/test_shadow_writer_boundaries.py
+++ b/tests/control_plane/test_shadow_writer_boundaries.py
@@ -69,6 +69,39 @@ def test_omitted_writers_refuse_a_fence_before_primary(
     assert not (root / "authority-shadow").exists()
 
 
+def test_handoff_mode_rechecks_legacy_fence_when_canonical_unavailable(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A provider outage must not bypass an already-present legacy fence."""
+
+    registry, state, root = fixture(tmp_path)
+    fence = legacy_coordination_writer_fence_path(runtime_root=root, goal_id=GOAL)
+    fence.parent.mkdir(parents=True)
+    fence.write_text("{}", encoding="utf-8")
+    monkeypatch.setattr(
+        "loopx.control_plane.todos.provider_handoff_mode.effect_runtime_result",
+        lambda *_args, **_kwargs: {
+            "status": "unavailable",
+            "reason_code": "canonical_provider_unavailable",
+            "reason": "canonical provider is unavailable",
+        },
+    )
+    monkeypatch.setattr(
+        "loopx.control_plane.coordination.legacy_writer_fence.effect_runtime_result",
+        lambda *_args, **_kwargs: {
+            "status": "blocked",
+            "reason_code": "legacy_coordination_writer_fenced",
+        },
+    )
+    before = state.read_bytes()
+    with pytest.raises(LegacyCoordinationWriterFenced) as error:
+        set_goal_handoff_mode(registry_path=registry, goal_id=GOAL, mode="soft_claim")
+    assert error.value.code == "legacy_coordination_writer_fenced"
+    assert error.value.payload["write_check"]["reason_code"] == "legacy_coordination_writer_fenced"
+    assert state.read_bytes() == before
+    assert not (root / "authority-shadow").exists()
+
+
 def test_corrupt_management_state_blocks_before_transaction_body(tmp_path: Path) -> None:
     registry, state, root = fixture(tmp_path)
     digest = hashlib.sha256(GOAL.encode()).hexdigest()[:16]
diff --git a/tests/control_plane_ts/handoff_mode_conformance.ts b/tests/control_plane_ts/handoff_mode_conformance.ts
index 7e3e88587a..0eccfeda9a 100644
--- a/tests/control_plane_ts/handoff_mode_conformance.ts
+++ b/tests/control_plane_ts/handoff_mode_conformance.ts
@@ -58,8 +58,9 @@ export function registerHandoffModeConformance(provider: string, factory: Author
       assert.equal(replay.changed, false);
       assert.deepEqual(await head(store), later);
     }
-    assert.equal((await executeHandoffModeSet(store, {...request, requested_mode: "hard_lease"})).reason_code,
-      "coordination_operation_identity_mismatch");
+    const mismatch = await executeHandoffModeSet(store, {...request, requested_mode: "hard_lease"});
+    assert.equal(mismatch.reason_code, "coordination_operation_identity_mismatch");
+    assert.equal(mismatch.failure_kind, "decision_rejection");
   });
 
   for (const kind of ["claim", "lease", "invalid_expiry", "unknown_lease_schema"] as const) {