fix(todo): reconcile explicit CAS after expired claim recovery

docs/reference/handoff-mode.md

@@ -84,6 +84,16 @@ For an eligible task edit without explicit proof, recovery now offers ordinary
 through inspect/acquire/release. Expired or replaced references remain rejected;
 recovery advice itself never grants execution or changes state.
 
+If the refused edit supplied `--update-expected-provider-revision`, reclaiming
+execution does not validate that old revision, even when the claim reports no
+Todo change. Recovery then gives a targeted `todo list --todo-id ...` step after
+claiming. Reconcile that Todo with the intended edit before filling the retry's
+claim reference and `authority_read.provider_revision` from the read. No full
+Goal reread is needed by this recovery contract. The CLI does not replace the
+CAS automatically; a later concurrent edit still rejects the retry. Keep the
+original operation id and reconciled intent. Calls without an explicit CAS
+retain the direct claim-and-retry path, and recovery never overrides a Stop.
+
 An expired hard-lease execution can be reclaimed with ordinary `todo claim`
 by an eligible registered Agent. When the expired holder still matches the Todo
 claim, the claim owner and a fresh execution generation change in one canonical
@@ -363,6 +373,13 @@ claim+lease 均不可执行，必须先审核迁移；合并式 claim+lease 仍
 提示不包含 execution key、不授予权限，也不修改 mode 或 provider。并发造成 version
 变化时重新读取，不能绕过 CAS；dry-run 和拒绝路径不产生 Todo 或租约写入。
 
+如果被拒绝的编辑提供了 `--update-expected-provider-revision`，重新认领不证明旧版本
+仍可用于编辑，即使认领回包显示 Todo 没有变化。恢复提示会在认领后给出只读取该 Todo
+的 `todo list --todo-id ...`。核对当前内容与原编辑意图后，再把新认领引用和这次读取的
+`authority_read.provider_revision` 填入重试命令；此恢复契约不要求重读整个 Goal。
+CLI 不会自动替换 CAS，读取后若再有并发修改，重试仍会拒绝。保留原 operation id 和
+核对后的意图。不带显式 CAS 的调用保持直接认领并重试，恢复也不会覆盖 Stop。
+
 修改已 deferred Todo 的等待条件若被拒绝，恢复命令先以 `--status open
 --clear-resume-when`、审核后的 reason、新 update operation 和当前 provider CAS
 显式恢复原任务；读回后，再用另一个 operation 和新的 CAS 提交审核后的延期。

loopx/cli_commands/todo_event.py

@@ -62,6 +62,21 @@ def _route_native_recovery(
         arguments = descriptor.get("cli_args")
         if descriptor.get("reference_source") == "claim_operation_id_from_claim":
             arguments = _original_todo_arguments(args)
+            required = descriptor.get("requires_flags")
+            if isinstance(required, list) and "--update-expected-provider-revision" in required:
+                # Core requires a post-claim targeted read. Project its pending
+                # references, not the old values or the claim's new revision.
+                for flag, placeholder in (
+                    ("--claim-operation-id", "<claim_operation_id_from_claim>"),
+                    ("--update-expected-provider-revision", "<provider_revision_from_reconciled_todo_read>"),
+                ):
+                    for index, argument in enumerate(arguments):
+                        if argument == flag and index + 1 < len(arguments):
+                            arguments[index + 1] = placeholder
+                            break
+                        if argument.startswith(flag + "="):
+                            arguments[index] = flag + "=" + placeholder
+                            break
         else:
             arguments = list(arguments) if isinstance(arguments, list) else command[1:]
         # The update adapter may already bind selectors. Inspect only the global

loopx/control_plane/coordination/claim_execution_recovery.ts

@@ -9,6 +9,7 @@ import {executeCoordinationTodoClaim, type CoordinationTodoClaimInput} from "./t
 type RecoveryInput = Pick<CoordinationTodoClaimInput, "goal_id" | "todo_id" | "actor_agent_id" |
   "registered_agents" | "expected_role" | "operation_id" | "now"> & {
     claim_operation_id: string; expected_provider_revision: string;
+    caller_expected_provider_revision?: string;
   };
 
 export async function claimExecutionRecovery(store: AuthorityStore, input: RecoveryInput,
@@ -35,6 +36,20 @@ export async function claimExecutionRecovery(store: AuthorityStore, input: Recov
   // An optional observation failure must not hide the original proof refusal.
   if (preview === null) return undefined;
   if (preview.dry_run === true && ["planned", "no_change"].includes(String(preview.status))) {
+    if (input.caller_expected_provider_revision !== undefined) {
+      // Claim advances authority even when the Todo body is unchanged. It does
+      // not prove that no other writer changed the Todo before the new claim.
+      // Retain the caller's CAS obligation; never silently rebase its edit.
+      return {action: "claim_task", lease_state: "expired", execution_authority_granted: false,
+        reason: "The original execution expired. Explicitly claim this Todo, then inspect this same Todo and reconcile its current content with the intended edit. A new claim does not validate the old provider revision. Only after reconciliation, retry with the returned claim_operation_id and authority_read.provider_revision from that inspection. Keep the operation intent and id; do not reuse either stale reference or assume unchanged content from the claim receipt. Claim and inspection grant no permission to overwrite intervening edits.",
+        claim: {command: "loopx todo claim", cli_args: ["todo", "claim", "--goal-id", input.goal_id,
+          "--todo-id", input.todo_id, "--agent-id", input.actor_agent_id, "--claimed-by", input.actor_agent_id]},
+        inspect: {command: "loopx todo list", cli_args: ["todo", "list", "--goal-id", input.goal_id,
+          "--todo-id", input.todo_id]},
+        retry: {command: `loopx todo ${command}`,
+          requires_flags: ["--claim-operation-id", "--update-expected-provider-revision"],
+          reference_source: "claim_operation_id_from_claim"}};
+    }
     return {action: "claim_task", lease_state: "expired", execution_authority_granted: false,
       reason: "The original execution expired. Explicitly run a new ordinary Todo claim, then retry the unchanged operation with its returned claim_operation_id. Do not reuse the expired claim id. Claim rechecks current ownership and eligibility; this preview grants no execution authority.",
       claim: {command: "loopx todo claim", cli_args: ["todo", "claim", "--goal-id", input.goal_id,

loopx/control_plane/coordination/todo_update.ts

@@ -225,6 +225,7 @@ export async function executeCoordinationTodoUpdate(
     try { input = {...input, ...await claimExecutionProof(store, head.head, input, claimReference), claim_operation_id: undefined}; }
     catch (error) {
       const recovery = await claimExecutionRecovery(store, {...input, claim_operation_id: claimReference,
+        caller_expected_provider_revision: input.expected_provider_revision,
         expected_provider_revision: head.provider_revision}, "update", error, authoritySourcesCurrent, observeWorkspace);
       return {...rejection("claim_execution_unavailable", error instanceof Error ? error.message : "Original task claim unavailable"),
         ...(recovery === undefined ? {} : {handoff_mode: head.head.handoff_mode, recovery})};

tests/control_plane/test_canonical_claim_execution_proof.py

@@ -104,7 +104,12 @@ def expired_managed_claim(tmp_path, monkeypatch, request):
         f"import {{prepareCoordinationProjectionCommit}} from {json.dumps((module / 'coordination_projection.ts').as_uri())};"
         "let text='';for await(const chunk of process.stdin)text+=chunk;const input=JSON.parse(text);"
         "const store=await openLocalAuthorityStore(input.runtime,input.goal);let result;"
-        "if(input.patch){const head=await store.loadAuthority();"
+        "if(input.todo_patch){const head=await store.loadAuthority();"
+        "const todo=head.head.todos.find(todo=>todo.todo_id===input.todo);"
+        "result=await store.commitAuthority(prepareCoordinationProjectionCommit({goal_id:input.goal,"
+        "operation_id:'synthetic-edit-'+head.provider_revision,expected_provider_revision:head.provider_revision,projection:head.head,"
+        "mutations:[{kind:'todo_upsert',todo:{...todo,...input.todo_patch}}]}));}"
+        "else if(input.patch){const head=await store.loadAuthority();"
         "result=await store.commitAuthority(prepareCoordinationProjectionCommit({goal_id:input.goal,"
         "operation_id:'synthetic-identity-change',expected_provider_revision:head.provider_revision,projection:head.head,"
         "mutations:[{kind:'lease_upsert',lease:{...head.head.leases[0],...input.patch}}]}));}"
@@ -114,9 +119,9 @@ def expired_managed_claim(tmp_path, monkeypatch, request):
         "process.stdout.write(JSON.stringify(result));"
     )
 
-    def author(patch=None):
+    def author(patch=None, *, todo_patch=None):
         result = subprocess.run(["node", "--no-warnings", "--experimental-strip-types", "--input-type=module", "-e", script],
-            input=json.dumps({"runtime": str(runtime), "goal": goal, "todo": target, "patch": patch}),
+            input=json.dumps({"runtime": str(runtime), "goal": goal, "todo": target, "patch": patch, "todo_patch": todo_patch}),
             capture_output=True, text=True, timeout=45, check=True)
         accepted = json.loads(result.stdout)
         assert accepted["status"] == "applied", accepted
@@ -196,6 +201,8 @@ def test_changed_execution_packet_never_recommends_claim_adoption(expired_manage
     before = snapshot()
     for command in ("update", "complete"):
         extra = ["--note", "Reviewed progress"] if command == "update" else ["--evidence", "evidence:output"]
+        if command == "update":
+            extra += ["--update-expected-provider-revision", before[1]]
         rejected = cli(["todo", command, *selected, "--agent-id", "agent-a", "--claim-operation-id",
                         "expired-original-claim", *extra], expected_exit=1)
         assert rejected["error_code"] == "claim_execution_unavailable"
@@ -205,3 +212,80 @@ def test_changed_execution_packet_never_recommends_claim_adoption(expired_manage
         assert "todo list" in render_todo_markdown(rejected)
         assert cli(recovery["inspect"]["cli_args"], use_fixture_route=False)["todo"] == before[0]
     assert snapshot() == before
+
+
+@pytest.mark.parametrize("intervening_edit", ["none", "before-claim", "after-read"])
+def test_expired_claim_with_cas_reconciles_only_selected_todo(
+    expired_managed_claim, intervening_edit, record_property,
+):
+    cli, selected, snapshot, author, _ = expired_managed_claim
+    original = snapshot()
+    attempt = ["todo", "update", *selected, "--agent-id", "agent-a", "--claim-operation-id",
+               "expired-original-claim", "--update-operation-id", "original-reviewed-edit",
+               "--update-expected-provider-revision", original[1], "--note", "Validated progress"]
+    before = snapshot()
+    refused = cli(attempt, expected_exit=1)
+    assert refused.get("error_code") == "claim_execution_unavailable", json.dumps(refused)
+    recovery = refused["recovery"]
+    assert recovery["action"] == "claim_task" and recovery["execution_authority_granted"] is False
+    assert snapshot() == before
+    assert recovery["inspect"]["cli_args"][-2:] == selected[-2:]
+    assert recovery["retry"]["requires_flags"] == ["--claim-operation-id", "--update-expected-provider-revision"]
+    retry = recovery["retry"]["cli_args"]
+    assert "expired-original-claim" not in retry
+    assert retry[retry.index("--update-expected-provider-revision") + 1] == "<provider_revision_from_reconciled_todo_read>"
+    rendered = render_todo_markdown(refused)
+    assert rendered.index("todo claim") < rendered.index("todo list") < rendered.index("todo update")
+    if intervening_edit == "before-claim":
+        author(todo_patch={"text": "Preserve another writer's accepted correction"})
+        before = snapshot()
+    prepared = cli(recovery["claim"]["cli_args"], use_fixture_route=False)
+    assert prepared["ok"] and prepared["todo_changed"] is False
+    # Claim only describes its own change, not changes since the old caller read.
+    current = cli(recovery["inspect"]["cli_args"], use_fixture_route=False)
+    assert current["todo"] == before[0]
+    revision = current["authority_read"]["provider_revision"]
+    assert revision != original[1]
+    retry = [prepared["claim_operation_id"] if value == "<claim_operation_id_from_claim>" else
+             revision if value == "<provider_revision_from_reconciled_todo_read>" else value for value in retry]
+    if intervening_edit == "after-read":
+        author(todo_patch={"text": "Preserve a concurrent correction after inspection"})
+        concurrent = snapshot()
+        rejected = cli(retry, expected_exit=1, use_fixture_route=False)
+        assert rejected["error_code"] == "provider_revision_mismatch"
+        assert snapshot() == concurrent, "recovery must not silently rebase a concurrent edit"
+        current = cli(recovery["inspect"]["cli_args"], use_fixture_route=False)
+        retry[retry.index("--update-expected-provider-revision") + 1] = current["authority_read"]["provider_revision"]
+    accepted = cli(retry, use_fixture_route=False)
+    assert accepted["ok"]
+    final = snapshot()
+    assert final[0]["text"] == current["todo"]["text"]
+    assert final[0]["note"] == "Validated progress"
+    assert final[0]["status"] == "open"
+    assert cli(retry, use_fixture_route=False)["ok"] and snapshot() == final
+    record_property("recovery_interaction", json.dumps({
+        "commands": ["todo update (refused)", "todo claim", "todo list --todo-id", "todo update (reconciled retry)"],
+        "whole_goal_reads": 0, "automatic_revision_rebase": False,
+        "intervening_edit": intervening_edit, "verification_reads_excluded": True,
+    }))
+
+
+@pytest.mark.parametrize("reference", ["old-reference", "-old-reference"])
+def test_cas_retry_projection_preserves_intent_and_routes_without_stale_references(tmp_path, reference):
+    from loopx.cli import build_parser
+    from loopx.cli_commands.todo_event import _route_native_recovery
+
+    args = build_parser().parse_args([
+        "todo", "update", "--goal-id", "example-goal", "--todo-id", "todo_example", "--agent-id", "agent-a",
+        "--claim-operation-id=" + reference, "--update-operation-id", "original-intent",
+        "--update-expected-provider-revision=" + reference, "--note=--update-expected-provider-revision",
+    ])
+    packet = {"recovery": {"retry": {"command": "loopx todo update",
+        "requires_flags": ["--claim-operation-id", "--update-expected-provider-revision"],
+        "reference_source": "claim_operation_id_from_claim"}}}
+    _route_native_recovery(packet, args, registry_path=tmp_path / "registry.json", runtime_root_arg=str(tmp_path / "runtime"))
+    retried = build_parser().parse_args(packet["recovery"]["retry"]["cli_args"])
+    assert retried.claim_operation_id == "<claim_operation_id_from_claim>"
+    assert retried.update_expected_provider_revision == "<provider_revision_from_reconciled_todo_read>"
+    assert retried.update_operation_id == "original-intent" and retried.note == "--update-expected-provider-revision"
+    assert retried.registry == str(tmp_path / "registry.json") and retried.runtime_root == str(tmp_path / "runtime")