fix(todo): recover missing execution claims in one command

loopx/cli_commands/todo.py

@@ -3,13 +3,16 @@
 import argparse
 import shlex
 from collections.abc import Callable, Sequence
+from functools import partial
 from operator import itemgetter
 from pathlib import Path
 
 from ..command_invocation import loopx_cli_argv
 from ..capabilities.todo_replan_cadence.machine_defaults import resolve_todo_replan_cadence_goal
 from ..control_plane.work_items.replan_history_codec import effective_turn_cadence_context
 from ..control_plane.coordination.local_authority import (
+    LocalCoordinationAuthorityRejection,
+    LocalCoordinationAuthorityUnavailable,
     local_authority_is_promoted,
     read_canonical_todo_fields_if_promoted,
 )
@@ -65,6 +68,7 @@
     validate_todo_archive_completed_options,
     validate_todo_claim_options,
     validate_todo_complete_options,
+    validate_todo_execution_recovery_options,
     validate_todo_receipt_options,
     validate_todo_result_read_options,
     validate_todo_project_markdown_options,
@@ -276,6 +280,105 @@ def _todo_path_args(args: argparse.Namespace) -> dict[str, Path | None]:
     }
 
 
+def _recover_todo_execution_once(
+    args: argparse.Namespace,
+    initial_error: LocalCoordinationAuthorityRejection,
+    *,
+    registry_path: Path,
+    runtime_root_arg: str | None,
+    retry_operation: Callable[[str], dict[str, object]],
+) -> dict[str, object]:
+    """Consume only the typed, claim-only lifecycle recovery in one CLI call.
+
+    The TypeScript lifecycle owner decides whether a new ordinary claim is a
+    legal repair.  This adapter neither infers eligibility from prose nor
+    replaces explicit proof.  Recovery that requires reconciliation/inspection
+    remains visible to the caller instead of being silently rebased.
+    """
+
+    recovery = initial_error.payload.get("recovery")
+    if not isinstance(recovery, dict):
+        raise initial_error
+    claim = recovery.get("claim")
+    retry = recovery.get("retry")
+    if not (
+        recovery.get("action") == "claim_task"
+        and recovery.get("execution_authority_granted") is False
+        and isinstance(claim, dict)
+        and isinstance(retry, dict)
+        and retry.get("reference_source") == "claim_operation_id_from_claim"
+        and retry.get("requires_flags") == ["--claim-operation-id"]
+        and "inspect" not in recovery
+    ):
+        raise initial_error
+
+    prepared = update_goal_todo(
+        registry_path=registry_path,
+        runtime_root_arg=runtime_root_arg,
+        goal_id=args.goal_id,
+        todo_id=args.todo_id,
+        role=args.role,
+        claimed_by=args.agent_id,
+        agent_id=args.agent_id,
+        claim_only=True,
+        manage_execution=True,
+        **_todo_path_args(args),
+        dry_run=False,
+    )
+    claim_operation_id = str(prepared.get("claim_operation_id") or "").strip()
+    if not (
+        prepared.get("ok") is True
+        and prepared.get("execution_managed") is True
+        and claim_operation_id
+    ):
+        raise LocalCoordinationAuthorityUnavailable(
+            "ordinary Todo claim returned without a usable execution reference",
+            code="todo_execution_recovery_claim_incomplete",
+            payload={
+                "execution_recovery": {
+                    "schema_version": "todo_execution_recovery_v0",
+                    "status": "claim_incomplete",
+                    "initial_error_code": initial_error.code,
+                    "claim_status": prepared.get("status"),
+                }
+            },
+        )
+    try:
+        result = retry_operation(claim_operation_id)
+    except Exception as retry_error:
+        retry_code = getattr(retry_error, "code", type(retry_error).__name__)
+        retry_payload = getattr(retry_error, "payload", None)
+        raise LocalCoordinationAuthorityUnavailable(
+            "Todo execution claim committed but the unchanged lifecycle retry failed",
+            code="todo_execution_recovery_retry_failed",
+            payload={
+                "execution_recovery": {
+                    "schema_version": "todo_execution_recovery_v0",
+                    "status": "claim_committed_retry_failed",
+                    "initial_error_code": initial_error.code,
+                    "claim_operation_id": claim_operation_id,
+                    "claim_status": prepared.get("status"),
+                    "retry_error_code": str(retry_code),
+                },
+                **(
+                    {"retry_failure": dict(retry_payload)}
+                    if isinstance(retry_payload, dict)
+                    else {}
+                ),
+            },
+        ) from retry_error
+    result["execution_recovery"] = {
+        "schema_version": "todo_execution_recovery_v0",
+        "status": "recovered",
+        "initial_error_code": initial_error.code,
+        "claim_operation_id": claim_operation_id,
+        "claim_status": prepared.get("status"),
+        "external_invocations": 1,
+        "external_failed_calls": 0,
+    }
+    return result
+
+
 def _render_todo_receipt(payload: dict[str, object]) -> str:
     lines = [
         "# LoopX Canonical Operation Receipt",
@@ -330,6 +433,7 @@ def handle_todo_command(
         if args.todo_command == "claim":
             validate_todo_claim_options(args)
         validate_shared_todo_options(args)
+        validate_todo_execution_recovery_options(args)
         validate_capability_gap_options(args)
         if getattr(args, "turn_instance_id", None):
             goal_instance_id = str(
@@ -524,12 +628,12 @@ def handle_todo_command(
                     payload.pop(internal, None)
         elif args.todo_command == "update":
             validate_todo_update_options(args)
-            payload = update_goal_todo(
+            update_operation = partial(
+                update_goal_todo,
                 registry_path=registry_path,
                 runtime_root_arg=runtime_root_arg,
                 goal_id=args.goal_id,
                 todo_id=args.todo_id,
-                claim_operation_id=args.claim_operation_id,
                 text=args.text,
                 priority=args.priority,
                 clear_priority=args.clear_priority,
@@ -592,6 +696,22 @@ def handle_todo_command(
                 **_todo_path_args(args),
                 dry_run=bool(args.dry_run),
             )
+            try:
+                payload = update_operation(
+                    claim_operation_id=args.claim_operation_id,
+                )
+            except LocalCoordinationAuthorityRejection as initial_error:
+                if not args.recover_execution:
+                    raise
+                payload = _recover_todo_execution_once(
+                    args,
+                    initial_error,
+                    registry_path=registry_path,
+                    runtime_root_arg=runtime_root_arg,
+                    retry_operation=lambda reference: update_operation(
+                        claim_operation_id=reference,
+                    ),
+                )
         elif args.todo_command == "complete":
             validate_todo_complete_options(args)
             settlement_result = None
@@ -688,12 +808,12 @@ def handle_todo_command(
                 completion_turn_key = str(args.completion_identity_key)
                 completion_identity_source = "lifecycle_reentry"
             if completion_error is None:
-                payload = complete_goal_todo(
+                complete_operation = partial(
+                    complete_goal_todo,
                     registry_path=registry_path,
                     runtime_root_arg=runtime_root_arg,
                     goal_id=args.goal_id,
                     todo_id=args.todo_id,
-                    claim_operation_id=args.claim_operation_id,
                     role=args.role,
                     decision_outcome=args.decision_outcome,
                     evidence=args.evidence,
@@ -725,6 +845,22 @@ def handle_todo_command(
                     **_todo_path_args(args),
                     dry_run=bool(args.dry_run),
                 )
+                try:
+                    payload = complete_operation(
+                        claim_operation_id=args.claim_operation_id,
+                    )
+                except LocalCoordinationAuthorityRejection as initial_error:
+                    if not args.recover_execution:
+                        raise
+                    payload = _recover_todo_execution_once(
+                        args,
+                        initial_error,
+                        registry_path=registry_path,
+                        runtime_root_arg=runtime_root_arg,
+                        retry_operation=lambda reference: complete_operation(
+                            claim_operation_id=reference,
+                        ),
+                    )
                 if settlement_identity is not None:
                     payload["settlement_identity"] = settlement_identity.as_dict()
                     payload["settlement_result"] = settlement_result_payload(

loopx/cli_commands/todo_argument_validation.py

@@ -16,6 +16,7 @@
     ("--todo-id", "todo_id"),
     ("--operation-id", "operation_id"),
     ("--claim-operation-id", "claim_operation_id"),
+    ("--recover-execution", "recover_execution"),
     ("--update-operation-id", "update_operation_id"),
     ("--update-expected-provider-revision", "update_expected_provider_revision"),
     ("--turn-instance-id", "turn_instance_id"),
@@ -156,6 +157,7 @@
     "update": frozenset(_TODO_UPDATE_MUTABLE_FIELDS) | frozenset({
         "role", "todo_id", "agent_id", "state_file", "project", "dry_run", "authority_reason",
         "claim_operation_id", "update_operation_id", "update_expected_provider_revision",
+        "recover_execution",
     }),
 }
 
@@ -449,6 +451,20 @@ def validate_todo_update_options(args: argparse.Namespace) -> None:
         )
     if not any(getattr(args, field) for field in _TODO_UPDATE_MUTABLE_FIELDS):
         raise ValueError("todo update requires at least one mutable todo field")
+    if args.status == "deferred" and args.recover_execution:
+        missing = [
+            flag
+            for flag, value in (
+                ("--reason", args.reason),
+                ("--resume-when", args.resume_when),
+            )
+            if not str(value or "").strip()
+        ]
+        if missing:
+            raise ValueError(
+                "todo update --status deferred requires --reason and "
+                "--resume-when; missing: " + ", ".join(missing)
+            )
     validation_fields = (
         args.validation_command,
         args.validation_command_json,
@@ -533,6 +549,50 @@ def validate_todo_complete_options(args: argparse.Namespace) -> None:
     validate_successor_routing_options(args)
 
 
+def validate_todo_execution_recovery_options(args: argparse.Namespace) -> None:
+    """Keep the one-command helper narrower than the typed recovery owner."""
+    if not getattr(args, "recover_execution", False):
+        return
+    if args.todo_command not in {"update", "complete"}:
+        raise ValueError(
+            "--recover-execution is supported only by todo update/complete"
+        )
+    if not args.agent_id:
+        raise ValueError("--recover-execution requires a registered --agent-id")
+    if args.dry_run:
+        raise ValueError(
+            "--recover-execution cannot be combined with --dry-run because the "
+            "retry requires a committed claim reference"
+        )
+    supplied_proof = [
+        flag
+        for flag, present in (
+            ("--claim-operation-id", args.claim_operation_id is not None),
+            (
+                "--task-lease-idempotency-key",
+                args.task_lease_idempotency_key is not None,
+            ),
+            (
+                "--task-lease-expected-version",
+                args.task_lease_expected_version is not None,
+            ),
+        )
+        if present
+    ]
+    if supplied_proof:
+        raise ValueError(
+            "--recover-execution accepts only omitted execution proof; inspect "
+            "and retry explicit stale proof unchanged instead of replacing: "
+            + ", ".join(supplied_proof)
+        )
+    if args.todo_command == "update" and args.update_expected_provider_revision:
+        raise ValueError(
+            "--recover-execution cannot silently rebase "
+            "--update-expected-provider-revision; claim, inspect the same Todo, "
+            "then retry with the reconciled provider revision"
+        )
+
+
 def validate_todo_supersede_options(args: argparse.Namespace) -> None:
     if not args.todo_id:
         raise ValueError("todo supersede requires --todo-id")

loopx/cli_commands/todo_registration.py

@@ -81,6 +81,15 @@ def register_todo_command(
             "Omit to retain a fresh operation id per invocation."
         ),
     )
+    todo_parser.add_argument(
+        "--recover-execution",
+        action="store_true",
+        help=(
+            "For todo update/complete without execution proof, consume only typed "
+            "claim_task recovery: claim once, then retry unchanged. Incompatible "
+            "with explicit claim/lease proof, --dry-run, or provider-CAS edits."
+        ),
+    )
     todo_parser.add_argument(
         "--turn-instance-id",
         help=(

loopx/control_plane/coordination/task_lease_acquire.ts

@@ -85,14 +85,45 @@ export async function executeCanonicalTaskLeaseAcquire(store: AuthorityStore, ra
     const decision = decideTaskLeaseAcquire({handoff_mode: mode!, registered_agents: input.registered_agents,
       ...facts, command: input});
     if (decision.outcome === "rejected" || decision.outcome === "conflict") {
+      const conflictLeases = decision.conflict_indexes.map(i => facts.other_leases[i]);
+      const ownConflicts = conflictLeases.filter(lease => lease.owner === input.owner);
+      const foreignConflictCount = conflictLeases.length - ownConflicts.length;
+      const ownConflict = ownConflicts.length === 1 ? ownConflicts[0] : null;
+      const ownConflictLease = ownConflict === null ? null : canonicalTaskLease(
+        indexCoordinationProjection(head.head, input.goal_id).leases.get(String(ownConflict.todo_id))!,
+        input.goal_id, String(ownConflict.todo_id));
+      const ownConflictRecovery = ownConflict === null || ownConflictLease === null ? {} : {
+        own_conflicting_lease: {
+          todo_id: ownConflict.todo_id,
+          owner: ownConflict.owner,
+          write_scopes: ownConflict.write_scopes,
+          version: leaseVersion(ownConflictLease),
+        },
+        recovery: {
+          action: "inspect_own_conflicting_lease",
+          execution_authority_granted: false,
+          reason: "Inspect the active lease for your other Todo. Release it only after confirming that execution has stopped; otherwise keep it and use an isolated worktree for independent code edits.",
+          inspect: {command: "loopx task-lease inspect", cli_args: ["task-lease", "inspect",
+            "--goal-id", input.goal_id, "--todo-id", String(ownConflict.todo_id)]},
+          release: {command: "loopx task-lease release", cli_args: ["task-lease", "release",
+            "--goal-id", input.goal_id, "--todo-id", String(ownConflict.todo_id),
+            "--owner", input.owner, "--idempotency-key", String(ownConflictLease.idempotency_key),
+            "--expected-version", String(leaseVersion(ownConflictLease))]},
+        },
+        ...(foreignConflictCount > 0 ? {foreign_conflict_count: foreignConflictCount} : {}),
+      };
       return failed(decision.code, decision.code === "handoff_mode_retired"
         ? "Legacy ownership cannot grant new execution; release retained leases and review handoff-mode plan-migration/migrate before acquiring again."
         : `canonical task lease acquire rejected: ${decision.code}`, {
         ...(decision.code === "write_scope_conflict" ? {recommended_action:
           "Coordinate shared writes with the listed holders. For isolated code edits, use your own separate Git worktree and acquire with --write-worktree; file overlaps become integration advisories. A known same-checkout collision remains rejected. Never take over a foreign lease or use code-edit mode for shared runtime state."} : {}),
         handoff_mode: mode, expected_version: input.expected_version, actual_version: leaseVersion(facts.current),
         ...(facts.todo ? {todo_status: facts.todo.status, claimed_by: facts.todo.claimed_by, excluded_agents: [...facts.todo.excluded_agents]} : {}),
-        ...(decision.conflict_indexes.length ? {conflicts: decision.conflict_indexes.map(i => facts.other_leases[i])} : {})});
+        ...ownConflictRecovery,
+        // Own conflicts get a bounded inspect/release route above. Foreign
+        // leases remain immutable and visible only when no own recovery can
+        // resolve the caller's collision.
+        ...(conflictLeases.length && ownConflict === null ? {conflicts: conflictLeases} : {})});
     }
     const objective = decision.outcome === "apply" ? goalObjectiveWorkGuard(head.head, input.owner) : null;
     if (objective !== null) return failed(String(objective.reason_code), String(objective.reason), {goal_objective_guard: objective});