feat(todo): expose promoted claim retry identity

docs/architecture/rfcs/typescript-control-plane-migration-v0.md

@@ -45,9 +45,15 @@ not renew a lease or assert current ownership. Successful non-preview
 may advance, but Todo state, `updated_at`, and domain events do not change.
 A structurally valid empty registration list permits historical replay, never
 a fresh claim; malformed lists still fail. Preview remains zero-write, and
-invalid preview booleans fail before provider access. The CLI creates a fresh operation id per invocation;
-cross-invocation retry identity and combined claim/lease acquisition remain
-follow-up work, not guarantees of this claim-only transaction.
+invalid preview booleans fail before provider access. The CLI still creates a
+fresh operation id by default. On an already promoted canonical authority,
+callers can opt into cross-invocation retry with
+`loopx todo claim --goal-id <goal> --todo-id <todo> --claimed-by <agent> --agent-id <agent> --claim-operation-id <public-safe-id>`.
+Reuse the same id and intent after a lost response; changed intent under that
+id fails closed. A preview does not consume the id. The option rejects legacy
+mode without writing or promoting anything; omit it to retain default behavior.
+It grants neither a lease nor current ownership on historical replay. Combined
+claim/lease acquisition remains follow-up work.
 
 The next replacement slice makes promoted `todo add` a native create
 transaction on that same authority owner. Python validates the established CLI

docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md

@@ -38,8 +38,13 @@ decision 处理 actor、registration、role、status、archive、exclusion 与
 同样在 head CAS 下持久化终态 receipt：存储 revision 可以前进，但 Todo 状态、
 `updated_at` 和 domain events 不变。结构合法的空注册名单允许历史回放，不能发起
 新 claim；非法名单仍失败。preview 保持零写入，非法 preview boolean 在访问 provider
-前失败。CLI 每次调用仍生成新的 operation id；跨调用重试身份和 claim/lease
-联合获取仍是后续工作，不能视为当前 claim-only 事务已提供的保证。
+前失败。CLI 默认仍为每次调用生成新 operation id。在已经 promotion 的 canonical
+authority 上，可显式使用
+`loopx todo claim --goal-id <goal> --todo-id <todo> --claimed-by <agent> --agent-id <agent> --claim-operation-id <public-safe-id>`
+进行跨进程重试：响应丢失后复用相同 id 和请求意图；同 id 搭配不同意图会失败。
+preview 不消耗该 id。legacy 模式会拒绝此选项，不写入也不自动 promotion；省略
+选项即可保持默认行为。历史 replay 不授予 lease 或当前所有权，claim/lease 联合
+获取仍是后续工作。
 
 下一 replacement slice 让 promotion 后的 `todo add` 成为同一 authority owner 上的
 原生 create transaction。Python 只校验既有 CLI 参数并一次性适配为带版本的 domain

loopx/cli_commands/todo.py

@@ -377,6 +377,15 @@ def register_todo_command(
         help="For capture-followups, append one public-safe agent follow-up todo. Repeat up to the requested batch.",
     )
     todo_parser.add_argument("--todo-id", help="Structured todo id from status/quota, such as todo_ab12cd34ef56.")
+    todo_parser.add_argument(
+        "--claim-operation-id",
+        help=(
+            "For todo claim on promoted canonical authority only, reuse this public-safe "
+            "operation id across retries. Changed intent with the same id is rejected; "
+            "receipt replay proves historical acceptance, not current lease ownership. "
+            "Omit to retain a fresh operation id per invocation."
+        ),
+    )
     todo_parser.add_argument(
         "--turn-instance-id",
         help=(
@@ -964,6 +973,7 @@ def handle_todo_command(
                 claimed_by=args.claimed_by,
                 agent_id=args.agent_id,
                 claim_only=True,
+                claim_operation_id=args.claim_operation_id,
                 **_todo_path_args(args),
                 dry_run=bool(args.dry_run),
             )

loopx/cli_commands/todo_argument_validation.py

@@ -10,6 +10,7 @@
     ("--text", "text"),
     ("--follow-up", "followups"),
     ("--todo-id", "todo_id"),
+    ("--claim-operation-id", "claim_operation_id"),
     ("--turn-instance-id", "turn_instance_id"),
     ("--completion-identity-key", "completion_identity_key"),
     ("--replan-obligation-id", "replan_obligation_id"),
@@ -332,9 +333,9 @@ def validate_todo_claim_options(args: argparse.Namespace) -> None:
         )
     _validate_todo_option_subset(
         args,
-        {"role", "todo_id", "claimed_by", "agent_id", "state_file"},
+        {"role", "todo_id", "claimed_by", "agent_id", "state_file", "claim_operation_id"},
         "todo claim only accepts --todo-id, --claimed-by, --agent-id, optional --role, "
-        "--project, --state-file, and --dry-run; unsupported: ",
+        "--claim-operation-id, --project, --state-file, and --dry-run; unsupported: ",
     )
 
 
@@ -499,6 +500,8 @@ def validate_shared_todo_options(args: argparse.Namespace) -> None:
         raise ValueError(
             "--turn-instance-id is supported only by todo complete settlement"
         )
+    if getattr(args, "claim_operation_id", None) is not None and args.todo_command != "claim":
+        raise ValueError("--claim-operation-id is supported only by todo claim")
     if (
         getattr(args, "completion_identity_key", None)
         and args.todo_command != "complete"

loopx/control_plane/coordination/local_authority.py

@@ -71,6 +71,7 @@ def claim_canonical_todo_if_promoted(
     claimed_by: str,
     actor_agent_id: str | None,
     dry_run: bool,
+    operation_id: str | None = None,
 ) -> dict[str, Any] | None:
     """Route a post-cutover claim to the TypeScript transaction owner."""
 
@@ -89,7 +90,7 @@ def claim_canonical_todo_if_promoted(
             "registered_agents": registered_agent_ids_from_registry(
                 registry_path, goal_id
             ),
-            "operation_id": f"todo-claim:{goal_id}:{todo_id}:{uuid4().hex}",
+            "operation_id": operation_id if operation_id is not None else f"todo-claim:{goal_id}:{todo_id}:{uuid4().hex}",
             "observed_at": now_local(),
             "dry_run": dry_run,
         },

loopx/todos.py

@@ -1087,6 +1087,7 @@ def update_goal_todo(
     enforce_monitor_boundedness: bool = True,
     clear_claim: bool = False,
     claim_only: bool = False,
+    claim_operation_id: str | None = None,
     project: Path | None = None,
     state_file: Path | None = None,
     dry_run: bool = False,
@@ -1104,9 +1105,15 @@ def update_goal_todo(
         raise ValueError(
             "todo update accepts either resume_when or clear_resume_when, not both"
         )
-    if claim_only and local_authority_is_promoted(
+    promoted_claim = claim_only and local_authority_is_promoted(
         runtime_root=shadow_runtime_root, goal_id=goal_id
-    ):
+    )
+    if claim_operation_id is not None:
+        if not claim_only:
+            raise ValueError("claim_operation_id is supported only by todo claim")
+        if not promoted_claim:
+            raise ValueError("--claim-operation-id requires promoted canonical authority; no legacy write attempted")
+    if promoted_claim:
         unsupported_claim_values = (
             text, status, note, evidence, reason, task_class, action_kind,
             task_domain, task_repository, continuation_policy,
@@ -1138,6 +1145,7 @@ def update_goal_todo(
             claimed_by=claimed_by or "",
             actor_agent_id=agent_id,
             dry_run=dry_run,
+            operation_id=claim_operation_id,
         )
         if canonical_claim is not None:
             return canonical_claim

tests/control_plane/test_local_coordination_authority.py

@@ -4,6 +4,7 @@
 import hashlib
 import subprocess
 import sys
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 
 import pytest
@@ -21,7 +22,7 @@
 from loopx.control_plane.coordination.legacy_writer_fence import (
     legacy_coordination_writer_fence_path,
 )
-from loopx.todos import add_goal_todo, list_goal_todos, update_goal_todo
+from loopx.todos import add_goal_todo, list_goal_todos
 
 
 def _engage_fence(runtime_root: Path, goal_id: str = "goal-a") -> None:
@@ -505,14 +506,23 @@ def test_real_shadow_projection_promotes_complete_complex_todo_semantics(
     assert by_id["todo_successor"]["completion_continuation"] == "no_followup"
     assert result["authority_read"]["todo_read_model"]["todo_count"] == 3
 
-    claimed = update_goal_todo(
-        registry_path=registry_path,
-        goal_id="goal-a",
-        todo_id="todo_claimable",
-        claimed_by="agent-a",
-        agent_id="agent-a",
-        claim_only=True,
-    )
+    claim_command = [
+        sys.executable, "-m", "loopx.cli", "--format", "json",
+        "--registry", str(registry_path), "todo", "claim", "--goal-id", "goal-a",
+        "--todo-id", "todo_claimable", "--claimed-by", "agent-a", "--agent-id", "agent-a",
+        "--claim-operation-id", "initial-cli-claim",
+    ]
+    # Duplicate callers race from separate processes, but one operation must
+    # produce exactly one accepted claim and the same durable receipt.
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        attempts = [pool.submit(subprocess.run, claim_command,
+            capture_output=True, text=True, check=True, timeout=30) for _ in range(2)]
+        claims = [json.loads(attempt.result().stdout) for attempt in attempts]
+    assert sum(item["status"] == "applied" for item in claims) == 1
+    assert all(item["status"] in {"applied", "recovered", "replayed"} for item in claims)
+    assert claims[0]["original_receipt"] == claims[1]["original_receipt"]
+    assert claims[0]["provider_revision"] == claims[1]["provider_revision"]
+    claimed = next(item for item in claims if item["status"] == "applied")
     assert claimed["ok"] is True
     assert claimed["source_authority"] == "file_v0"
     assert claimed["legacy_fallback_used"] is False
@@ -524,6 +534,32 @@ def test_real_shadow_projection_promotes_complete_complex_todo_semantics(
     )
     assert claimed_item["claimed_by"] == "agent-a"
 
+    # Separate CLI processes must replay one durable operation, not mint a
+    # fresh receipt for every retry. Preview does not consume that identity.
+    claim_command = [*claim_command[:-1], "retryable-cli-claim"]
+    preview = json.loads(subprocess.run(
+        [*claim_command, "--dry-run"], capture_output=True, text=True, check=True,
+    ).stdout)
+    assert preview["dry_run"] is True
+    original = json.loads(subprocess.run(
+        claim_command, capture_output=True, text=True, check=True,
+    ).stdout)
+    replay = json.loads(subprocess.run(
+        claim_command, capture_output=True, text=True, check=True,
+    ).stdout)
+    assert original["status"] == "no_change"
+    assert replay["status"] == "replayed"
+    assert replay["original_receipt"] == original["original_receipt"]
+    assert replay["provider_revision"] == original["provider_revision"]
+    changed_intent = ["agent-b" if part == "agent-a" else part for part in claim_command]
+    rejected = subprocess.run(changed_intent, capture_output=True, text=True)
+    assert rejected.returncode != 0
+    assert json.loads(rejected.stdout)["error"] == "operation id already names a different coordination request"
+    for invalid_key in ("", " padded-operation "):
+        invalid = subprocess.run([*claim_command[:-1], invalid_key], capture_output=True, text=True)
+        assert invalid.returncode != 0
+    assert not state_file.exists()
+
     create_command = [
         sys.executable, "-m", "loopx.cli", "--format", "json",
         "--registry", str(registry_path), "todo", "add", "--goal-id", "goal-a",

tests/control_plane/test_unpromoted_todo_claim.py

@@ -36,6 +36,9 @@ def test_unpromoted_claim_retains_markdown_semantics(
     if with_note:
         request["note"] = "legacy combined claim remains supported"
     before = state.read_bytes()
+    with pytest.raises(ValueError, match="requires promoted canonical authority"):
+        update_goal_todo(**request, claim_operation_id="explicit-retry")
+    assert state.read_bytes() == before
     preview = update_goal_todo(**request, dry_run=True)
     assert preview["ok"] is True
     assert preview["changed"] is True

tests/test_cli_argument_diagnostics.py

@@ -19,6 +19,7 @@
     validate_todo_suggest_options,
     validate_todo_supersede_options,
     validate_todo_update_options,
+    validate_shared_todo_options,
 )
 from loopx.control_plane.work_items.task_lease import TaskLeaseError
 
@@ -565,7 +566,7 @@ def test_quota_action_selection_requires_turn_identity() -> None:
                 "Continue the work.",
             ],
             "todo claim only accepts --todo-id, --claimed-by, --agent-id, optional --role, "
-            "--project, --state-file, and --dry-run; unsupported: "
+            "--claim-operation-id, --project, --state-file, and --dry-run; unsupported: "
             "--decision-outcome, --next-agent-todo",
         ),
     ],
@@ -584,6 +585,16 @@ def test_todo_claim_validation_preserves_exact_diagnostics(
     assert str(exc_info.value) == expected
 
 
+@pytest.mark.parametrize("command", ["add", "list", "update", "complete", "supersede"])
+@pytest.mark.parametrize("operation_id", ["retry-one", ""])
+def test_claim_operation_id_is_not_silently_ignored_by_other_commands(command, operation_id):
+    args = build_parser().parse_args([
+        "todo", command, "--goal-id", "example-goal", "--claim-operation-id", operation_id,
+    ])
+    with pytest.raises(ValueError, match="supported only by todo claim"):
+        validate_shared_todo_options(args)
+
+
 @pytest.mark.parametrize(
     ("extra_args", "expected"),
     [