perf(cli): reuse selected dispatch for handoff commands

docs/architecture/rfcs/typescript-control-plane-migration-v0.md

@@ -68,6 +68,19 @@ on their established paths. This adds no domain rule, bridge operation or state
 cache. Complete command cost and real write/recovery behavior qualify separately
 from import timing; installed and live-provider behavior need their own evidence.
 
+### Handoff CLI caller adoption
+
+`handoff` and `handoff-mode` use the existing selected parser and shared
+dispatcher. The full compatibility CLI reaches the same unchanged handlers;
+its separate continuation and mode dispatch blocks are retired. Each command
+joins the existing executable-revision request scope, while registry, provider,
+workspace and authority observations remain live. Existing help, diagnostics,
+formats, native-controller guards and operation replay identities remain with
+their current owners. This adds no module, semantic owner, bridge operation or
+data cache. It is bounded Stage 3 caller adoption; complete CLI cost and real
+handoff/policy recovery qualify separately from the full native CLI and host
+or provider promotion.
+
 ### Product conversations and asynchronous inbox
 
 **Chat HTTP POST scope.** `ChatRequestHandler.do_POST` now joins the existing

docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md

@@ -60,6 +60,16 @@ registry/runtime 选择和 native-controller guard 保持既有路径。这不
 桥接操作或状态缓存。完整命令成本及真实写入／恢复行为与 import 计时分别验证；
 安装后行为和真实 provider 行为仍需各自的证据。
 
+### Handoff CLI 调用者接入
+
+`handoff` 和 `handoff-mode` 复用现有 selected parser 与共享 dispatcher。
+完整兼容 CLI 调用相同且未改动的处理器，原有 continuation 和 mode 分发块已删除。
+每条命令进入既有可执行版本请求 scope；registry、provider、workspace 和 authority
+观察仍实时读取。帮助、诊断、格式、native-controller guard 和操作重放身份保留
+原有所有者。不新增 module、语义所有者、bridge operation 或数据 cache。这是
+有界 Stage 3 调用者接入；完整 CLI 成本及真实 handoff／策略恢复，与完整 native CLI、
+Host 或 provider 晋升分别资格化。
+
 ### 产品对话与通用异步 inbox
 
 **Chat HTTP POST 作用域。** `ChatRequestHandler.do_POST` 现在为每次完整 HTTP

loopx/cli.py

@@ -22,7 +22,7 @@
 from .capabilities.multi_subagent.cli import (
     register_native_child_commands, handle_native_child_command,
 )
-from .cli_commands.todo_continuation import register_todo_continuation, handle_todo_continuation
+from .cli_commands.todo_continuation import register_todo_continuation
 from .cli_commands.manager_inbox import register_manager_inbox
 from .cli_commands.delegation import register_delegation
 from .capabilities.content_ops.cli import (
@@ -117,7 +117,6 @@
     handle_registry_admin_command,
     handle_starter_command,
     handle_support_control_command,
-    handle_handoff_mode_command,
     handle_task_lease_command,
     handle_authority_shadow_command,
     handle_version_command,
@@ -852,23 +851,6 @@ def _main(argv: list[str] | None = None) -> int:
     if authority_shadow_result is not None:
         return authority_shadow_result
 
-    continuation_result = handle_todo_continuation(
-        args, registry_path=registry_path, runtime_root_arg=args.runtime_root,
-        output_format=output_format, print_payload=print_payload,
-    )
-    if continuation_result is not None:
-        return continuation_result
-
-    handoff_mode_result = handle_handoff_mode_command(
-        args,
-        registry_path=registry_path,
-        output_format=output_format,
-        print_payload=print_payload,
-        runtime_root_arg=args.runtime_root,
-    )
-    if handoff_mode_result is not None:
-        return handoff_mode_result
-
     goal_acceptance_result = handle_goal_acceptance_command(
         args, registry_path=registry_path, runtime_root_arg=args.runtime_root,
         output_format=output_format, print_payload=print_payload,

loopx/cli_commands/todo_continuation.py

@@ -1,4 +1,8 @@
 """Host CLI adapter; all continuation and ownership decisions stay in TypeScript."""
+from __future__ import annotations
+
+import argparse
+from collections.abc import Callable
 from pathlib import Path
 import json
 
@@ -90,7 +94,10 @@ def _render_digest(payload: dict) -> str:
     return "\n".join(lines)
 
 
-def register_todo_continuation(subparsers, add_format):
+def register_todo_continuation(
+    subparsers: argparse._SubParsersAction[argparse.ArgumentParser],
+    add_format: Callable[[argparse.ArgumentParser], None],
+) -> None:
     parser = subparsers.add_parser(
         "handoff", help="Explicit cross-agent Todo handoff: prepare/inspect/adopt ownership, or restore content without authority changes."
     )
@@ -118,7 +125,11 @@ def register_todo_continuation(subparsers, add_format):
         help="Output format. 'digest' renders inspect as a readable handoff summary (inspect only).")
 
 
-def handle_todo_continuation(args, *, registry_path, runtime_root_arg, output_format, print_payload):
+def handle_todo_continuation(
+    args: argparse.Namespace, *, registry_path: Path, runtime_root_arg: str | None,
+    output_format: Callable[[argparse.Namespace], str],
+    print_payload: Callable[[dict[str, object], str, Callable[[dict[str, object]], str]], None],
+) -> int | None:
     if args.command != "handoff":
         return None
     if args.action == "restore":

loopx/cli_runtime.py

@@ -75,6 +75,7 @@
 	"authority-archive", "coordination-shadow", "extension", "slash-commands",
 	"manager-inbox",
 	"agent-context", "agent-directory", "agent-capabilities",
+	"handoff", "handoff-mode",
 }
 
 
@@ -317,6 +318,14 @@ def _build_selected_parser(command: str) -> LoopXArgumentParser:
 		from .cli_commands.task_lease import register_task_lease_command
 
 		register_task_lease_command(subparsers, add_subcommand_format)
+	elif command == "handoff":
+		from .cli_commands.todo_continuation import register_todo_continuation
+
+		register_todo_continuation(subparsers, add_subcommand_format)
+	elif command == "handoff-mode":
+		from .cli_commands.handoff_mode import register_handoff_mode_command
+
+		register_handoff_mode_command(subparsers, add_subcommand_format)
 	elif command == "change-window":
 		from .capabilities.repository_change_window.cli import register_repository_change_window_commands
 
@@ -427,6 +436,15 @@ def _dispatch_common_command(
 			args, registry_path=registry_path, runtime_root_arg=args.runtime_root,
 			output_format=output_format, print_payload=print_payload,
 		)
+	if args.command in {"handoff", "handoff-mode"}:
+		if args.command == "handoff":
+			from .cli_commands.todo_continuation import handle_todo_continuation as handoff_handler
+		else:
+			from .cli_commands.handoff_mode import handle_handoff_mode_command as handoff_handler
+		return handoff_handler(
+			args, registry_path=registry_path, runtime_root_arg=args.runtime_root,
+			output_format=output_format, print_payload=print_payload,
+		)
 	if args.command == "authority-archive":
 		from .cli_commands.authority_archive import handle_authority_archive_command
 

tests/control_plane/test_canonical_handoff_mode.py

@@ -10,9 +10,14 @@
 from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
 
 
-@pytest.fixture(params=["file", "sqlite"])
+@pytest.fixture(params=[
+    pytest.param((provider, module), id=f"{provider}-{module}")
+    for provider in ("file", "sqlite")
+    for module in ("loopx.entrypoint", "loopx.cli")
+])
 def canonical_mode(tmp_path, monkeypatch, request):
-    if request.param == "sqlite":
+    provider, module = request.param
+    if provider == "sqlite":
         isolate_sqlite_runtime(tmp_path, monkeypatch)
     state = tmp_path / "state.md"
     state.write_text("---\nhandoff_mode: hard_lease\n---\n\n## Agent Todo\n")
@@ -22,10 +27,10 @@ def canonical_mode(tmp_path, monkeypatch, request):
         {"id": "mode-goal", "repo": str(tmp_path), "state_file": state.name,
          "coordination": {"registered_agents": ["agent-a", "agent-b"]}}]}))
     projection = build_todo_runtime_shadow_projection(goal_id="mode-goal", handoff_mode="soft_claim", todos=[])
-    initialize_canonical_authority(runtime, "mode-goal", projection, state_path=state, provider=request.param)
-    def cli(*args):
-        result = subprocess.run([sys.executable, "-m", "loopx.cli", "--registry", str(registry),
-            "--format", "json", "handoff-mode", *args, "--goal-id", "mode-goal"],
+    initialize_canonical_authority(runtime, "mode-goal", projection, state_path=state, provider=provider)
+    def cli(*args, family="handoff-mode"):
+        result = subprocess.run([sys.executable, "-m", module, "--registry", str(registry),
+            "--format", "json", family, *args, "--goal-id", "mode-goal"],
             capture_output=True, text=True, timeout=90)
         if result.returncode:
             # Keep the complete native refusal in pytest's captured output;
@@ -37,6 +42,41 @@ def cli(*args):
     return state, runtime, cli
 
 
+def test_handoff_prepare_adopt_and_replay_keep_the_original_todo(canonical_mode):
+    state, runtime, cli = canonical_mode
+    from loopx.todos import add_goal_todo
+    from loopx.control_plane.coordination.local_authority import read_canonical_todos_if_promoted
+
+    created = add_goal_todo(
+        registry_path=state.parent / "registry.json", goal_id="mode-goal", role="agent",
+        text="[P1] Preserve the complete continuation requirement.", claimed_by="agent-a",
+        agent_id="agent-a", operation_id="handoff-source",
+    )
+    todo_id = created["todo_id"]
+    args = ["--todo-id", todo_id, "--agent-id", "agent-a", "--workspace", str(state.parent)]
+    code, initial = cli("inspect", *args, "--session-id", "source-session", family="handoff")
+    assert code == 0 and initial["note_state"] == "missing", initial
+    code, prepared = cli(
+        "prepare", *args, "--session-id", "source-session", "--operation-id", "prepare-source",
+        "--expected-revision", initial["provider_revision"], "--rationale", "Keep the accepted validation route.",
+        "--source-ref", "artifact:validation", family="handoff",
+    )
+    assert code == 0 and prepared["ok"] is True, prepared
+    code, inspected = cli("inspect", *args, "--session-id", "target-session", family="handoff")
+    assert code == 0 and inspected["can_adopt"] is True and inspected["todo_id"] == todo_id, inspected
+    adoption = ["adopt", *args, "--session-id", "target-session", "--operation-id", "adopt-source",
+                "--expected-revision", inspected["provider_revision"]]
+    code, adopted = cli(*adoption, family="handoff")
+    assert code == 0 and adopted["current_authority_verified"] is True, adopted
+    code, replayed = cli(*adoption, family="handoff")
+    assert code == 0 and replayed["current_authority_verified"] is True, replayed
+    current = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="mode-goal", include_leases=True)
+    assert [(todo["todo_id"], todo["text"], todo["claimed_by"]) for todo in current["todos"]] == [
+        (todo_id, "[P1] Preserve the complete continuation requirement.", "agent-a"),
+    ]
+    assert current["leases"] == []
+
+
 def test_canonical_show_ignores_stale_or_missing_display(canonical_mode):
     state, _, cli = canonical_mode
     before = state.read_bytes()

tests/control_plane/test_goal_handoff_mode_cli.py

@@ -1,4 +1,4 @@
-"""CLI coverage for canonical handoff mode and pre-cutoff read compatibility."""
+"""CLI coverage for canonical handoff mode and refusal of unpromoted state."""
 
 from __future__ import annotations
 
@@ -11,7 +11,6 @@
 from loopx.cli import main as cli_main
 from loopx.control_plane.todos.handoff_mode import (
     HANDOFF_MODE_HARD_LEASE,
-    HANDOFF_MODE_LEGACY,
     HANDOFF_MODE_SOFT_CLAIM,
 )
 from loopx.control_plane.work_items.task_lease import acquire_task_lease
@@ -111,20 +110,26 @@ def _add_todo(registry: Path, *, claimed_by: str | None = None) -> dict[str, Any
 # ---------------------------------------------------------------------------
 
 
-def test_handoff_mode_show_defaults_to_legacy(
+@pytest.mark.parametrize("handoff_mode", [None, "banana"])
+def test_unpromoted_show_requires_canonical_authority_without_mutation(
     tmp_path: Path,
     capsys: pytest.CaptureFixture[str],
+    handoff_mode: str | None,
 ) -> None:
-    registry, _state = _write_workspace(tmp_path)
+    registry, state = _write_workspace(tmp_path, handoff_mode=handoff_mode)
+    before = state.read_bytes()
 
     exit_code, payload = _run_cli(
         capsys, registry, "handoff-mode", "show", "--goal-id", GOAL_ID
     )
 
-    assert exit_code == 0
-    assert payload["ok"] is True
-    assert payload["handoff_mode"] == HANDOFF_MODE_LEGACY
-    assert payload["source"] == "default"
+    assert exit_code == 1
+    assert payload["ok"] is False
+    assert payload["error_code"] == "canonical_authority_required"
+    assert payload["decision_read_from_provider"] is False
+    assert payload["legacy_fallback_used"] is False
+    assert payload["recovery"] == {"action": "migrate_goal_authority", "backup_required": True}
+    assert state.read_bytes() == before
 
 
 def test_unpromoted_legacy_set_is_retired_without_mutation(tmp_path, capsys):
@@ -144,24 +149,6 @@ def test_unpromoted_legacy_set_is_retired_without_mutation(tmp_path, capsys):
     assert state.read_bytes() == before
 
 
-def test_handoff_mode_show_rejects_invalid_frontmatter_without_mutation(
-    tmp_path: Path,
-    capsys: pytest.CaptureFixture[str],
-) -> None:
-    registry, state = _write_workspace(tmp_path, handoff_mode="banana")
-    before = state.read_bytes()
-
-    exit_code, payload = _run_cli(
-        capsys, registry, "handoff-mode", "show", "--goal-id", GOAL_ID
-    )
-
-    assert exit_code == 1
-    assert payload["ok"] is False
-    assert payload["error_code"] == "invalid_handoff_mode"
-    assert payload["handoff_mode"] == "banana"
-    assert state.read_bytes() == before
-
-
 def test_handoff_mode_set_on_quiescent_goal_updates_provider(
     tmp_path: Path,
     capsys: pytest.CaptureFixture[str],

tests/test_cli_entrypoint.py

@@ -277,6 +277,8 @@ def test_command_reference_does_not_require_default_runtime_route(
 		(["review-packet", "--help"], "loopx.cli_commands.status_registration", "loopx.cli_commands.status"),
 		(["quota", "--help"], "loopx.cli_commands.quota_registration", "loopx.cli_commands.quota"),
 		(["task-lease", "--help"], "loopx.cli_commands.task_lease", "loopx.cli_commands.task_lease"),
+		(["handoff", "--help"], "loopx.cli_commands.todo_continuation", "loopx.cli_commands.todo_continuation"),
+		(["handoff-mode", "--help"], "loopx.cli_commands.handoff_mode", "loopx.cli_commands.handoff_mode"),
 		(["todo", "--help"], "loopx.cli_commands.todo_registration", "loopx.cli_commands.todo"),
 		(["delegation", "--help"], "loopx.cli_commands.delegation", "loopx.cli_commands.delegation"),
 		(["agent-context", "--help"], "loopx.cli_commands.agent_context", "loopx.cli_commands.agent_context"),
@@ -552,6 +554,52 @@ def test_selected_manager_inbox_refusals_match_without_loading_full_cli(tmp_path
 	assert json.loads(selected.stdout)["ok"] is False
 
 
+@pytest.mark.parametrize("arguments", [
+	["handoff", "inspect"],
+	["handoff", "prepare", "--goal-id", "perf-goal"],
+	["handoff", "inspect", "--format", "digest"],
+	["handoff-mode", "show", "--goal-id", "perf-goal", "--mode", "hard_lease"],
+	["handoff-mode", "migrate", "--goal-id", "perf-goal"],
+])
+def test_selected_handoff_refusals_match_without_loading_full_cli(tmp_path, arguments):
+	registry, runtime_root = write_command_fixture(tmp_path)
+	argv = ["--registry", str(registry), "--runtime-root", str(runtime_root), "--format", "json", *arguments]
+	before = registry.read_bytes()
+	selected = run_cli_main("loopx.entrypoint", argv, forbidden_modules=("loopx.cli",))
+	full = run_cli_main("loopx.cli", argv)
+	assert selected.returncode == full.returncode == 1
+	assert selected.stdout == full.stdout
+	assert selected.stderr == full.stderr
+	assert registry.read_bytes() == before
+
+
+def test_selected_handoff_parser_diagnostics_match_full_cli():
+	cases = [
+		["handoff"], ["handoff", "unknown"], ["handoff-mode"], ["handoff-mode", "unknown"],
+		["handoff", "inspect", "--format", "yaml"],
+		["handoff-mode", "set", "--mode", "invalid"],
+		["handoff-mode", "show", "--goal-id", "demo", "--unrecognized"],
+	]
+	assert run_cli_batch("loopx.entrypoint", cases) == run_cli_batch("loopx.cli", cases)
+
+
+@pytest.mark.parametrize("input_format", ["json", "markdown"])
+@pytest.mark.parametrize("format_name", ["json", "markdown"])
+def test_selected_handoff_restore_preserves_content_without_authority(tmp_path, input_format, format_name):
+	text = "保持完整要求。\n\nValidate the original Todo.\n"
+	source = tmp_path / "handoff.txt"
+	source.write_text(json.dumps({"ok": True, "handoff_text": text}) if input_format == "json" else text, encoding="utf-8")
+	registry, runtime = tmp_path / "absent-registry.json", tmp_path / "absent-runtime"
+	argv = ["--registry", str(registry), "--runtime-root", str(runtime), "handoff", "restore",
+		"--input", str(source), "--input-format", input_format, "--format", format_name]
+	selected = run_cli_main("loopx.entrypoint", argv, forbidden_modules=("loopx.cli",))
+	full = run_cli_main("loopx.cli", argv)
+	assert selected.returncode == full.returncode == 0
+	assert selected.stdout == full.stdout
+	assert (json.loads(selected.stdout)["handoff_text"] if format_name == "json" else selected.stdout) == text
+	assert not registry.exists() and not runtime.exists()
+
+
 @pytest.mark.parametrize("arguments", [
 	["refresh-state", "--dry-run"],
 	["checkpoint-context", "--agent-id", "perf-agent", "--turn-instance-id", "missing", "--todo-id", "todo_missing"],