fix(todo): accept managed claim references for supersede

docs/reference/handoff-mode.md

@@ -62,6 +62,28 @@ A read-only quota guard never claims or renews. Long-running delegated Hosts ret
 their existing supervision; this CLI path does not supervise an idle external
 process. The App's manual completion boundary described above is unchanged.
 
+To retire obsolete work after a managed self-claim, pass the same returned
+reference to `todo supersede` and name its replacement. For example, after the
+claim above, use this **instead of** completing `todo_example`:
+
+```bash
+loopx --format json todo supersede --goal-id example-goal --todo-id todo_example \
+  --agent-id worker-a --claim-operation-id work-a \
+  --reason 'Replace obsolete work' --next-agent-todo 'Deliver replacement work'
+```
+
+Use `--successor-todo-id <existing-todo-id>` instead of `--next-agent-todo` when
+the replacement already exists. Supersede retains the terminal owner's lease,
+acceptance and successor checks, retires the original execution atomically, and
+recovers the accepted receipt on an unchanged retry. An expired or replaced
+reference is rejected; inspect the current Todo and reconcile ownership before
+explicitly claiming again. Do not reuse the stale reference or copy lease counters.
+
+废弃已认领的任务时，可将自认领返回的 `claim_operation_id` 传给 `todo supersede`，
+并用 `--next-agent-todo` 创建替代任务，或用 `--successor-todo-id` 关联已有任务。
+租约、验收和后继检查仍由原终态 owner 执行；重试恢复原回执，不重复创建后继。
+过期或已换代的引用会拒绝，须先查看当前任务、核对归属，再显式重新认领。
+
 For a selected `hard_lease` Todo, the quota settlement plan now gives the ordinary
 claim command before task execution and carries its returned reference in the
 completion template. Quota and TurnEnvelope share that plan. The policy comes

loopx/cli_commands/todo.py

@@ -949,6 +949,7 @@ def handle_todo_command(
                 authority_reason=args.authority_reason,
                 task_lease_idempotency_key=args.task_lease_idempotency_key,
                 task_lease_expected_version=args.task_lease_expected_version,
+                claim_operation_id=args.claim_operation_id,
                 **_todo_path_args(args),
                 dry_run=bool(args.dry_run),
             )

loopx/cli_commands/todo_argument_validation.py

@@ -683,8 +683,8 @@ def validate_shared_todo_options(args: argparse.Namespace) -> None:
         raise ValueError("--update-operation-id is supported only by todo update")
     if getattr(args, "update_expected_provider_revision", None) is not None and args.todo_command != "update":
         raise ValueError("--update-expected-provider-revision is supported only by todo update")
-    if getattr(args, "claim_operation_id", None) is not None and args.todo_command not in {"claim", "update", "complete"}:
-        raise ValueError("--claim-operation-id is supported only by todo claim/update/complete")
+    if getattr(args, "claim_operation_id", None) is not None and args.todo_command not in {"claim", "update", "complete", "supersede"}:
+        raise ValueError("--claim-operation-id is supported only by todo claim/update/complete/supersede")
     if (
         getattr(args, "completion_identity_key", None)
         and args.todo_command != "complete"

loopx/cli_commands/todo_registration.py

@@ -75,10 +75,12 @@ def register_todo_command(
     todo_parser.add_argument(
         "--claim-operation-id",
         help=(
-            "For claim/update/complete on canonical authority, reuse this public-safe "
-            "operation id across retries. Changed intent with the same id is rejected; "
+            "For canonical claim, name the managed claim with this public-safe id "
+            "and reuse it across retries (omit for a fresh claim id). For "
+            "update/complete/supersede, pass the claim_operation_id returned by "
+            "your self-claim. 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."
+            "Do not combine a retained reference with explicit task-lease proof."
         ),
     )
     todo_parser.add_argument(

loopx/control_plane/todos/mutation_api.py

@@ -503,6 +503,7 @@ def supersede_goal_todo(
     authority_reason: str | None = None,
     task_lease_idempotency_key: str | None = None,
     task_lease_expected_version: int | None = None,
+    claim_operation_id: str | None = None,
     project: Path | None = None,
     state_file: Path | None = None,
     dry_run: bool = False,

tests/control_plane/test_canonical_claim_execution_proof.py

@@ -189,6 +189,24 @@ def test_expired_claim_packet_recovers_without_help_or_list(expired_managed_clai
     }))
 
 
+def test_supersede_rejects_expired_and_replaced_claim(expired_managed_claim):
+    cli, selected, snapshot, _, _ = expired_managed_claim
+    attempt = ["todo", "supersede", *selected, "--agent-id", "agent-a",
+               "--claim-operation-id", "expired-original-claim", "--reason", "Replace obsolete work",
+               "--next-agent-todo", "Deliver replacement work"]
+    before = snapshot()
+    rejected = cli(attempt, expected_exit=1)
+    assert rejected["error_code"] == "claim_execution_unavailable"
+    assert snapshot() == before
+    prepared = cli(["todo", "claim", *selected, "--agent-id", "agent-a", "--claimed-by", "agent-a",
+                    "--claim-operation-id", "replacement-claim"])
+    assert prepared["ok"] and prepared["execution_managed"]
+    current = snapshot()
+    rejected = cli(attempt, expected_exit=1)
+    assert rejected["error_code"] == "claim_execution_unavailable"
+    assert snapshot() == current
+
+
 @pytest.mark.parametrize("change", [
     {"owner": "agent-b"}, {"lease_epoch": 2, "version": 2},
     {"write_repository": "git:github.com/example/other"},

tests/control_plane/test_cli_action_help.py

@@ -40,6 +40,18 @@ def test_action_help_from_real_cli_with_unavailable_registry(tmp_path, command,
     assert not registry.exists()
 
 
+def test_supersede_help_explains_the_retained_claim_reference(capsys):
+    with pytest.raises(SystemExit) as exited:
+        build_parser().parse_args(["todo", "supersede", "--help"])
+    assert exited.value.code == 0
+    output = capsys.readouterr().out
+    declared = [line.strip().split()[0] for line in output.splitlines() if line.startswith("  --")]
+    assert "--claim-operation-id" in declared
+    assert "update/complete/supersede" in output
+    assert "--task-lease-idempotency-key" not in declared
+    assert "--task-lease-expected-version" not in declared
+
+
 def test_action_help_keeps_required_usage_and_lists_optional_grammar_once(capsys):
     parser = build_parser()
     command = ["todo", "update", "--goal-id", "fixture", "--todo-id", "todo_a", "--note", "progress"]

tests/control_plane/test_todo_claim_argument_recovery.py

@@ -222,6 +222,43 @@ def test_ordinary_claim_progress_completion_need_no_lease_flags(claim_cli):
     assert snapshot() == before
 
 
+def test_managed_claim_supersede_preserves_fences_and_replays_once(claim_cli):
+    _, run, snapshot, prefix, claim, _, _, _ = claim_cli
+    accepted = run([*prefix, *claim, "--claimed-by", "peer-a", "--claim-operation-id", "managed-claim"])
+    reference = ["--claim-operation-id", accepted["claim_operation_id"]]
+    supersede = [*prefix, "todo", "supersede", "--goal-id", GOAL, "--todo-id", TODO,
+                 "--reason", "Replace obsolete work", "--next-agent-todo", "Deliver replacement work"]
+    before = snapshot()
+    for actor, proof, code in (
+        ("peer-a", [], "lease_fence_required"),
+        ("peer-b", reference, "claim_execution_unavailable"),
+        ("peer-a", ["--claim-operation-id", "unrelated-claim"], "claim_execution_unavailable"),
+    ):
+        rejected = run([*supersede, "--agent-id", actor, *proof], expected_exit=1)
+        assert rejected["error_code"] == code
+        assert snapshot() == before
+    attempt = [*supersede, "--agent-id", "peer-a", *reference]
+    mixed = run([*attempt, "--task-lease-idempotency-key", "unrelated-lease",
+                 "--task-lease-expected-version", "1"], expected_exit=1)
+    assert mixed["ok"] is False and snapshot() == before
+    preview = run([*attempt, "--dry-run"])
+    assert preview["ok"] and preview["status"] == "planned"
+    assert snapshot() == before
+    applied = run(attempt)
+    assert applied["ok"] and applied["superseded"] and applied["provider_status"] == "applied"
+    final = snapshot()
+    todos = {todo["todo_id"]: todo for todo in final[0]}
+    assert len(todos) == 2
+    assert todos[TODO]["status"] == "done"
+    assert todos[todos[TODO]["superseded_by"]]["text"] == "Deliver replacement work"
+    lease = run([*prefix, "task-lease", "inspect", "--goal-id", GOAL, "--todo-id", TODO])
+    assert lease["lease"]["status"] == "released"
+    replay = run(attempt)
+    assert replay["ok"] and replay["idempotent_replay"]
+    assert snapshot() == final
+    assert run([*prefix, "task-lease", "inspect", "--goal-id", GOAL, "--todo-id", TODO]) == lease
+
+
 @pytest.fixture
 def selected_runtime(claim_cli, tmp_path, monkeypatch):
     _, _, _, _, _, _, project, _ = claim_cli