fix(todos): centralize snapshot refusals before retiring Python guards

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

@@ -99,6 +99,30 @@ conversion. Local CLI adoption at `db3672f3c` verifies a clean source manifest,
 qualified SQLite runtime, current known authority formats and healthy canonical
 contract readback. This does not certify every installed Host or D2.
 
+### Todo snapshot refusal: retained semantics, one native owner
+
+The next bounded T4 slice moves the live Todo bridge's two snapshot refusals to
+`coordination/todo_lifecycle_decision.ts`. Claim/update and complete/supersede
+now share `invalid_lease_snapshot` before `todo_not_found`, actor admission and
+terminal replay. An active lease cannot be absent or released; an explicit null
+Todo is a domain refusal. Missing/malformed wire fields still fail strict
+decoding. Direct native callers gain the protection previously confined to the
+Python bridge; normalized Python outcomes and refusal precedence stay intact.
+
+| Retired symbol | Caller / replacement | Retained obligation and rollback |
+| --- | --- | --- |
+| Python `_invalid_lease_snapshot`, `_result` and the bridge's early missing-Todo decision | `authority_core.decide` / `_typescript_todo_decision`; existing terminal/mutation RPCs reach the same TS owner | Preserve rejection codes, no proposed write/release, actor/delegation/CAS rules and valid holder behavior. Keep the live Python snapshot/result adapter used by `todos/mutation_authority.py`, including native result-shape validation. Revert the code package without state conversion. |
+
+Independent negative cases cover all four verbs and three handoff modes,
+including absent Todo, done replay and an unauthorized actor. The baseline
+Python bridge already rejects these states; direct baseline TS does not.
+Qualification also exercises real File/SQLite creation, update, terminal
+effects, original-operation recovery, unavailable-provider refusal and leased
+settlement in isolated source and wheel installations. These are synthetic
+entrypoint tests, not attached-App or sustained-cost qualification. No store,
+writer, outbox, historical format/receipt reader or permanent Host IO is retired;
+C1, D2 and release-default adoption remain separate acceptance boundaries.
+
 ## Current closeout: validation, migration and deletion (2026-10-02)
 
 Rechecked against main `8b5335a72` and the linked PR heads. This is the current

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

@@ -86,6 +86,27 @@ Host IO 保留。这是最后调用方切片，不代表整项 C1／M4、D2 或
 `db3672f3c`，验证了干净源码清单、具备资格的 SQLite runtime、已知权威格式均为
 当前版本及健康的 canonical 合同读回。这不证明所有已安装 Host 或 D2 已验收。
 
+### Todo snapshot 拒绝：保留语义，统一原生 owner
+
+下一批有界 T4 切片把活跃 Todo bridge 的两项 snapshot 拒绝迁入
+`coordination/todo_lifecycle_decision.ts`。claim／update 和 complete／supersede
+共享拒绝顺序：`invalid_lease_snapshot`、`todo_not_found`，然后才是 actor
+准入和终态重放。active lease 不能同时 absent 或 released；显式 null Todo
+属于领域拒绝。缺失／格式错误的 wire 字段仍严格解码失败。直接 native
+调用获得此前只在 Python bridge 中生效的保护；合法归一化 Python 输入的
+结果和拒绝优先级保持。
+
+| 退役符号 | 调用方／替代 owner | 保留义务与回退 |
+| --- | --- | --- |
+| Python `_invalid_lease_snapshot`、`_result` 及 bridge 的缺失 Todo 提前判断 | `authority_core.decide`／`_typescript_todo_decision`；既有 terminal／mutation RPC 到同一 TS owner | 保留错误码、无写入／释放提案、actor／delegation／CAS 规则和合法 holder 行为。`todos/mutation_authority.py` 使用的 Python snapshot／结果适配及 native 结果形状验证保留；回退代码包即可，无需转换状态。 |
+
+独立负例覆盖四个 verb、三种 handoff mode，以及缺失 Todo、done 重放和未授权
+actor。基线 Python bridge 已拒绝这些状态，直接基线 TS 尚未如此。
+验收还在隔离源码和 wheel 安装中运行真实 File／SQLite 创建、修改、终态效果、
+原操作恢复、provider 不可用拒绝和租约结算。这是合成入口验收，不代表 attached
+App 或持续成本资格。未退役 store、writer、outbox、历史格式／回执 reader 或永久
+Host IO；C1、D2 和发布默认采用仍是独立验收边界。
+
 ## 当前收尾：验证、迁移与删除（2026-10-02）
 
 按 main `8b5335a72` 和所列 PR head 重新核对。本节是 **R5 / D1–D3 / T0–T4**

loopx/control_plane/coordination/authority_core.py

@@ -126,37 +126,6 @@ class TransitionPlan:
     idempotent: bool = False
 
 
-def _result(
-    outcome: DecisionOutcome,
-    code: str,
-    *,
-    next_snapshot: CoordinationSnapshot | None = None,
-    authority_mode: str | None = None,
-    ownership_gate: OwnershipGate = OwnershipGate.NOT_REQUIRED,
-    lease_fence: LeaseFence = LeaseFence.NOT_REQUIRED,
-    idempotent: bool = False,
-) -> TransitionPlan:
-    return TransitionPlan(
-        outcome=outcome,
-        code=code,
-        next_snapshot=next_snapshot,
-        authority_mode=authority_mode,
-        ownership_gate=ownership_gate,
-        lease_fence=lease_fence,
-        idempotent=idempotent,
-    )
-
-
-def _invalid_lease_snapshot(lease: LeaseSnapshot | None) -> bool:
-    """Reject contradictory normalized states at the pure-core boundary."""
-
-    return bool(
-        lease is not None
-        and lease.active
-        and (not lease.present or lease.status == "released")
-    )
-
-
 def write_scopes_overlap(
     left: tuple[str, ...] | list[str],
     right: tuple[str, ...] | list[str],
@@ -250,8 +219,6 @@ def _typescript_todo_decision(
     """Project the typed lifecycle decision; storage and locks stay with callers."""
 
     todo = snapshot.todo
-    if todo is None:
-        return _result(DecisionOutcome.REJECTED, "todo_not_found")
     terminal = command.action in {TodoAction.COMPLETE, TodoAction.SUPERSEDE}
     operation = "terminal" if terminal else "mutation"
     payload = effect_runtime_result(
@@ -269,7 +236,7 @@ def _typescript_todo_decision(
                 }
                 for grant in snapshot.lifecycle_grants
             ],
-            "todo": _todo_fact_payload(todo),
+            "todo": _todo_fact_payload(todo) if todo is not None else None,
             "decision_target": (
                 _todo_fact_payload(snapshot.decision_target)
                 if snapshot.decision_target is not None
@@ -302,7 +269,9 @@ def _typescript_todo_decision(
         ) from exc
     next_snapshot = None
     if outcome is DecisionOutcome.APPLY:
-        if payload.get("next_todo_status") != ("done" if terminal else todo.status):
+        if todo is None or payload.get("next_todo_status") != (
+            "done" if terminal else todo.status
+        ):
             raise RuntimeError(
                 "TypeScript Todo lifecycle decision returned invalid next Todo state"
             )
@@ -372,8 +341,6 @@ def decide(
 ) -> TransitionPlan:
     """Evaluate one normalized command without reading or writing state."""
 
-    if _invalid_lease_snapshot(snapshot.lease):
-        return _result(DecisionOutcome.REJECTED, "invalid_lease_snapshot")
     if isinstance(command, TodoMutationCommand):
         return _typescript_todo_decision(snapshot, command)
     raise TypeError(f"unsupported coordination command: {type(command).__name__}")

loopx/control_plane/coordination/todo_lifecycle_decision.ts

@@ -262,7 +262,8 @@ function executorReclaimGrant(request: JsonObject, actor: string | null): Lifecy
   return [{agent_id: agent, actions: [EXECUTOR_RECLAIM_ACTION], requires_reason: false}];
 }
 
-function decodeRequest(value: unknown, kind: "terminal" | "mutation" = "terminal"): LifecycleDecisionRequest {
+function decodeRequest(value: unknown, kind: "terminal" | "mutation" = "terminal"):
+    LifecycleDecisionRequest | CoordinationTodoTerminalDecisionResult {
   const request = requireJsonObject(value, "Todo terminal decision request");
   requireStringLiteral(
     request.schema_version,
@@ -276,14 +277,14 @@ function decodeRequest(value: unknown, kind: "terminal" | "mutation" = "terminal
   const actor = optionalAgent(request.actor_agent_id, "actor_agent_id");
   const internalReclaim = kind === "mutation" && request.authority_action === EXECUTOR_RECLAIM_ACTION;
   const outcome = kind === "terminal" ? optionalString(request.decision_outcome, "decision_outcome") : null;
-  return {
+  const decoded = {
     command: requireStringLiteral(request.command,
       kind === "terminal" ? COMMANDS : MUTATION_COMMANDS, "command"),
     handoff_mode: requireStringLiteral(request.handoff_mode, HANDOFF_MODES, "handoff_mode"),
     registered_agents: registeredAgents,
     lifecycle_grants: internalReclaim ? executorReclaimGrant(request, actor)
       : lifecycleGrants(request.lifecycle_grants ?? [], registeredAgents),
-    todo: todoFact(request.todo, "todo"),
+    todo: request.todo === null ? null : todoFact(request.todo, "todo"),
     decision_target: kind !== "terminal" || request.decision_target === null || request.decision_target === undefined
       ? null
       : todoFact(request.decision_target, "decision_target"),
@@ -316,6 +317,13 @@ function decodeRequest(value: unknown, kind: "terminal" | "mutation" = "terminal
     ownership_mutation: kind === "mutation"
       ? requireBoolean(request.ownership_mutation, "ownership_mutation") : false,
   };
+  // Decode the wire first. For normalized snapshots these refusals precede
+  // actor admission and terminal replay, including on direct native calls.
+  if (decoded.lease?.active && (!decoded.lease.present || decoded.lease.status === "released")) {
+    return result("rejected", "invalid_lease_snapshot");
+  }
+  if (decoded.todo === null) return result("rejected", "todo_not_found");
+  return {...decoded, todo: decoded.todo};
 }
 
 function scopeKey(scope: DecisionScope): string {
@@ -548,6 +556,7 @@ export function evaluateCoordinationTodoTerminalDecision(
   value: unknown,
 ): CoordinationTodoTerminalDecisionResult {
   const request = decodeRequest(value);
+  if ("outcome" in request) return request;
   const authorityResult = authority(request);
   if ("outcome" in authorityResult) return authorityResult;
   if (request.todo.status === "done") {
@@ -581,6 +590,9 @@ export function evaluateTodoOwnershipGate(value: unknown): JsonObject {
 /** Claim/update admission only: this neither edits arbitrary fields nor releases a lease. */
 export function evaluateCoordinationTodoMutationDecision(value: unknown): JsonObject {
   const request = decodeRequest(value, "mutation");
+  if ("outcome" in request) {
+    return {...request, schema_version: COORDINATION_TODO_MUTATION_DECISION_RESULT_SCHEMA};
+  }
   const decided = authority(request);
   if ("outcome" in decided) {
     return { ...decided, schema_version: COORDINATION_TODO_MUTATION_DECISION_RESULT_SCHEMA };

tests/control_plane/test_coordination_authority_core.py

@@ -85,6 +85,56 @@ def terminal(
     return TodoMutationCommand(**values)  # type: ignore[arg-type]
 
 
+@pytest.mark.parametrize("mode", list(HandoffMode))
+@pytest.mark.parametrize("action", list(TodoAction))
+@pytest.mark.parametrize("invalid", [lease(present=False), lease(status="released")])
+@pytest.mark.parametrize("target", [None, todo(), todo(status="done")])
+@pytest.mark.parametrize("actor", [AGENT_A, "unknown"])
+def test_contradictory_lease_rejection_precedes_missing_todo_authority_and_replay(
+    mode, action, invalid, target, actor
+):
+    state = snapshot(handoff_mode=mode, lease=invalid, todo=target)
+    # Snapshot corruption cannot become an ordinary actor denial or a replay.
+    result = decide(state, TodoMutationCommand(action=action, actor_agent_id=actor))
+    assert result.outcome is DecisionOutcome.REJECTED
+    assert result.code == "invalid_lease_snapshot"
+    assert result.next_snapshot is None
+    assert result.authority_mode is None
+    assert result.ownership_gate is OwnershipGate.NOT_REQUIRED
+    assert result.lease_fence is LeaseFence.NOT_REQUIRED
+    assert not result.idempotent
+
+
+@pytest.mark.parametrize("mode", list(HandoffMode))
+@pytest.mark.parametrize("action", list(TodoAction))
+def test_missing_todo_is_a_typed_refusal_before_actor_admission(mode, action):
+    result = decide(
+        snapshot(handoff_mode=mode, todo=None),
+        TodoMutationCommand(action=action, actor_agent_id="unknown"),
+    )
+    assert result.outcome is DecisionOutcome.REJECTED
+    assert result.code == "todo_not_found"
+    assert result.next_snapshot is None
+    assert result.authority_mode is None
+    assert not result.idempotent
+
+
+@pytest.mark.parametrize("mode", list(HandoffMode))
+@pytest.mark.parametrize(
+    "holder",
+    [None, lease(active=False, status="expired"), lease(active=False, status="released"), lease()],
+)
+def test_snapshot_guard_keeps_claim_neutral_edits_and_holder_state(mode, holder):
+    state = snapshot(handoff_mode=mode, lease=holder)
+    result = decide(
+        state, TodoMutationCommand(action=TodoAction.UPDATE, actor_agent_id=AGENT_A)
+    )
+    assert result.outcome is DecisionOutcome.APPLY
+    assert result.next_snapshot.todo == state.todo
+    assert result.next_snapshot.lease == holder
+    assert result.lease_fence is LeaseFence.NOT_REQUIRED
+
+
 @pytest.mark.parametrize("mode", list(HandoffMode))
 @pytest.mark.parametrize("ownership", [False, True])
 def test_update_authority_keeps_claim_neutral_edits_separate_from_ownership(

tests/control_plane_ts/todo_terminal_decision.test.ts

@@ -3,8 +3,10 @@ import test from "node:test";
 
 import {
   COORDINATION_TODO_TERMINAL_DECISION_REQUEST_SCHEMA,
+  COORDINATION_TODO_TERMINAL_DECISION_RESULT_SCHEMA,
   evaluateCoordinationTodoTerminalDecision,
   COORDINATION_TODO_MUTATION_DECISION_REQUEST_SCHEMA,
+  COORDINATION_TODO_MUTATION_DECISION_RESULT_SCHEMA,
   COORDINATION_TERMINAL_FENCE_REQUEST_SCHEMA,
   evaluateCoordinationTodoMutationDecision,
   evaluateCoordinationTerminalFence,
@@ -52,6 +54,70 @@ function mutation(overrides: Record<string, unknown> = {}) {
   });
 }
 
+test("normalized snapshot corruption precedes Todo absence, actor denial and replay for every verb", () => {
+  const holder = {present: true, active: true, status: "active", owner: "agent-a",
+    idempotency_key: "holder", version: 3, lease_epoch: 1, write_scopes: []};
+  for (const mode of ["legacy", "soft_claim", "hard_lease"]) {
+    for (const command of ["claim", "update", "complete", "supersede"]) {
+      const terminal = command === "complete" || command === "supersede";
+      const build = terminal ? request : mutation;
+      const evaluate = terminal ? evaluateCoordinationTodoTerminalDecision : evaluateCoordinationTodoMutationDecision;
+      for (const lease of [{...holder, present: false}, {...holder, status: "released"}]) {
+        for (const todo of [request().todo, {...request().todo, status: "done"}, null]) {
+          for (const actor of ["agent-a", "unknown"]) {
+            const result = evaluate(build({command, authority_action: command, handoff_mode: mode,
+              lease, todo, actor_agent_id: actor}));
+            assert.equal(result.schema_version, terminal ? COORDINATION_TODO_TERMINAL_DECISION_RESULT_SCHEMA
+              : COORDINATION_TODO_MUTATION_DECISION_RESULT_SCHEMA);
+            assert.equal(result.outcome, "rejected");
+            assert.equal(result.code, "invalid_lease_snapshot");
+            assert.equal(result.authority_mode, null);
+            assert.equal(result.ownership_gate, "not_required");
+            assert.equal(result.lease_fence, "not_required");
+            assert.equal(result.idempotent, false);
+            assert.equal(result.next_todo_status, null);
+            assert.equal(result.next_lease, null);
+          }
+        }
+      }
+    }
+  }
+});
+
+test("explicit missing Todo is a typed refusal; missing or malformed wire fields remain errors", () => {
+  for (const mode of ["legacy", "soft_claim", "hard_lease"]) {
+    for (const command of ["claim", "update", "complete", "supersede"]) {
+      const terminal = command === "complete" || command === "supersede";
+      const build = terminal ? request : mutation;
+      const evaluate = terminal ? evaluateCoordinationTodoTerminalDecision : evaluateCoordinationTodoMutationDecision;
+      const result = evaluate(build({command, authority_action: command, handoff_mode: mode,
+        todo: null, actor_agent_id: "unknown"}));
+      assert.equal(result.outcome, "rejected");
+      assert.equal(result.code, "todo_not_found");
+      assert.equal(result.idempotent, false);
+      assert.equal(result.next_todo_status, null);
+      for (const todo of [undefined, {}, "missing"]) assert.throws(() => evaluate(build({todo})));
+      assert.throws(() => evaluate(build({todo: null, command: "not-a-command"})));
+      assert.throws(() => evaluate(build({todo: null, handoff_mode: "not-a-mode"})));
+      assert.throws(() => evaluate(build({todo: null, lease: {active: "true"}})));
+    }
+  }
+});
+
+test("snapshot admission preserves inactive holders and claim-neutral mutation effects", () => {
+  const holder = {present: true, active: true, status: "active", owner: "agent-a",
+    idempotency_key: "holder", version: 3, lease_epoch: 1, write_scopes: []};
+  for (const mode of ["legacy", "soft_claim", "hard_lease"]) {
+    for (const lease of [null, holder, {...holder, active: false, status: "expired"},
+      {...holder, active: false, status: "released"}]) {
+      const result = evaluateCoordinationTodoMutationDecision(mutation({handoff_mode: mode, lease}));
+      assert.equal(result.outcome, "apply");
+      assert.equal(result.next_todo_claimed_by, "agent-a");
+      assert.equal(result.next_lease, null, "admission does not edit a lease");
+    }
+  }
+});
+
 test("bound User action closure is administrative, not a fabricated execution lease", () => {
   const todo = {...request().todo, role: "user", task_class: "user_action",
     claimed_by: null, bound_agent: "agent-a"};