perf(cli): share selected agent collaboration dispatch

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

@@ -45,6 +45,17 @@ resolution, guards, formatting and typed owners. This adds no domain rule, bridg
 operation or data cache. Incremental qualification accounts for pending quota
 request-scope adoption separately rather than counting its gain again.
 
+### Agent collaboration CLI caller adoption
+
+`agent-context`, `agent-directory` and `agent-capabilities` now use the existing
+selected parser and shared dispatcher. Full-parser compatibility callers use
+the same dispatcher; their old handler branches are removed. Each command keeps
+its original registration, handler, registry/runtime resolution, native-controller
+guard and output. The dispatcher reuses the existing executable-revision request
+scope; source and capability observations remain live. No new module, bridge,
+semantic owner or state cache is introduced. This is bounded caller adoption;
+it does not qualify peer execution, model outcomes or the complete native CLI.
+
 ### 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

@@ -41,6 +41,15 @@ SQLite CLI 性能。仅减少请求次数不能证明性能收益。
 guard、格式和 typed owner。这不新增领域规则、桥接操作或数据缓存。增量资格验证
 单独计入待合入的 quota 请求作用域接入，避免重复计算其收益。
 
+### Agent 协作 CLI 调用者接入
+
+`agent-context`、`agent-directory` 和 `agent-capabilities` 现在使用既有按命令 parser
+和共享 dispatcher。完整 parser 的兼容调用者复用相同 dispatcher，旧 handler 分支
+已删除。各命令保留原有注册、handler、registry/runtime 解析、native-controller
+guard 和输出。Dispatcher 复用既有 executable-revision 请求作用域；源数据与
+capability 观察仍实时读取。这不新增 module、桥接、语义 owner 或状态缓存。
+这是有界调用者接入，不证明 peer 执行、模型结果或完整原生 CLI 已通过资格验证。
+
 ### 产品对话与通用异步 inbox
 
 **Chat HTTP POST 作用域。** `ChatRequestHandler.do_POST` 现在为每次完整 HTTP

loopx/cli.py

@@ -16,9 +16,9 @@
 import argparse
 import sys
 
-from .cli_commands.agent_capabilities import register_agent_capabilities, handle_agent_capabilities
-from .cli_commands.agent_directory import register_agent_directory, handle_agent_directory
-from .cli_commands.agent_context import register_agent_context, handle_agent_context
+from .cli_commands.agent_capabilities import register_agent_capabilities
+from .cli_commands.agent_directory import register_agent_directory
+from .cli_commands.agent_context import register_agent_context
 from .capabilities.multi_subagent.cli import (
     register_native_child_commands, handle_native_child_command,
 )
@@ -792,33 +792,12 @@ def _main(argv: list[str] | None = None) -> int:
     if lark_kanban_result is not None:
         return lark_kanban_result
 
-    if args.command == "agent-capabilities":
-        return handle_agent_capabilities(
-            args, registry_path, effective_runtime_root(registry_path, args.runtime_root),
-            print_payload, output_format,
-        )
-
-    if args.command == "agent-context":
-        return handle_agent_context(
-            args,
-            registry_path,
-            effective_runtime_root(registry_path, args.runtime_root),
-            print_payload,
-            output_format,
-        )
-
     if args.command == "native-child":
         return handle_native_child_command(
             args, registry_path, effective_runtime_root(registry_path, args.runtime_root),
             print_payload, output_format,
         )
 
-    if args.command == "agent-directory":
-        return handle_agent_directory(
-            args, registry_path, effective_runtime_root(registry_path, args.runtime_root),
-            print_payload, output_format,
-        )
-
     lark_inbox_result = handle_lark_inbox_command(
         args,
         registry_path=registry_path,

loopx/cli_runtime.py

@@ -71,6 +71,7 @@
 	"todo", "quota", "task-lease", "change-window", "delegation", "turn", "doctor", "commands",
 	"authority-archive", "coordination-shadow", "extension", "slash-commands",
 	"manager-inbox",
+	"agent-context", "agent-directory", "agent-capabilities",
 }
 
 
@@ -325,6 +326,18 @@ def _build_selected_parser(command: str) -> LoopXArgumentParser:
 		from .cli_commands.manager_inbox import register_manager_inbox
 
 		register_manager_inbox(subparsers, add_subcommand_format)
+	elif command == "agent-context":
+		from .cli_commands.agent_context import register_agent_context
+
+		register_agent_context(subparsers, add_subcommand_format)
+	elif command == "agent-directory":
+		from .cli_commands.agent_directory import register_agent_directory
+
+		register_agent_directory(subparsers, add_subcommand_format)
+	elif command == "agent-capabilities":
+		from .cli_commands.agent_capabilities import register_agent_capabilities
+
+		register_agent_capabilities(subparsers, add_subcommand_format)
 	elif command == "authority-archive":
 		from .cli_commands.authority_archive import register_authority_archive_command
 
@@ -360,6 +373,19 @@ def _dispatch_common_command(
 	registry_path: Path,
 	allow_missing_registry: bool,
 ) -> int | None:
+	if args.command in {"agent-context", "agent-directory", "agent-capabilities"}:
+		from .paths import effective_runtime_root
+
+		if args.command == "agent-context":
+			from .cli_commands.agent_context import handle_agent_context as handle
+		elif args.command == "agent-directory":
+			from .cli_commands.agent_directory import handle_agent_directory as handle
+		else:
+			from .cli_commands.agent_capabilities import handle_agent_capabilities as handle
+		return handle(
+			args, registry_path, effective_runtime_root(registry_path, args.runtime_root),
+			print_payload, output_format,
+		)
 	if args.command == "manager-inbox":
 		from .cli_commands.manager_inbox import handle_manager_inbox
 		from .control_plane.projects.registry_codec import load_project_registry

tests/test_cli_entrypoint.py

@@ -279,6 +279,9 @@ def test_command_reference_does_not_require_default_runtime_route(
 		(["task-lease", "--help"], "loopx.cli_commands.task_lease", "loopx.cli_commands.task_lease"),
 		(["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"),
+		(["agent-directory", "--help"], "loopx.cli_commands.agent_directory", "loopx.cli_commands.agent_directory"),
+		(["agent-capabilities", "--help"], "loopx.cli_commands.agent_capabilities", "loopx.cli_commands.agent_capabilities"),
 		(["goal-portfolio", "--help"], "loopx.cli_commands.summary_all_registration", "loopx.cli_commands.summary_all"),
 		(["global-summary", "--help"], "loopx.cli_commands.summary_all_registration", "loopx.cli_commands.summary_all"),
 		(["global-gates", "--help"], "loopx.cli_commands.summary_all_registration", "loopx.cli_commands.summary_all"),
@@ -465,12 +468,66 @@ def test_selected_parser_matches_full_help_and_diagnostics() -> None:
 	for command in ("goal-portfolio", "global-summary", "global-gates", "global-todos", "global-risks"):
 		argv_cases.extend([[command, "--help"], [command, "--unknown-option"], [command, "--limit", "invalid"]])
 	argv_cases.append(["goal-portfolio", "--manager-view", "unknown"])
+	for command in ("agent-context", "agent-directory", "agent-capabilities"):
+		argv_cases.extend([[command, "--help"], [command], [command, "--unknown-option"],
+			[command, "--goal-id", "goal", "--agent-i", "agent"]])
+	argv_cases.append(["agent-context", "--goal-id", "goal", "--agent-id", "agent", "--phase", "unknown"])
 	selected = run_cli_batch("loopx.entrypoint", argv_cases)
 	full = run_cli_batch("loopx.cli", argv_cases)
 
 	assert selected == full
 
 
+@pytest.mark.parametrize("module", ["loopx.entrypoint", "loopx.cli"])
+@pytest.mark.parametrize("command", ["agent-context", "agent-directory", "agent-capabilities"])
+def test_agent_commands_read_registered_peer_without_changing_goal(
+	tmp_path: Path, module: str, command: str,
+) -> None:
+	from tests.control_plane.canonical_authority_fixture import (
+		assert_todo_read_unchanged, initialize_canonical_source_fixture,
+	)
+	from loopx.control_plane.todos.list_readback import list_goal_todos
+
+	registry, runtime_root = write_command_fixture(tmp_path)
+	state_path = registry.parent.parent / ".codex/goals/perf-goal/ACTIVE_GOAL_STATE.md"
+	initialize_canonical_source_fixture(runtime_root, "perf-goal", state_path=state_path)
+	before = list_goal_todos(registry_path=registry, runtime_root_arg=str(runtime_root), goal_id="perf-goal")
+	registry_before = registry.read_bytes()
+	argv = ["--registry", str(registry), "--runtime-root", str(runtime_root),
+		"--format", "json", command, "--goal-id", "perf-goal", "--agent-id", "perf-agent"]
+	if command == "agent-context":
+		argv += ["--phase", "before_plan"]
+	completed = run_cli_main(module, argv,
+		forbidden_modules=("loopx.cli",) if module == "loopx.entrypoint" else ())
+	assert completed.returncode == 0, completed.stderr
+	payload = json.loads(completed.stdout)
+	assert payload["ok"] is True
+	if command == "agent-context":
+		assert payload["read_only"] is True
+		assert payload["agent_context"] is None
+	elif command == "agent-directory":
+		assert [row["agent_id"] for row in payload["rows"]] == ["perf-agent"]
+		assert payload["scope"]["caller_membership"] == "registered_agent"
+	else:
+		assert payload["agent_capabilities"]["agent_id"] == "perf-agent"
+		assert payload["agent_capabilities"]["goal_id"] == "perf-goal"
+		preview = run_cli_main(module, [*argv, "--unavailable-capability", "network"])
+		assert preview.returncode == 0, preview.stderr
+		assert json.loads(preview.stdout)["agent_capabilities"]["proposed"]["unavailable"] == ["network"]
+		assert not (runtime_root / "agent-runtime-capabilities").exists()
+		written = run_cli_main(module, [*argv, "--unavailable-capability", "network", "--execute"])
+		assert written.returncode == 0, written.stderr
+		assert json.loads(written.stdout)["agent_capabilities"]["written"] is True
+		other = "loopx.cli" if module == "loopx.entrypoint" else "loopx.entrypoint"
+		readback = run_cli_main(other, argv)
+		assert readback.returncode == 0, readback.stderr
+		assert json.loads(readback.stdout)["agent_capabilities"]["unavailable"] == ["network"]
+	assert registry.read_bytes() == registry_before
+	assert_todo_read_unchanged(
+		list_goal_todos(registry_path=registry, runtime_root_arg=str(runtime_root), goal_id="perf-goal"), before,
+	)
+
+
 @pytest.mark.parametrize("arguments", [
 	["acknowledge", "--request-id", "a" * 64, "--decision", "adopt", "--reason", "Checked the source"],
 	["report", "--request-id", "a" * 64, "--reply-text", "The evidence check passed."],