perf(cli): share selected lifecycle command dispatch

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

@@ -56,6 +56,18 @@ 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.
 
+### Lifecycle CLI caller adoption
+
+`refresh-state`, `checkpoint-context`, `read-only-map`, `reward` and
+`operator-gate` reuse the selected parser and shared dispatcher. Both CLI
+entrypoints use the existing lifecycle registration, handlers and writeback-hook
+composition; the old full-CLI dispatch branch is removed. The command set lives
+with routing and remains available through its lifecycle-module import. Existing
+help, diagnostics, registry/runtime selection and native-controller guards stay
+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.
+
 ### 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

@@ -50,6 +50,16 @@ guard 和输出。Dispatcher 复用既有 executable-revision 请求作用域；
 capability 观察仍实时读取。这不新增 module、桥接、语义 owner 或状态缓存。
 这是有界调用者接入，不证明 peer 执行、模型结果或完整原生 CLI 已通过资格验证。
 
+### Lifecycle CLI 调用者接入
+
+`refresh-state`、`checkpoint-context`、`read-only-map`、`reward` 和
+`operator-gate` 复用按命令 parser 和共享 dispatcher。两个 CLI 入口使用既有
+lifecycle 注册、handler 和 writeback-hook 组合；旧的完整 CLI 分发分支已删除。
+命令集合由路由持有，lifecycle module 的旧 import 仍可使用它。Help、诊断、
+registry/runtime 选择和 native-controller guard 保持既有路径。这不新增领域规则、
+桥接操作或状态缓存。完整命令成本及真实写入／恢复行为与 import 计时分别验证；
+安装后行为和真实 provider 行为仍需各自的证据。
+
 ### 产品对话与通用异步 inbox
 
 **Chat HTTP POST 作用域。** `ChatRequestHandler.do_POST` 现在为每次完整 HTTP

examples/cli-project-lifecycle-command-modularization-smoke.py

@@ -220,7 +220,6 @@ def main() -> None:
         "PROJECT_LIFECYCLE_COMMANDS",
         "register_project_lifecycle_commands",
         "handle_project_lifecycle_command",
-        "refresh-state",
         "read-only-map",
         "reward",
         "operator-gate",
@@ -240,7 +239,9 @@ def main() -> None:
     ):
         require(marker in sink_source, f"project lifecycle sink module missing {marker}")
     require("register_project_lifecycle_commands" in cli_source, "cli.py did not register project lifecycle commands")
-    require("handle_project_lifecycle_command" in cli_source, "cli.py did not dispatch project lifecycle commands")
+    runtime_source = (CLI.parent / "cli_runtime.py").read_text(encoding="utf-8")
+    require("dispatch_common_command" in cli_source and "handle_project_lifecycle_command" in runtime_source,
+            "CLI entrypoints did not share project lifecycle dispatch")
     require("register_project_lifecycle_commands" in init_source, "__init__ did not export project lifecycle registration")
     require("handle_project_lifecycle_command" in init_source, "__init__ did not export project lifecycle handler")
 

loopx/cli.py

@@ -69,10 +69,6 @@
     handle_machine_configuration_command,
     register_machine_configuration_commands,
 )
-from .capabilities.periodic_report.post_writeback_hook import (
-    build_periodic_report_post_writeback_projection,
-    periodic_report_post_writeback_hooks_for_goal,
-)
 from .paths import (
     effective_runtime_root,
 )
@@ -114,7 +110,6 @@
     handle_presentation_command,
     handle_dash_command,
     handle_project_command,
-    handle_project_lifecycle_command,
     handle_pr_review_command,
     handle_deepresearch_command,
     handle_ready_score_command,
@@ -196,7 +191,6 @@
     handle_reliability_diagnostics_command,
     register_reliability_diagnostics_commands,
 )
-from .cli_rollout import append_cli_rollout_event
 from .capabilities.project_skill_delivery.cli import (
     handle_project_skill_command,
     register_project_skill_commands,
@@ -745,33 +739,6 @@ def _main(argv: list[str] | None = None) -> int:
             print_payload=print_payload,
         )
 
-    project_lifecycle_result = handle_project_lifecycle_command(
-        args,
-        registry_path=registry_path,
-        print_payload=print_payload,
-        output_format=output_format,
-        append_cli_rollout_event=append_cli_rollout_event,
-        post_writeback_hooks=(
-            periodic_report_post_writeback_hooks_for_goal(
-                registry_path=registry_path,
-                goal_id=args.goal_id,
-                runtime_root=effective_runtime_root(
-                    registry_path,
-                    args.runtime_root,
-                ),
-            )
-            if args.command == "refresh-state"
-            else ()
-        ),
-        post_writeback_projection_builder=(
-            build_periodic_report_post_writeback_projection
-            if args.command == "refresh-state"
-            else None
-        ),
-    )
-    if project_lifecycle_result is not None:
-        return project_lifecycle_result
-
     goal_channel_result = handle_goal_channel_command(
         args,
         registry_path=registry_path,

loopx/cli_commands/project_lifecycle.py

@@ -4,6 +4,7 @@
 from collections.abc import Callable, Sequence
 from pathlib import Path
 
+from ..cli_runtime import PROJECT_LIFECYCLE_COMMANDS as PROJECT_LIFECYCLE_COMMANDS
 from ..control_plane.capability_hooks import (
     PostWritebackHookRegistration,
 )
@@ -40,15 +41,6 @@
 OutputFormat = Callable[[argparse.Namespace], str]
 AppendCliRolloutEvent = Callable[..., dict[str, object]]
 
-PROJECT_LIFECYCLE_COMMANDS = {
-    "refresh-state",
-    "checkpoint-context",
-    "read-only-map",
-    "reward",
-    "operator-gate",
-}
-
-
 def register_project_lifecycle_commands(
     subparsers: argparse._SubParsersAction,
     add_subcommand_format: Callable[[argparse.ArgumentParser], None],

loopx/cli_runtime.py

@@ -67,7 +67,10 @@
 )
 
 _STATUS_COMMANDS = frozenset({"check", "status", "diagnose", "review-packet"})
-_SELECTED_COMMANDS = _STATUS_COMMANDS | GLOBAL_READ_COMMANDS | {
+PROJECT_LIFECYCLE_COMMANDS = {
+	"refresh-state", "checkpoint-context", "read-only-map", "reward", "operator-gate",
+}
+_SELECTED_COMMANDS = _STATUS_COMMANDS | GLOBAL_READ_COMMANDS | PROJECT_LIFECYCLE_COMMANDS | {
 	"todo", "quota", "task-lease", "change-window", "delegation", "turn", "doctor", "commands",
 	"authority-archive", "coordination-shadow", "extension", "slash-commands",
 	"manager-inbox",
@@ -298,6 +301,10 @@ def _build_selected_parser(command: str) -> LoopXArgumentParser:
 		from .cli_commands.summary_all_registration import register_summary_all_command
 
 		register_summary_all_command(subparsers, add_subcommand_format)
+	elif command in PROJECT_LIFECYCLE_COMMANDS:
+		from .cli_commands.project_lifecycle import register_project_lifecycle_commands
+
+		register_project_lifecycle_commands(subparsers, add_subcommand_format)
 	elif command == "todo":
 		from .cli_commands.todo_registration import register_todo_command
 
@@ -386,6 +393,25 @@ def _dispatch_common_command(
 			args, registry_path, effective_runtime_root(registry_path, args.runtime_root),
 			print_payload, output_format,
 		)
+	if args.command in PROJECT_LIFECYCLE_COMMANDS:
+		from .cli_commands.project_lifecycle import handle_project_lifecycle_command
+		from .cli_rollout import append_cli_rollout_event
+		from .capabilities.periodic_report.post_writeback_hook import (
+			build_periodic_report_post_writeback_projection,
+			periodic_report_post_writeback_hooks_for_goal,
+		)
+		from .paths import effective_runtime_root
+
+		return handle_project_lifecycle_command(
+			args, registry_path=registry_path, print_payload=print_payload,
+			output_format=output_format, append_cli_rollout_event=append_cli_rollout_event,
+			post_writeback_hooks=(periodic_report_post_writeback_hooks_for_goal(
+				registry_path=registry_path, goal_id=args.goal_id,
+				runtime_root=effective_runtime_root(registry_path, args.runtime_root),
+			) if args.command == "refresh-state" else ()),
+			post_writeback_projection_builder=(build_periodic_report_post_writeback_projection
+				if args.command == "refresh-state" else None),
+		)
 	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/control_plane/test_post_writeback_capability_hooks.py

@@ -2593,13 +2593,15 @@ def capture_hooks(
     assert captured["runtime_root"] == runtime_override
 
 
+@pytest.mark.parametrize("entrypoint", ["loopx.cli", "loopx.entrypoint"])
 def test_refresh_state_composition_passes_effective_runtime_root_to_hooks(
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,
+    entrypoint: str,
 ) -> None:
     """The refresh-state hook composition reads machine defaults under the override."""
 
-    from loopx import cli as cli_module
+    from importlib import import_module
 
     registry_path, _runtime_registry, runtime_override = _split_root_registry(tmp_path)
     captured: dict[str, object] = {}
@@ -2611,17 +2613,15 @@ def capture_hooks(
         return ()
 
     monkeypatch.setattr(
-        cli_module,
-        "periodic_report_post_writeback_hooks_for_goal",
+        "loopx.capabilities.periodic_report.post_writeback_hook.periodic_report_post_writeback_hooks_for_goal",
         capture_hooks,
     )
     monkeypatch.setattr(
-        cli_module,
-        "handle_project_lifecycle_command",
+        "loopx.cli_commands.project_lifecycle.handle_project_lifecycle_command",
         lambda *_args, **_kwargs: 7,
     )
 
-    exit_code = cli_module.main(
+    exit_code = import_module(entrypoint).main(
         [
             "--registry",
             str(registry_path),

tests/control_plane/test_refresh_external_delivery.py

@@ -9,13 +9,14 @@
 from loopx.control_plane.quota.settlement import SettlementIdentity
 from loopx.state_refresh import refresh_state_run
 from tests.control_plane.test_quota_settlement_cli import (
-    AGENT_ID, GOAL_ID, TODO_ID, TURN_ID, _write_fixture,
+    AGENT_ID, GOAL_ID, TODO_ID, TURN_ID, _write_fixture, _initialize_fixture_authority,
 )
 
 
 @pytest.fixture
 def session(tmp_path, monkeypatch, capsys):
     project, runtime, registry = _write_fixture(tmp_path)
+    _initialize_fixture_authority(project, runtime)
     monkeypatch.setenv("LOOPX_RUNTIME_ROOT", str(tmp_path / "shared"))
     monkeypatch.chdir(project)
     prefix = ["--registry", str(registry), "--runtime-root", str(runtime)]

tests/test_cli_entrypoint.py

@@ -289,6 +289,8 @@ def test_command_reference_does_not_require_default_runtime_route(
 		(["global-risks", "--help"], "loopx.cli_commands.summary_all_registration", "loopx.cli_commands.summary_all"),
 		(["turn", "--help"], "loopx.cli_commands.turn_registration", "loopx.cli_commands.turn"),
 		(["turn", "run-once", "--help"], "loopx.cli_commands.turn_registration", "loopx.cli_commands.turn"),
+		*[( [command, "--help"], "loopx.cli_commands.project_lifecycle", "loopx.cli_commands.project_lifecycle")
+		  for command in ("refresh-state", "checkpoint-context", "read-only-map", "reward", "operator-gate")],
 	],
 )
 def test_common_command_help_loads_only_its_owner(
@@ -416,6 +418,12 @@ def test_quota_facade_lazy_exports_preserve_owner_identity() -> None:
 
 def test_selected_parser_matches_full_help_and_diagnostics() -> None:
 	argv_cases = [
+		*[[command, flag] for command in
+		  ("refresh-state", "checkpoint-context", "read-only-map", "reward", "operator-gate")
+		  for flag in ("--help", "--unknown-option")],
+		["refresh-state", "--goal-id", "goal", "--delivery-outcome", "invalid"],
+		["checkpoint-context", "--goal-id", "goal", "--agent-id", "agent", "--turn-instance-id", "turn",
+		 "--todo-id", "todo_work", "--replan-obligation-id", "replan"],
 		["commands", "--help"],
 		["commands", "--format", "json"],
 		["commands", "--format", "markdown"],
@@ -544,6 +552,48 @@ 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", [
+	["refresh-state", "--dry-run"],
+	["checkpoint-context", "--agent-id", "perf-agent", "--turn-instance-id", "missing", "--todo-id", "todo_missing"],
+	["read-only-map"],
+	["reward", "--decision", "continue_route", "--reward", "positive", "--reason-summary", "Checked evidence", "--dry-run"],
+	["operator-gate", "--decision", "approve", "--reason-summary", "Checked evidence", "--dry-run"],
+])
+def test_selected_lifecycle_refusals_match_without_full_cli(tmp_path, arguments):
+	registry, runtime = write_command_fixture(tmp_path)
+	before = registry.read_bytes()
+	argv = ["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json",
+		*arguments, "--goal-id", "missing-goal"]
+	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 json.loads(selected.stdout)["ok"] is False
+	assert registry.read_bytes() == before
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_selected_refresh_preview_keeps_canonical_state_without_full_cli(tmp_path, monkeypatch, provider):
+	from tests.control_plane.canonical_authority_fixture import initialize_canonical_source_fixture, isolate_sqlite_runtime
+	from loopx.control_plane.coordination.local_authority import read_canonical_todos_if_promoted
+
+	isolate_sqlite_runtime(tmp_path, monkeypatch)
+	registry, runtime = write_command_fixture(tmp_path)
+	state = registry.parent.parent / ".codex/goals/perf-goal/ACTIVE_GOAL_STATE.md"
+	initialize_canonical_source_fixture(runtime, "perf-goal", state_path=state, provider=provider)
+	before = state.read_bytes(), read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="perf-goal")
+	argv = ["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json",
+		"refresh-state", "--goal-id", "perf-goal", "--agent-id", "perf-agent",
+		"--next-action", "Continue bounded fixture work.", "--no-global-sync", "--dry-run"]
+	for module in ("loopx.entrypoint", "loopx.cli"):
+		result = run_cli_main(module, argv, forbidden_modules=("loopx.cli",) if module == "loopx.entrypoint" else ())
+		assert result.returncode == 0, result.stderr
+		payload = json.loads(result.stdout)
+		assert payload["ok"] and payload["dry_run"]
+		assert (state.read_bytes(), read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="perf-goal")) == before
+
+
 def test_selected_manager_result_matches_full_replay_without_loading_full_cli(tmp_path):
 	from test_collaboration_goal_instance import _create_source_registry, _manager_request
 	from loopx.control_plane.collaboration.inbox import _root, acknowledge