fix(cli): preserve native Todo refusal routes and retry inputs

loopx/cli_commands/task_lease_arguments.py

@@ -5,7 +5,7 @@
 import shlex
 from pathlib import Path
 
-from ..command_invocation import loopx_cli_argv
+from ..command_invocation import cli_option_args, loopx_cli_argv
 
 
 LEASE_OPTION_FIELDS = (
@@ -30,13 +30,6 @@ def _present(value: object) -> bool:
     return value is not None and value is not False and value != "" and value != []
 
 
-def _option_args(flag: str, value: object) -> list[str]:
-    text = str(value)
-    # A legal value may look like an option. Bind it explicitly so argparse
-    # cannot reinterpret the repair's identity, route or repeated scope.
-    return [f"{flag}={text}"] if text.startswith("-") else [flag, text]
-
-
 class TaskLeaseArgumentError(ValueError):
     def __init__(self, *, missing: list[str], unsupported: list[str]) -> None:
         self.missing = missing
@@ -51,12 +44,12 @@ def __init__(self, *, missing: list[str], unsupported: list[str]) -> None:
     def recovery(self, args: argparse.Namespace, *, registry_path: Path,
                  runtime_root_arg: str | None) -> dict[str, object]:
         route = loopx_cli_argv()
-        cli_args = [*route[1:], *_option_args("--registry", registry_path), "--format", "json"]
+        cli_args = [*route[1:], *cli_option_args("--registry", registry_path), "--format", "json"]
         if runtime_root_arg is not None:
-            cli_args.extend(_option_args("--runtime-root", runtime_root_arg))
+            cli_args.extend(cli_option_args("--runtime-root", runtime_root_arg))
         action = args.task_lease_command
-        cli_args.extend(["task-lease", action, *_option_args("--goal-id", args.goal_id),
-                         *_option_args("--todo-id", args.todo_id)])
+        cli_args.extend(["task-lease", action, *cli_option_args("--goal-id", args.goal_id),
+                         *cli_option_args("--todo-id", args.todo_id)])
         for flag, field in LEASE_OPTION_FIELDS:
             value = getattr(args, field, None)
             if field not in LEASE_ACTION_FIELDS[action] or not _present(value):
@@ -65,7 +58,7 @@ def recovery(self, args: argparse.Namespace, *, registry_path: Path,
                 cli_args.append(flag)
             else:
                 for item in value if isinstance(value, list) else [value]:
-                    cli_args.extend(_option_args(flag, item))
+                    cli_args.extend(cli_option_args(flag, item))
         return {
             "command": shlex.join([*route, "task-lease", action]), "cli_args": cli_args,
             "requires_flags": self.missing, "remove_flags": self.unsupported,

loopx/cli_commands/todo_event.py

@@ -1,11 +1,14 @@
 from __future__ import annotations
 
 import argparse
+import shlex
 from collections.abc import Callable, Mapping
 from pathlib import Path
 
+from ..command_invocation import cli_option_args, loopx_cli_argv
 from ..control_plane.todos.contract import decision_scope_metadata_value
 from ..file_lock import lock_timeout_error_fields
+from .todo_argument_validation import TODO_OPTION_FIELDS
 
 
 RolloutEventAppender = Callable[..., dict[str, object]]
@@ -20,6 +23,74 @@
 }
 
 
+def _original_todo_arguments(args: argparse.Namespace) -> list[str]:
+    """Carry parsed caller inputs; Core still decides whether a retry is legal."""
+    arguments = ["todo", args.todo_command, "--goal-id", args.goal_id]
+    for flag, field in (*TODO_OPTION_FIELDS, ("--project", "project"),
+                        ("--dry-run", "dry_run"), ("--result-file", "result_file")):
+        value = getattr(args, field, None)
+        if isinstance(value, bool):
+            if value:
+                arguments.append(flag)
+        elif isinstance(value, list):
+            for item in value:
+                arguments.extend(cli_option_args(flag, item))
+        elif value is not None:
+            arguments.extend(cli_option_args(flag, value))
+    return arguments
+
+
+def _route_native_recovery(
+    payload: dict[str, object], args: argparse.Namespace, *,
+    registry_path: Path, runtime_root_arg: str | None,
+) -> None:
+    """Adapt native guidance to this CLI without changing its recovery decision."""
+    recovery = payload.get("recovery")
+    if not isinstance(recovery, dict):
+        return
+    route = loopx_cli_argv()
+    routed = dict(recovery)
+    for step, descriptor in recovery.items():
+        if not isinstance(descriptor, dict) or not isinstance(descriptor.get("command"), str):
+            continue
+        try:
+            command = shlex.split(descriptor["command"])
+        except ValueError:
+            continue  # Malformed optional guidance must not hide the refusal.
+        if not command or command[0] != "loopx":
+            continue
+        arguments = descriptor.get("cli_args")
+        if descriptor.get("reference_source") == "claim_operation_id_from_claim":
+            arguments = _original_todo_arguments(args)
+        else:
+            arguments = list(arguments) if isinstance(arguments, list) else command[1:]
+        # The update adapter may already bind selectors. Inspect only the global
+        # prelude, never user prose that happens to resemble an option.
+        start = next((i for i in range(len(arguments) - 1)
+                      if arguments[i:i + 2] == command[1:3]), 0)
+        prelude = arguments[:start]
+        selectors = list(route[1:])
+        for flag, value in (("--registry", str(registry_path)),
+                            ("--runtime-root", runtime_root_arg), ("--format", "json")):
+            if value is not None and not any(part == flag or part.startswith(flag + "=") for part in prelude):
+                selectors.extend(cli_option_args(flag, value))
+        for flag, field in (("--goal-id", "goal_id"), ("--todo-id", "todo_id")):
+            value = descriptor.get(field) or getattr(args, field, None)
+            if flag not in arguments and value is not None:
+                arguments.extend(cli_option_args(flag, value))
+        if arguments[:1] == ["todo"]:
+            for flag, field in (("--agent-id", "agent_id"), ("--role", "role"),
+                                ("--project", "project"), ("--state-file", "state_file")):
+                value = getattr(args, field, None)
+                if flag not in arguments and value is not None:
+                    arguments.extend(cli_option_args(flag, value))
+            if getattr(args, "dry_run", False) and "--dry-run" not in arguments:
+                arguments.append("--dry-run")
+        routed[step] = {**descriptor, "command": shlex.join([*route, *command[1:]]),
+                        "cli_args": [*selectors, *arguments]}
+    payload["recovery"] = routed
+
+
 def todo_error_payload(
     args: argparse.Namespace, exc: Exception, *, registry_path: Path,
     runtime_root_arg: str | None,
@@ -73,6 +144,7 @@ def todo_error_payload(
         payload["error_code"] = exc.code
         if exc.authoring_contract is not None:
             payload["authoring_contract"] = exc.authoring_contract
+    _route_native_recovery(payload, args, registry_path=registry_path, runtime_root_arg=runtime_root_arg)
     return payload
 
 

loopx/command_invocation.py

@@ -36,6 +36,12 @@ def loopx_cli_argv(cli_bin: str | None = None) -> list[str]:
     return [program] if program == "loopx" else [program, "--cli-bin", program]
 
 
+def cli_option_args(flag: str, value: object) -> list[str]:
+    """Keep a legal option-looking value from becoming another CLI option."""
+    text = str(value)
+    return [f"{flag}={text}"] if text.startswith("-") else [flag, text]
+
+
 def current_loopx_cli_argv() -> list[str]:
     """Pin captured readers to this installed interpreter or explicit CLI.
 

loopx/control_plane/todos/markdown.py

@@ -14,16 +14,25 @@ def _render_lease_recovery(payload: dict[str, Any]) -> list[str]:
     if not isinstance(recovery, dict) or not recovery.get("reason") or not payload.get("handoff_mode"):
         return []
     retry = recovery.get("retry")
+    retry_args = retry.get("cli_args", []) if isinstance(retry, dict) else []
     complete_claim_recovery = (recovery.get("action") == "claim_task"
-        and isinstance(retry, dict) and retry.get("command") == "loopx todo complete")
+        and isinstance(retry_args, list)
+        and any(retry_args[i:i + 2] == ["todo", "complete"] for i in range(len(retry_args) - 1)))
     lines = ([] if complete_claim_recovery
         else [f"- handoff_mode: `{payload['handoff_mode']}`"])
     lines.append(f"- recovery: {recovery['reason']}")
     for step in ("claim", "inspect", "acquire", "retry", "release"):
         instruction = recovery.get(step)
         if isinstance(instruction, dict) and instruction.get("command"):
             arguments = instruction.get("cli_args")
-            command = shlex.join(["loopx", *arguments]) if isinstance(arguments, list) else instruction["command"]
+            command = instruction["command"]
+            if isinstance(arguments, list):
+                try:
+                    program = shlex.split(command)[0]
+                except (ValueError, IndexError):
+                    pass  # Keep the original diagnostic when guidance is malformed.
+                else:
+                    command = shlex.join([program, *arguments])
             lines.append(f"- {step}: `{command}`")
             if instruction.get("requires_flags"):
                 lines.append(f"  - supply: `{', '.join(instruction['requires_flags'])}`")

tests/control_plane/test_todo_claim_argument_recovery.py

@@ -216,9 +216,9 @@ def test_ordinary_claim_progress_completion_need_no_lease_flags(claim_cli):
     assert snapshot() == before
 
 
-@pytest.mark.skipif(sys.platform == "win32", reason="POSIX executable alias and PATH fixture")
-def test_explicit_runtime_survives_repair_progress_and_completion(claim_cli, tmp_path, monkeypatch):
-    _, run, snapshot, prefix, claim, _, project, _ = claim_cli
+@pytest.fixture
+def selected_runtime(claim_cli, tmp_path, monkeypatch):
+    _, _, _, _, _, _, project, _ = claim_cli
     program = "candidate's cli"
     binaries = tmp_path / "candidate bin"
     binaries.mkdir()
@@ -238,6 +238,13 @@ def execute(argv, expected_exit=0):
         assert not marker.exists()
         return json.loads(result.stdout)
 
+    return program, execute
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX executable alias and PATH fixture")
+def test_explicit_runtime_survives_repair_progress_and_completion(claim_cli, selected_runtime):
+    _, run, snapshot, prefix, claim, _, _, _ = claim_cli
+    program, execute = selected_runtime
     before = snapshot()
     rejected = run(["--cli-bin", program, *prefix, *claim, "--note", "invalid"], expected_exit=1)
     assert snapshot() == before
@@ -275,6 +282,108 @@ def execute(argv, expected_exit=0):
     assert snapshot() == completed
 
 
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX executable alias and PATH fixture")
+@pytest.mark.parametrize("command", ["update", "complete"])
+def test_native_refusal_preserves_runtime_and_original_operation(claim_cli, selected_runtime, command):
+    _, run, snapshot, prefix, claim, _, project, state = claim_cli
+    program, execute = selected_runtime
+    if command == "update":
+        # Copy edits need no lease. Clearing a wait retains the execution fence;
+        # exercise omitted proof while this actor owns a current execution.
+        assert run([*prefix, *claim, "--claimed-by", "peer-a",
+                    "--claim-operation-id", "original-execution"])["ok"]
+    original = ["--cli-bin", program, *prefix, "todo", command, "--goal-id", GOAL,
+        "--todo-id", TODO, "--agent-id", "peer-a", "--role", "agent",
+        "--project", str(project), "--state-file", str(state),
+        "--note=--keep the original 中文 'note' and evidence.", "--evidence", "evidence:original-output"]
+    if command == "update":
+        # Preserve the caller's operation across a lost response and replay.
+        original += ["--update-operation-id", "original-native-update", "--clear-resume-when"]
+    before = snapshot()
+    rejected = run(original, expected_exit=1)
+    assert snapshot() == before
+    recovery = rejected["recovery"]
+    assert recovery["action"] == "claim_task"
+    assert recovery["execution_authority_granted"] is False
+    from loopx.control_plane.todos.markdown import render_todo_markdown
+    markdown_args = list(original)
+    markdown_args[markdown_args.index("--format") + 1] = "markdown"
+    markdown = subprocess.run([sys.executable, "-m", "loopx.cli", *markdown_args],
+        cwd=REPO, capture_output=True, text=True, timeout=90)
+    assert markdown.returncode == 1, markdown.stdout + markdown.stderr
+    assert snapshot() == before
+    rendered = markdown.stdout.rstrip("\n")
+    assert rendered == render_todo_markdown(rejected)
+
+    def emitted(step):
+        line = next(line for line in rendered.splitlines() if line.startswith(f"- {step}: `"))
+        argv = shlex.split(line.split("`", 2)[1])
+        # JSON and Markdown must carry the same executable arguments.
+        assert argv[1:] == recovery[step]["cli_args"]
+        return argv
+
+    # Execute the emitted claim unchanged from outside the checkout. An ambient
+    # global CLI is poisoned; both route selectors must reach the real backend.
+    claim_argv = emitted("claim")
+    prepared = execute(claim_argv)
+    assert prepared["ok"] and prepared["execution_managed"]
+    claimed = snapshot()
+    assert execute([*claim_argv, "--claim-operation-id", prepared["claim_operation_id"]])["ok"]
+    assert snapshot() == claimed
+    retry = [*emitted("retry"), "--claim-operation-id", prepared["claim_operation_id"]]
+    for flag, value in (("--project", str(project)), ("--state-file", str(state)), ("--role", "agent")):
+        assert retry[retry.index(flag) + 1] == value
+    if command == "update":
+        assert "--clear-resume-when" in retry
+        assert retry[retry.index("--update-operation-id") + 1] == "original-native-update"
+    wrong_actor = ["peer-b" if i and retry[i - 1] == "--agent-id" else part
+                   for i, part in enumerate(retry)]
+    assert execute(wrong_actor, expected_exit=1)["ok"] is False
+    assert snapshot() == claimed
+    assert execute(retry)["ok"]
+    after = snapshot()
+    task = after[0][0]
+    assert task["note"] == "--keep the original 中文 'note' and evidence."
+    assert task["evidence"] == "evidence:original-output"
+    assert task["status"] == ("open" if command == "update" else "done")
+    assert execute(retry)["ok"]
+    assert snapshot() == after
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX executable alias and PATH fixture")
+def test_native_inspection_keeps_route_and_rejects_stale_proof(claim_cli, selected_runtime):
+    _, run, snapshot, prefix, claim, _, _, _ = claim_cli
+    program, execute = selected_runtime
+    assert run([*prefix, *claim, "--claimed-by", "peer-a",
+                "--claim-operation-id", "before-stale-proof"])["ok"]
+    before = snapshot()
+    rejected = run(["--cli-bin", program, *prefix, "todo", "update", "--goal-id", GOAL,
+        "--todo-id", TODO, "--agent-id", "peer-a", "--note", "Do not weaken CAS.",
+        "--task-lease-idempotency-key", "wrong-key", "--task-lease-expected-version", "0"], expected_exit=1)
+    assert snapshot() == before
+    from loopx.control_plane.todos.markdown import render_todo_markdown
+    rendered = render_todo_markdown(rejected)
+    line = next(line for line in rendered.splitlines() if line.startswith("- inspect: `"))
+    inspected = execute(shlex.split(line.split("`", 2)[1]))
+    assert inspected["active"] is True
+    assert inspected["lease"]["version"] == 1
+    assert snapshot() == before
+
+
+def test_native_refusal_preview_preserves_no_write_intent(claim_cli):
+    _, run, snapshot, prefix, _, _, _, _ = claim_cli
+    before = snapshot()
+    rejected = run([*prefix, "todo", "complete", "--goal-id", GOAL, "--todo-id", TODO,
+        "--agent-id", "peer-a", "--evidence", "evidence:preview", "--dry-run"], expected_exit=1)
+    recovery = rejected["recovery"]
+    assert recovery["execution_authority_granted"] is False
+    assert "--dry-run" in recovery["claim"]["cli_args"]
+    assert "--dry-run" in recovery["retry"]["cli_args"]
+    preview = run(recovery["claim"]["cli_args"])
+    assert preview["dry_run"] is True
+    assert snapshot() == before
+
+
 def test_terminal_refusal_recovers_through_ordinary_claim(claim_cli):
     _, run, snapshot, prefix, _, _, _, _ = claim_cli
     complete = [*prefix, "todo", "complete", "--goal-id", GOAL, "--todo-id", TODO,
@@ -288,12 +397,14 @@ def test_terminal_refusal_recovers_through_ordinary_claim(claim_cli):
     assert recovery["execution_authority_granted"] is False
     from loopx.control_plane.todos.markdown import render_todo_markdown
     rendered = render_todo_markdown(rejected)
-    assert "loopx todo claim" in rendered and "--claim-operation-id" in rendered
+    assert recovery["claim"]["command"] == "loopx todo claim"
+    assert "- claim: `loopx " in rendered and "--claim-operation-id" in rendered
+    assert "--cli-bin" not in recovery["claim"]["cli_args"]
     assert "task-lease" not in rendered and "handoff_mode:" not in rendered
-    prepared = run([*prefix, *recovery["claim"]["cli_args"],
+    prepared = run([*recovery["claim"]["cli_args"],
         "--claim-operation-id", "after-terminal-refusal"])
     assert prepared["ok"] and prepared["execution_managed"]
-    retry = [*complete, "--claim-operation-id", prepared["claim_operation_id"]]
+    retry = [*recovery["retry"]["cli_args"], "--claim-operation-id", prepared["claim_operation_id"]]
     assert run(retry)["ok"]
     completed = snapshot()
     assert run(retry)["ok"]

tests/control_plane/test_todo_update_lease_recovery.py

@@ -173,7 +173,7 @@ def snapshot() -> dict:
         assert recovery["execution_authority_granted"] is False
         rendered = render_todo_markdown(rejected)
         assert f"handoff_mode: `{mode}`" in rendered
-        assert "loopx todo claim" in rendered
+        assert "- claim: `loopx " in rendered
         assert "task-lease" not in json.dumps(recovery)
         assert "first-execution" not in rendered
         assert "first-execution" not in json.dumps(rejected)