From edd3123ff1c6929c15f322904fa2b03474a86b06 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Sun, 8 Nov 2026 02:43:01 +0800 Subject: [PATCH] fix(todo): recover missing execution claims in one command Signed-off-by: huangruiteng --- loopx/cli_commands/todo.py | 144 +++++++++- .../cli_commands/todo_argument_validation.py | 60 +++++ loopx/cli_commands/todo_registration.py | 9 + .../coordination/task_lease_acquire.ts | 33 ++- tests/control_plane/test_task_lease_cli.py | 27 +- .../test_todo_claim_argument_recovery.py | 254 ++++++++++++++++++ 6 files changed, 520 insertions(+), 7 deletions(-) diff --git a/loopx/cli_commands/todo.py b/loopx/cli_commands/todo.py index 4e515182fe..8e946156ae 100644 --- a/loopx/cli_commands/todo.py +++ b/loopx/cli_commands/todo.py @@ -3,6 +3,7 @@ import argparse import shlex from collections.abc import Callable, Sequence +from functools import partial from operator import itemgetter from pathlib import Path @@ -10,6 +11,8 @@ 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( diff --git a/loopx/cli_commands/todo_argument_validation.py b/loopx/cli_commands/todo_argument_validation.py index b9039f2842..de223c2a7b 100644 --- a/loopx/cli_commands/todo_argument_validation.py +++ b/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") diff --git a/loopx/cli_commands/todo_registration.py b/loopx/cli_commands/todo_registration.py index 5af11265db..a72defdbfd 100644 --- a/loopx/cli_commands/todo_registration.py +++ b/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=( diff --git a/loopx/control_plane/coordination/task_lease_acquire.ts b/loopx/control_plane/coordination/task_lease_acquire.ts index a30101e6ac..6eeaaae671 100644 --- a/loopx/control_plane/coordination/task_lease_acquire.ts +++ b/loopx/control_plane/coordination/task_lease_acquire.ts @@ -85,6 +85,33 @@ 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}`, { @@ -92,7 +119,11 @@ export async function executeCanonicalTaskLeaseAcquire(store: AuthorityStore, ra "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}); diff --git a/tests/control_plane/test_task_lease_cli.py b/tests/control_plane/test_task_lease_cli.py index ad44546c78..a362df36cb 100644 --- a/tests/control_plane/test_task_lease_cli.py +++ b/tests/control_plane/test_task_lease_cli.py @@ -35,7 +35,9 @@ def git(*args): (workspace / "src/node_modules/.bin").mkdir(parents=True) (workspace / "src/node_modules/tool").mkdir() (workspace / "src/node_modules/tool/bin.mjs").write_text("export {};\n") - goal, task_a, task_b = "checkout-admission", "todo_worker_a", "todo_worker_b" + goal, task_a, task_b, task_c = ( + "checkout-admission", "todo_worker_a", "todo_worker_b", "todo_worker_c" + ) state = project / "state.md" state.write_text("# Synthetic worktree admission\n\n## Agent Todo\n") registry.write_text(json.dumps({"common_runtime_root": str(runtime), "goals": [{ @@ -47,7 +49,11 @@ def git(*args): "text": "Edit isolated source", "archive_state": "active", "source_section": "Agent Todo", "index": index, "task_class": "advancement_task", "claimed_by": owner, "task_repository": "git:github.com/example/project", "required_write_scopes": ["src/**"], - } for index, task, owner in [(1, task_a, "agent-a"), (2, task_b, "agent-b")]]) + } for index, task, owner in [ + (1, task_a, "agent-a"), + (2, task_b, "agent-b"), + (3, task_c, "agent-a"), + ]]) initialize_canonical_authority(runtime, goal, projection, state_path=state, provider=provider) def cli(action, task, *flags, exit_code=0): @@ -100,6 +106,23 @@ def release(task, owner, key, version): denied = acquire(task_b, "agent-b", "same-checkout", 0, first, exit_code=1) assert denied["error_code"] == "write_scope_conflict" + assert "recovery" not in denied + assert denied["conflicts"][0]["todo_id"] == task_a + own_denied = acquire(task_c, "agent-a", "own-same-checkout", 0, first, exit_code=1) + assert own_denied["error_code"] == "write_scope_conflict" + assert "conflicts" not in own_denied + assert own_denied["own_conflicting_lease"] == { + "todo_id": task_a, + "owner": "agent-a", + "write_scopes": ["src/**"], + "version": current["version"], + } + recovery = own_denied["recovery"] + assert recovery["action"] == "inspect_own_conflicting_lease" + assert recovery["execution_authority_granted"] is False + assert recovery["inspect"]["cli_args"][-1] == task_a + assert recovery["release"]["cli_args"][-1] == str(current["version"]) + assert cli("inspect", task_a)["lease"] == current independent = acquire(task_b, "agent-b", "separate-checkout", 0, second) assert any(row["todo_id"] == task_a for row in independent["integration_overlap_advisories"]) assert independent["lease"]["write_workspace"]["worktree"] != admitted["lease"]["write_workspace"]["worktree"] diff --git a/tests/control_plane/test_todo_claim_argument_recovery.py b/tests/control_plane/test_todo_claim_argument_recovery.py index 981890f4f1..3c7ce25980 100644 --- a/tests/control_plane/test_todo_claim_argument_recovery.py +++ b/tests/control_plane/test_todo_claim_argument_recovery.py @@ -6,11 +6,17 @@ import shlex import subprocess import sys +from argparse import Namespace from pathlib import Path import pytest from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime +from loopx.cli_commands import todo as todo_command +from loopx.control_plane.coordination.local_authority import ( + LocalCoordinationAuthorityRejection, + LocalCoordinationAuthorityUnavailable, +) from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection REPO = Path(__file__).resolve().parents[2] @@ -411,6 +417,254 @@ def test_terminal_refusal_recovers_through_ordinary_claim(claim_cli): assert snapshot() == completed +@pytest.mark.parametrize("command", ["update", "complete"]) +def test_explicit_one_command_execution_recovery(claim_cli, command): + _, run, snapshot, prefix, claim, _, _, _ = claim_cli + if command == "update": + assert run([ + *prefix, *claim, "--claimed-by", "peer-a", + "--claim-operation-id", "active-before-one-command-recovery", + ])["ok"] + arguments = [ + *prefix, + "todo", + command, + "--goal-id", + GOAL, + "--todo-id", + TODO, + "--agent-id", + "peer-a", + "--recover-execution", + ] + arguments += ( + [ + "--note", "Recovered without a caller-visible failed command.", + "--clear-resume-when", + ] + if command == "update" + else ["--evidence", "evidence:validated-output"] + ) + result = run(arguments) + assert result["ok"] is True + recovery = result["execution_recovery"] + assert recovery == { + "schema_version": "todo_execution_recovery_v0", + "status": "recovered", + "initial_error_code": ( + "lease_fence_required" + if command == "update" + else "handoff_mode_requires_lease" + ), + "claim_operation_id": recovery["claim_operation_id"], + "claim_status": "no_change" if command == "update" else "applied", + "external_invocations": 1, + "external_failed_calls": 0, + } + task = snapshot()[0][0] + assert task["status"] == ("open" if command == "update" else "done") + if command == "update": + assert task["note"] == "Recovered without a caller-visible failed command." + + +def test_one_command_recovery_never_replaces_explicit_stale_proof(claim_cli): + _, run, snapshot, prefix, claim, _, _, _ = claim_cli + assert run([ + *prefix, *claim, "--claimed-by", "peer-a", + "--claim-operation-id", "current-execution", + ])["ok"] + before = snapshot() + rejected = run([ + *prefix, "todo", "update", "--goal-id", GOAL, "--todo-id", TODO, + "--agent-id", "peer-a", "--note", "Do not replace proof.", + "--recover-execution", "--task-lease-idempotency-key", "stale-proof", + "--task-lease-expected-version", "0", + ], expected_exit=1) + assert "accepts only omitted execution proof" in rejected["error"] + assert "execution_recovery" not in rejected + assert snapshot() == before + + +def test_one_command_recovery_does_not_take_foreign_owner(claim_cli): + _, run, snapshot, prefix, claim, _, _, _ = claim_cli + assert run([ + *prefix, *claim, "--claimed-by", "peer-b", "--agent-id", "peer-b", + "--claim-operation-id", "foreign-execution", + ])["ok"] + before = snapshot() + rejected = run([ + *prefix, "todo", "update", "--goal-id", GOAL, "--todo-id", TODO, + "--agent-id", "peer-a", "--note", "Forbidden takeover.", + "--recover-execution", + ], expected_exit=1) + assert rejected["error_code"] == "update_owner_mismatch" + assert snapshot() == before + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_one_command_recovery_preserves_missing_dependency( + tmp_path, monkeypatch, provider, +): + isolate_sqlite_runtime(tmp_path, monkeypatch) + project = tmp_path / "dependency-project" + project.mkdir() + runtime, state = tmp_path / "runtime", project / "state.md" + registry = tmp_path / "registry.json" + state.write_text("# Synthetic dependency\n", encoding="utf-8") + registry.write_text(json.dumps({ + "common_runtime_root": str(runtime), + "goals": [{ + "id": "dependency-goal", + "repo": str(project), + "state_file": state.name, + "coordination": {"registered_agents": ["peer-a"]}, + }], + }), encoding="utf-8") + projection = build_todo_runtime_shadow_projection( + goal_id="dependency-goal", + handoff_mode="hard_lease", + leases=[], + todos=[{ + "schema_version": "todo_item_v0", + "todo_id": "todo_dependency_wait", + "role": "agent", + "status": "open", + "done": False, + "text": "Wait for a missing prerequisite.", + "archive_state": "active", + "source_section": "Agent Todo", + "index": 1, + "task_class": "advancement_task", + "claimed_by": "peer-a", + "resume_when": "todo_done:todo_missing_dependency", + }], + ) + initialize_canonical_authority( + runtime, + "dependency-goal", + projection, + state_path=state, + provider=provider, + ) + state.unlink() + before = subprocess.run([ + sys.executable, "-m", "loopx.cli", "--registry", str(registry), + "--runtime-root", str(runtime), "--format", "json", "todo", "list", + "--goal-id", "dependency-goal", "--todo-id", "todo_dependency_wait", + ], cwd=REPO, capture_output=True, text=True, timeout=90, check=True).stdout + completed = subprocess.run([ + sys.executable, "-m", "loopx.cli", "--registry", str(registry), + "--runtime-root", str(runtime), "--format", "json", "todo", "complete", + "--goal-id", "dependency-goal", "--todo-id", "todo_dependency_wait", + "--agent-id", "peer-a", "--evidence", "evidence:must-not-commit", + "--recover-execution", + ], cwd=REPO, capture_output=True, text=True, timeout=90) + assert completed.returncode == 1, completed.stdout + completed.stderr + rejected = json.loads(completed.stdout) + assert rejected["error_code"] == "handoff_mode_requires_lease" + assert rejected["recovery"]["action"] == "resolve_acquire_rejection" + assert rejected["recovery"]["reason_code"] == "todo_dependency_pending" + assert "execution_recovery" not in rejected + after = subprocess.run([ + sys.executable, "-m", "loopx.cli", "--registry", str(registry), + "--runtime-root", str(runtime), "--format", "json", "todo", "list", + "--goal-id", "dependency-goal", "--todo-id", "todo_dependency_wait", + ], cwd=REPO, capture_output=True, text=True, timeout=90, check=True).stdout + assert json.loads(after)["todo"] == json.loads(before)["todo"] + + +def test_one_command_recovery_reports_committed_claim_when_retry_fails(monkeypatch): + initial = LocalCoordinationAuthorityRejection( + "ordinary lifecycle operation needs a managed claim", + code="lease_fence_required", + payload={"recovery": { + "action": "claim_task", + "execution_authority_granted": False, + "claim": {"command": "loopx todo claim"}, + "retry": { + "reference_source": "claim_operation_id_from_claim", + "requires_flags": ["--claim-operation-id"], + }, + }}, + ) + args = Namespace( + goal_id="goal-retry-race", + todo_id="todo-retry-race", + role="agent", + agent_id="peer-a", + project=None, + state_file=None, + ) + + def claim(**kwargs): + assert kwargs["claim_only"] is True + assert kwargs["manage_execution"] is True + assert kwargs["claimed_by"] == kwargs["agent_id"] == "peer-a" + return { + "ok": True, + "execution_managed": True, + "claim_operation_id": "committed-claim", + "status": "applied", + } + + monkeypatch.setattr(todo_command, "update_goal_todo", claim) + + def retry(reference): + assert reference == "committed-claim" + raise LocalCoordinationAuthorityRejection( + "the unchanged retry lost a provider race", + code="provider_revision_mismatch", + payload={"expected_provider_revision": "1", "actual_provider_revision": "2"}, + ) + + with pytest.raises(LocalCoordinationAuthorityUnavailable) as caught: + todo_command._recover_todo_execution_once( + args, + initial, + registry_path=Path("registry.json"), + runtime_root_arg=None, + retry_operation=retry, + ) + + assert caught.value.code == "todo_execution_recovery_retry_failed" + assert caught.value.payload == { + "execution_recovery": { + "schema_version": "todo_execution_recovery_v0", + "status": "claim_committed_retry_failed", + "initial_error_code": "lease_fence_required", + "claim_operation_id": "committed-claim", + "claim_status": "applied", + "retry_error_code": "provider_revision_mismatch", + }, + "retry_failure": { + "expected_provider_revision": "1", + "actual_provider_revision": "2", + }, + } + + +@pytest.mark.parametrize( + "flags,missing", + [ + (["--resume-when", "capacity_available:test"], "--reason"), + (["--reason", "Wait for capacity"], "--resume-when"), + ], +) +def test_deferred_transition_names_missing_contract_before_claim( + claim_cli, flags, missing, +): + _, run, snapshot, prefix, _, _, _, _ = claim_cli + before = snapshot() + rejected = run([ + *prefix, "todo", "update", "--goal-id", GOAL, "--todo-id", TODO, + "--agent-id", "peer-a", "--status", "deferred", + "--recover-execution", *flags, + ], expected_exit=1) + assert "requires --reason and --resume-when" in rejected["error"] + assert missing in rejected["error"] + assert snapshot() == before + + @pytest.mark.parametrize("command", ["update", "complete"]) def test_empty_managed_reference_is_rejected_without_mutation(claim_cli, command): _, run, snapshot, prefix, claim, _, _, _ = claim_cli