fix(cli): preserve explicit runtime through claim and lease recovery

loopx/cli_commands/task_lease.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
 
 import argparse
+import shlex
 from collections.abc import Callable
 from pathlib import Path
 
+from ..command_invocation import loopx_cli_argv
 from ..control_plane.work_items.local_lease_record import TaskLeaseError
 from ..control_plane.work_items.task_lease import (
     inspect_task_lease,
@@ -40,16 +42,14 @@ def render_task_lease_markdown(payload: dict[str, object]) -> str:
         lines.append(f"- error_code: `{payload.get('error_code')}`")
     inspection = payload.get("inspection")
     if isinstance(inspection, dict):
-        import shlex
-        lines.append(f"- inspect current state: `{shlex.join(['loopx', *inspection['cli_args']])}`")
+        lines.append(f"- inspect current state: `{shlex.join([loopx_cli_argv()[0], *inspection['cli_args']])}`")
         lines.append(f"- {inspection['reason']}")
     recovery = payload.get("recovery")
     if isinstance(recovery, dict) and payload.get("error_code") == "canonical_authority_required":
         lines.append(f"- recovery: `{recovery['action']}`")
         lines.append(f"- backup_required: `{recovery['backup_required']}`")
     elif isinstance(recovery, dict):
-        import shlex
-        lines.append(f"- repair: `{shlex.join(['loopx', *recovery['cli_args']])}`")
+        lines.append(f"- repair: `{shlex.join([loopx_cli_argv()[0], *recovery['cli_args']])}`")
         lines.append(f"- supply: `{', '.join(recovery['requires_flags']) or 'none'}`")
         lines.append(f"- remove: `{', '.join(recovery['remove_flags']) or 'none'}`")
         lines.append(f"- {recovery['reason']}")
@@ -256,7 +256,7 @@ def handle_task_lease_command(
         # Presentation only: retain every native refusal and explicit proof.
         # An exact read is useful for both grammar and semantic failures; it
         # neither chooses another owner nor retries a mutation with new proof.
-        inspection_args = [f"--registry={registry_path}", "--format", "json"]
+        inspection_args = [*loopx_cli_argv()[1:], f"--registry={registry_path}", "--format", "json"]
         if runtime_root_arg is not None:
             inspection_args.append(f"--runtime-root={runtime_root_arg}")
         inspection_args.extend(["task-lease", "inspect", f"--goal-id={args.goal_id}", f"--todo-id={args.todo_id}"])

loopx/cli_commands/task_lease_arguments.py

@@ -2,8 +2,11 @@
 from __future__ import annotations
 
 import argparse
+import shlex
 from pathlib import Path
 
+from ..command_invocation import loopx_cli_argv
+
 
 LEASE_OPTION_FIELDS = (
     ("--owner", "owner"), ("--idempotency-key", "idempotency_key"),
@@ -47,7 +50,8 @@ 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]:
-        cli_args = [*_option_args("--registry", registry_path), "--format", "json"]
+        route = loopx_cli_argv()
+        cli_args = [*route[1:], *_option_args("--registry", registry_path), "--format", "json"]
         if runtime_root_arg is not None:
             cli_args.extend(_option_args("--runtime-root", runtime_root_arg))
         action = args.task_lease_command
@@ -63,7 +67,7 @@ def recovery(self, args: argparse.Namespace, *, registry_path: Path,
                 for item in value if isinstance(value, list) else [value]:
                     cli_args.extend(_option_args(flag, item))
         return {
-            "command": f"loopx task-lease {action}", "cli_args": cli_args,
+            "command": shlex.join([*route, "task-lease", action]), "cli_args": cli_args,
             "requires_flags": self.missing, "remove_flags": self.unsupported,
             "reason": "Review removed flags and supply missing values, then retry. "
                       "Identities and CAS are preserved; canonical lease checks still apply. "

loopx/cli_commands/todo.py

@@ -6,6 +6,7 @@
 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 (
@@ -508,7 +509,7 @@ def handle_todo_command(
                 dry_run=bool(args.dry_run),
             )
             if payload.get("ok") and payload.get("execution_managed") and not args.dry_run:
-                route = ["loopx", "--registry", str(registry_path), "--runtime-root",
+                route = [*loopx_cli_argv(), "--registry", str(registry_path), "--runtime-root",
                     str(resolve_runtime_root(load_registry(registry_path), runtime_root_arg)), "--format", "json"]
                 selected = ["--goal-id", args.goal_id, "--todo-id", args.todo_id,
                     "--agent-id", args.agent_id, "--claim-operation-id", payload["claim_operation_id"]]

loopx/cli_commands/todo_argument_validation.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
 
 import argparse
+import shlex
 from collections.abc import Iterable
 from pathlib import Path
 
+from ..command_invocation import loopx_cli_argv
 from ..control_plane.todos.contract import TODO_CONTINUATION_POLICY_VALUES
 
 TODO_OPTION_FIELDS = (
@@ -379,7 +381,8 @@ def recovery(
         ]
         if args.task_lease_expected_version is not None and not args.task_lease_idempotency_key:
             missing.append("--task-lease-idempotency-key")
-        cli_args = ["--registry", str(registry_path), "--format", "json"]
+        route = loopx_cli_argv()
+        cli_args = [*route[1:], "--registry", str(registry_path), "--format", "json"]
         if runtime_root_arg is not None:
             cli_args.extend(["--runtime-root", runtime_root_arg])
         cli_args.extend(["todo", "claim", "--goal-id", args.goal_id])
@@ -393,7 +396,7 @@ def recovery(
         if args.dry_run:
             cli_args.append("--dry-run")
         return {
-            "command": "loopx todo claim",
+            "command": shlex.join([*route, "todo", "claim"]),
             "cli_args": cli_args,
             "requires_flags": missing,
             "remove_flags": unsupported_todo_options(args, allowed_fields=_TODO_CLAIM_FIELDS),

tests/control_plane/test_cli_program_route.py

@@ -1,7 +1,6 @@
 """An explicitly selected runtime must execute its own generated control loop."""
 from __future__ import annotations
 
-import json
 import os
 import shlex
 import subprocess
@@ -11,6 +10,8 @@
 import pytest
 import test_quota_authority_settlement_journey as journey
 
+from codex_checkpoint_fixture import read_cli_policy_payload
+
 cli = journey.cli
 
 
@@ -54,7 +55,7 @@ def test_generated_commands_stay_on_candidate_through_once_only_settlement(
     def run(argv, *, expected_code=0):
         result = subprocess.run(argv, cwd=project, capture_output=True, text=True, timeout=60)
         assert result.returncode == expected_code, (result.returncode, result.stdout, result.stderr)
-        return json.loads(result.stdout)
+        return read_cli_policy_payload(result.stdout, returncode=result.returncode)
 
     prefix = [program, "--cli-bin", program, "--format", "json", "--registry", str(registry),
               "--runtime-root", str(runtime)]
@@ -71,12 +72,12 @@ def run(argv, *, expected_code=0):
         "--agent-id", cli.AGENT_ID, "--turn-instance-id", cli.TURN_ID,
         "--todo-id", "todo_not_projected",
     ], expected_code=1)
-    assert rejected["state"] == "quota_action_selection_rejected"
+    assert rejected["error_code"] == "quota_action_selection_rejected"
     assert not rejected["ok"] and not rejected["should_run"]
     assert rejected["heartbeat_receipt"]["status"] == "not_committed"
     assert cli._heartbeat_receipt_count(runtime, cli.TURN_ID) == 0
     assert cli._spend_run_count(runtime) == 0
-    [recovery] = rejected["interaction_contract"]["cli_channel"]["next_cli_actions"]
+    [recovery] = rejected["next_cli_actions"]
     recovery_args = shlex.split(recovery)
     # Execute the returned recovery unchanged; a lost route must hit the
     # poisoned global executable instead of being repaired by the test.

tests/control_plane/test_task_lease_argument_recovery.py

@@ -2,6 +2,8 @@
 from __future__ import annotations
 
 import json
+import os
+import shlex
 import subprocess
 import sys
 from pathlib import Path
@@ -87,6 +89,66 @@ def test_renew_repair_preserves_identity_and_exact_version(lease_cli):
     assert run(action("inspect"))["lease"] == renewed["lease"]
 
 
+@pytest.mark.skipif(os.name == "nt", reason="Exercises a POSIX console alias and PATH")
+def test_explicit_runtime_survives_grammar_repair_and_refusal_inspection(
+    lease_cli, tmp_path, monkeypatch,
+):
+    run, action, _, _ = lease_cli
+    program = "candidate's cli"
+    bin_dir = tmp_path / "candidate bin"
+    bin_dir.mkdir()
+    console = Path(sys.executable).parent / "loopx"
+    assert console.is_file(), "Use the checkout's installed test environment"
+    (bin_dir / program).symlink_to(console)
+    marker = tmp_path / "wrong-runtime"
+    poison = bin_dir / "loopx"
+    poison.write_text(f"#!{sys.executable}\nfrom pathlib import Path\n"
+                      f"Path({str(marker)!r}).write_text('wrong runtime')\nraise SystemExit(97)\n")
+    poison.chmod(0o755)
+    monkeypatch.setenv("PATH", str(bin_dir) + os.pathsep + os.environ.get("PATH", ""))
+
+    def execute(argv):
+        result = subprocess.run(argv, cwd=ROOT, capture_output=True, text=True, timeout=90)
+        assert result.returncode == 0, (result.returncode, result.stdout, result.stderr)
+        assert not marker.exists(), "A generated command escaped to the global runtime"
+        return json.loads(result.stdout)
+
+    def markdown_command(arguments, label):
+        arguments = list(arguments)
+        arguments[arguments.index("--format") + 1] = "markdown"
+        result = subprocess.run([sys.executable, "-m", "loopx.cli", "--cli-bin", program, *arguments],
+                                cwd=ROOT, capture_output=True, text=True, timeout=90)
+        assert result.returncode == 1, result.stdout + result.stderr
+        line = next(line for line in result.stdout.splitlines() if line.startswith(f"- {label}: `"))
+        return shlex.split(line.split("`", 2)[1])
+
+    before = run(action("inspect"))["lease"]
+    malformed = action("inspect", "--owner", "peer-a")
+    # Execute the displayed repair unchanged; a hard-coded executable hits poison.
+    assert execute(markdown_command(malformed, "repair"))["lease"] == before
+    rejected = run(["--cli-bin", program, *action(
+        "renew", "--owner", "peer-a", "--idempotency-key", "original",
+        "--expected-version", "1", "--write-scope", "other/**")], code=1)
+    assert run(action("inspect"))["lease"] == before
+    repair = rejected["recovery"]
+    # cli_args remains option-only; its existing command supplies the executable.
+    repair_argv = [shlex.split(repair["command"])[0], *repair["cli_args"]]
+    renewed = execute(repair_argv)["lease"]
+    assert renewed["version"] == before["version"] + 1
+    assert renewed["owner"] == before["owner"] and renewed["write_scopes"] == before["write_scopes"]
+    assert execute(repair_argv)["lease"] == renewed
+    wrong_owner = action("release", "--owner", "peer-b", "--idempotency-key", "original",
+                         "--expected-version", str(renewed["version"]))
+    denied = run(["--cli-bin", program, *wrong_owner], code=1)
+    assert denied["ok"] is False
+    inspection = denied["inspection"]
+    assert inspection["execution_authority_granted"] is False
+    assert inspection["cli_args"][inspection["cli_args"].index("--cli-bin") + 1] == program
+    assert execute([program, *inspection["cli_args"]])["lease"] == renewed
+    assert execute(markdown_command(wrong_owner, "inspect current state"))["lease"] == renewed
+    assert run(action("inspect"))["lease"] == renewed
+
+
 def test_missing_inputs_are_combined_and_wrong_owner_is_not_repaired(lease_cli):
     run, action, _, _ = lease_cli
     before = run(action("inspect"))["lease"]

tests/control_plane/test_todo_claim_argument_recovery.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import json
+import os
 import shlex
 import subprocess
 import sys
@@ -81,6 +82,7 @@ def test_one_packet_repairs_all_grammar_errors_and_preserves_route(claim_cli):
     assert snapshot() == before
     assert rejected["error_code"] == "todo_claim_invalid_arguments"
     recovery = rejected["recovery"]
+    assert recovery["command"] == "loopx todo claim"
     assert recovery["requires_flags"] == ["--claimed-by"]
     assert recovery["remove_flags"] == ["--turn-instance-id", "--note"]
     repaired = recovery["cli_args"]
@@ -196,6 +198,8 @@ def test_ordinary_claim_progress_completion_need_no_lease_flags(claim_cli):
     assert all(action in rendered for action in accepted["next_cli_actions"])
     assert accepted["claim_operation_id"] == "managed-claim"
     assert len(accepted["next_cli_actions"]) == 2
+    assert all(shlex.split(action)[0] == "loopx" and "--cli-bin" not in shlex.split(action)
+               for action in accepted["next_cli_actions"])
     assert all("--claim-operation-id managed-claim" in action for action in accepted["next_cli_actions"])
     assert all("task-lease" not in action for action in accepted["next_cli_actions"])
     common = ["--goal-id", GOAL, "--todo-id", TODO, "--agent-id", "peer-a", "--claim-operation-id", "managed-claim"]
@@ -212,6 +216,65 @@ 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
+    program = "candidate's cli"
+    binaries = tmp_path / "candidate bin"
+    binaries.mkdir()
+    (binaries / program).symlink_to(Path(sys.executable).parent / "loopx")
+    marker = tmp_path / "wrong-runtime"
+    poison = binaries / "loopx"
+    poison.write_text(f"#!{sys.executable}\nfrom pathlib import Path\n"
+                      f"Path({str(marker)!r}).write_text('wrong runtime')\nraise SystemExit(97)\n")
+    poison.chmod(0o755)
+    monkeypatch.setenv("PATH", str(binaries) + os.pathsep + os.environ.get("PATH", ""))
+    # The external cwd must import this candidate, even with an ambient installed package.
+    monkeypatch.setenv("PYTHONPATH", str(REPO))
+
+    def execute(argv, expected_exit=0):
+        result = subprocess.run(argv, cwd=project, capture_output=True, text=True, timeout=90)
+        assert result.returncode == expected_exit, result.stdout + result.stderr
+        assert not marker.exists()
+        return json.loads(result.stdout)
+
+    before = snapshot()
+    rejected = run(["--cli-bin", program, *prefix, *claim, "--note", "invalid"], expected_exit=1)
+    assert snapshot() == before
+    recovery = rejected["recovery"]
+    assert recovery["requires_flags"] == ["--claimed-by"]
+    assert recovery["remove_flags"] == ["--note"]
+    assert shlex.split(recovery["command"]) == [program, "--cli-bin", program, "todo", "claim"]
+    repaired = [program, *recovery["cli_args"], "--claimed-by", "peer-a",
+                "--claim-operation-id", "selected-runtime-claim"]
+    accepted = execute(repaired)
+    assert accepted["execution_managed"] is True
+    from loopx.control_plane.todos.markdown import render_todo_markdown
+    assert all(action in render_todo_markdown(accepted) for action in accepted["next_cli_actions"])
+    actions = [shlex.split(action) for action in accepted["next_cli_actions"]]
+    assert all(argv[:3] == [program, "--cli-bin", program] for argv in actions)
+    assert all(argv[argv.index("--claim-operation-id") + 1] == "selected-runtime-claim"
+               for argv in actions)
+    progress = [{"<progress-operation-id>": "selected-runtime-progress", "<progress>": "Validation passed"}
+                .get(part, part) for part in actions[0]]
+    claimed = snapshot()
+    wrong_actor = ["peer-b" if i and progress[i - 1] == "--agent-id" else part
+                   for i, part in enumerate(progress)]
+    assert execute(wrong_actor, expected_exit=1)["ok"] is False
+    assert snapshot() == claimed
+    assert execute(progress)["ok"] is True
+    progressed = snapshot()
+    assert execute(progress)["ok"] is True
+    assert snapshot() == progressed
+    complete = ["evidence:validated-output" if part == "<validated-evidence>" else part
+                for part in actions[1]]
+    assert execute(complete)["ok"] is True
+    completed = snapshot()
+    assert completed[0][0]["status"] == "done"
+    assert execute(complete)["ok"] is True
+    assert snapshot() == completed
+
+
 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,