From 1e5bedbad50509747e659bc992f2d921fc9ae87c Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 2 Nov 2026 17:59:24 +0530 Subject: [PATCH] fix(cli): retain native Todo refusal routes and original retry inputs Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/cli_commands/task_lease_arguments.py | 19 +-- loopx/cli_commands/todo_event.py | 72 ++++++++++ loopx/command_invocation.py | 6 + loopx/control_plane/todos/markdown.py | 13 +- .../test_todo_claim_argument_recovery.py | 123 +++++++++++++++++- .../test_todo_update_lease_recovery.py | 2 +- 6 files changed, 213 insertions(+), 22 deletions(-) diff --git a/loopx/cli_commands/task_lease_arguments.py b/loopx/cli_commands/task_lease_arguments.py index 3a2e746e4d..ce1ed3355b 100644 --- a/loopx/cli_commands/task_lease_arguments.py +++ b/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, diff --git a/loopx/cli_commands/todo_event.py b/loopx/cli_commands/todo_event.py index 59952eaa11..0181bd93a0 100644 --- a/loopx/cli_commands/todo_event.py +++ b/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 diff --git a/loopx/command_invocation.py b/loopx/command_invocation.py index aa250af90e..6375cf8a43 100644 --- a/loopx/command_invocation.py +++ b/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. diff --git a/loopx/control_plane/todos/markdown.py b/loopx/control_plane/todos/markdown.py index 9d8c2b9921..24ad7428e8 100644 --- a/loopx/control_plane/todos/markdown.py +++ b/loopx/control_plane/todos/markdown.py @@ -14,8 +14,10 @@ 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']}") @@ -23,7 +25,14 @@ def _render_lease_recovery(payload: dict[str, Any]) -> list[str]: 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'])}`") diff --git a/tests/control_plane/test_todo_claim_argument_recovery.py b/tests/control_plane/test_todo_claim_argument_recovery.py index 0a9308cf24..981890f4f1 100644 --- a/tests/control_plane/test_todo_claim_argument_recovery.py +++ b/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"] diff --git a/tests/control_plane/test_todo_update_lease_recovery.py b/tests/control_plane/test_todo_update_lease_recovery.py index ebb1acc4b3..79f7315aea 100644 --- a/tests/control_plane/test_todo_update_lease_recovery.py +++ b/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)