diff --git a/loopx/cli_commands/support_control.py b/loopx/cli_commands/support_control.py index df5fba80f2..7227b23c6c 100644 --- a/loopx/cli_commands/support_control.py +++ b/loopx/cli_commands/support_control.py @@ -33,12 +33,6 @@ from ..kiro_cli_goal_mode import KIRO_CLI_BIN from ..paths import default_public_scan_root from ..presentation.renderers.status_markdown import render_status_markdown -from ..promotion_gate import ( - build_promotion_gate, - record_promotion_readiness, - render_promotion_gate_markdown, - render_promotion_readiness_record_markdown, -) from ..registry import ( inspect_registry, inspect_registry_boundary, @@ -60,7 +54,6 @@ DEFAULT_STATUS_PORT, serve_status, ) -from ..upgrade import build_upgrade_plan, render_upgrade_plan_markdown from .support_control_backup import ( handle_backup_state_command, register_backup_state_command, @@ -72,6 +65,10 @@ from .support_control_heartbeat_registration import ( register_heartbeat_control_commands, ) +from .support_control_promotion import ( + handle_promotion_control_command, + register_promotion_control_commands, +) from .support_control_registry import ( explicit_global_registry, resolve_heartbeat_active_state, @@ -119,68 +116,7 @@ def register_support_control_commands( register_supervisor_control_commands(subparsers, add_subcommand_format) - promotion_gate_parser = subparsers.add_parser( - "promotion-gate", - help="Emit a compact machine-readable canary promotion readiness gate result.", - ) - add_subcommand_format(promotion_gate_parser) - - promotion_readiness_parser = subparsers.add_parser( - "promotion-readiness", - help="Record release-scoped canary promotion-readiness evidence.", - ) - promotion_readiness_subparsers = promotion_readiness_parser.add_subparsers( - dest="promotion_readiness_command", - required=True, - ) - promotion_readiness_record_parser = promotion_readiness_subparsers.add_parser( - "record", - help="Append one runtime-level readiness event after the canary checks pass.", - ) - add_subcommand_format(promotion_readiness_record_parser) - promotion_readiness_record_parser.add_argument( - "--dashboard-readiness", - choices=("passed", "skipped"), - required=True, - help="Whether dashboard readiness ran successfully or was explicitly skipped.", - ) - promotion_readiness_record_parser.add_argument( - "--execute", - action="store_true", - help="Append the evidence event. Without this flag, emit a dry-run plan.", - ) - - upgrade_plan_parser = subparsers.add_parser( - "upgrade-plan", - help="Plan local default upgrade propagation for managed heartbeat automations.", - ) - add_subcommand_format(upgrade_plan_parser) - upgrade_plan_parser.add_argument( - "--goal-id", - action="append", - default=[], - help="Only include one goal id. Repeatable.", - ) - upgrade_plan_parser.add_argument( - "--installed-manifest", - help=( - "Optional JSON manifest of installed automations with goal_id, mode, automation_id, and " - "prompt_sha256/task_body. If omitted, upgrade-plan auto-discovers Codex App heartbeat " - "automations from $CODEX_HOME/automations or ~/.codex/automations." - ), - ) - upgrade_plan_parser.add_argument( - "--cli-bin", - default="loopx", - help="CLI command embedded in generated heartbeat prompts for the promoted default.", - ) - upgrade_plan_parser.add_argument( - "--mode", - action="append", - choices=["thin", "brief", "compact"], - default=[], - help="Prompt mode to compare. Repeatable; defaults to the thin installed heartbeat contract.", - ) + register_promotion_control_commands(subparsers, add_subcommand_format) update_parser = subparsers.add_parser( "update", @@ -569,90 +505,14 @@ def handle_support_control_command( if supervisor_result is not None: return supervisor_result - if args.command == "promotion-gate": - try: - payload = build_promotion_gate( - registry_path=registry_path, - runtime_root_override=args.runtime_root, - ) - except Exception as exc: - payload = { - "ok": False, - "registry": str(registry_path), - "runtime_root": args.runtime_root, - "gate": "promotion_readiness", - "gate_state": "error", - "can_promote": False, - "should_warn": True, - "non_blocking": True, - "error": str(exc), - "recommended_action": "fix promotion readiness gate collection before promotion", - } - print_payload(payload, output_format(args), render_promotion_gate_markdown) - return 0 if payload.get("ok") else 1 - - if args.command == "promotion-readiness": - try: - payload = record_promotion_readiness( - registry_path=registry_path, - runtime_root_override=args.runtime_root, - dashboard_readiness=args.dashboard_readiness, - execute=args.execute, - ) - except Exception as exc: - payload = { - "ok": False, - "dry_run": not args.execute, - "appended": False, - "registry": str(registry_path), - "runtime_root": args.runtime_root, - "evidence_scope": "runtime_release", - "error": str(exc), - } - print_payload( - payload, - output_format(args), - render_promotion_readiness_record_markdown, - ) - return 0 if payload.get("ok") else 1 - - if args.command == "upgrade-plan": - try: - payload = build_upgrade_plan( - registry_path=registry_path, - runtime_root_override=args.runtime_root, - installed_manifest=Path(args.installed_manifest).expanduser() - if args.installed_manifest - else None, - cli_bin=args.cli_bin, - modes=args.mode or None, - goal_ids=args.goal_id or None, - ) - except Exception as exc: - payload = { - "ok": False, - "mode": "upgrade-plan", - "registry": str(registry_path), - "runtime_root": args.runtime_root, - "error": str(exc), - "summary": { - "managed_goal_count": 0, - "current_prompt_count": 0, - "stale_prompt_count": 0, - "unknown_prompt_count": 0, - "not_installed_prompt_count": 0, - "stage_deferred_goal_count": 0, - "ready_for_default_promotion": False, - "installed_manifest_available": False, - "installed_manifest_source": None, - "installed_manifest_entry_count": 0, - "installed_manifest_task_body_count": 0, - "installed_manifest_has_task_body": False, - }, - "recommended_action": "fix upgrade-plan collection before default promotion", - } - print_payload(payload, output_format(args), render_upgrade_plan_markdown) - return 0 if payload.get("ok") else 1 + promotion_result = handle_promotion_control_command( + args, + registry_path=registry_path, + print_payload=print_payload, + output_format=output_format, + ) + if promotion_result is not None: + return promotion_result if args.command == "update": update_action = UpdateAction.PLAN diff --git a/loopx/cli_commands/support_control_promotion.py b/loopx/cli_commands/support_control_promotion.py new file mode 100644 index 0000000000..ce8f522f32 --- /dev/null +++ b/loopx/cli_commands/support_control_promotion.py @@ -0,0 +1,201 @@ +"""Registration and dispatch for the promotion-gate, promotion-readiness, and +upgrade-plan commands. + +Refs GH-C06. This group was carved out of `support_control.py`, which registers +seven unrelated top-level commands in one module that sits just under the +1000-line default budget in +`examples/cli-command-module-size-ownership-command-modularization-smoke.py`. +The three commands are one group -- canary promotion readiness and the local +default upgrade plan that follows it -- so their parser flags and their +dispatch branches move together and the public invocation is unchanged. +""" + +from __future__ import annotations + +import argparse +from collections.abc import Callable +from pathlib import Path + +from ..promotion_gate import ( + build_promotion_gate, + record_promotion_readiness, + render_promotion_gate_markdown, + render_promotion_readiness_record_markdown, +) +from ..upgrade import build_upgrade_plan, render_upgrade_plan_markdown + +PrintPayload = Callable[ + [dict[str, object], str, Callable[[dict[str, object]], str]], + None, +] + +PROMOTION_CONTROL_COMMANDS = { + "promotion-gate", + "promotion-readiness", + "upgrade-plan", +} + + +def register_promotion_control_commands( + subparsers: argparse._SubParsersAction, + add_subcommand_format: Callable[[argparse.ArgumentParser], None], +) -> None: + promotion_gate_parser = subparsers.add_parser( + "promotion-gate", + help="Emit a compact machine-readable canary promotion readiness gate result.", + ) + add_subcommand_format(promotion_gate_parser) + + promotion_readiness_parser = subparsers.add_parser( + "promotion-readiness", + help="Record release-scoped canary promotion-readiness evidence.", + ) + promotion_readiness_subparsers = promotion_readiness_parser.add_subparsers( + dest="promotion_readiness_command", + required=True, + ) + promotion_readiness_record_parser = promotion_readiness_subparsers.add_parser( + "record", + help="Append one runtime-level readiness event after the canary checks pass.", + ) + add_subcommand_format(promotion_readiness_record_parser) + promotion_readiness_record_parser.add_argument( + "--dashboard-readiness", + choices=("passed", "skipped"), + required=True, + help="Whether dashboard readiness ran successfully or was explicitly skipped.", + ) + promotion_readiness_record_parser.add_argument( + "--execute", + action="store_true", + help="Append the evidence event. Without this flag, emit a dry-run plan.", + ) + + upgrade_plan_parser = subparsers.add_parser( + "upgrade-plan", + help="Plan local default upgrade propagation for managed heartbeat automations.", + ) + add_subcommand_format(upgrade_plan_parser) + upgrade_plan_parser.add_argument( + "--goal-id", + action="append", + default=[], + help="Only include one goal id. Repeatable.", + ) + upgrade_plan_parser.add_argument( + "--installed-manifest", + help=( + "Optional JSON manifest of installed automations with goal_id, mode, automation_id, and " + "prompt_sha256/task_body. If omitted, upgrade-plan auto-discovers Codex App heartbeat " + "automations from $CODEX_HOME/automations or ~/.codex/automations." + ), + ) + upgrade_plan_parser.add_argument( + "--cli-bin", + default="loopx", + help="CLI command embedded in generated heartbeat prompts for the promoted default.", + ) + upgrade_plan_parser.add_argument( + "--mode", + action="append", + choices=["thin", "brief", "compact"], + default=[], + help="Prompt mode to compare. Repeatable; defaults to the thin installed heartbeat contract.", + ) + +def handle_promotion_control_command( + args: argparse.Namespace, + *, + registry_path: Path, + print_payload: PrintPayload, + output_format: Callable[[argparse.Namespace], str], +) -> int | None: + if args.command not in PROMOTION_CONTROL_COMMANDS: + return None + + if args.command == "promotion-gate": + try: + payload = build_promotion_gate( + registry_path=registry_path, + runtime_root_override=args.runtime_root, + ) + except Exception as exc: + payload = { + "ok": False, + "registry": str(registry_path), + "runtime_root": args.runtime_root, + "gate": "promotion_readiness", + "gate_state": "error", + "can_promote": False, + "should_warn": True, + "non_blocking": True, + "error": str(exc), + "recommended_action": "fix promotion readiness gate collection before promotion", + } + print_payload(payload, output_format(args), render_promotion_gate_markdown) + return 0 if payload.get("ok") else 1 + + if args.command == "promotion-readiness": + try: + payload = record_promotion_readiness( + registry_path=registry_path, + runtime_root_override=args.runtime_root, + dashboard_readiness=args.dashboard_readiness, + execute=args.execute, + ) + except Exception as exc: + payload = { + "ok": False, + "dry_run": not args.execute, + "appended": False, + "registry": str(registry_path), + "runtime_root": args.runtime_root, + "evidence_scope": "runtime_release", + "error": str(exc), + } + print_payload( + payload, + output_format(args), + render_promotion_readiness_record_markdown, + ) + return 0 if payload.get("ok") else 1 + + if args.command == "upgrade-plan": + try: + payload = build_upgrade_plan( + registry_path=registry_path, + runtime_root_override=args.runtime_root, + installed_manifest=Path(args.installed_manifest).expanduser() + if args.installed_manifest + else None, + cli_bin=args.cli_bin, + modes=args.mode or None, + goal_ids=args.goal_id or None, + ) + except Exception as exc: + payload = { + "ok": False, + "mode": "upgrade-plan", + "registry": str(registry_path), + "runtime_root": args.runtime_root, + "error": str(exc), + "summary": { + "managed_goal_count": 0, + "current_prompt_count": 0, + "stale_prompt_count": 0, + "unknown_prompt_count": 0, + "not_installed_prompt_count": 0, + "stage_deferred_goal_count": 0, + "ready_for_default_promotion": False, + "installed_manifest_available": False, + "installed_manifest_source": None, + "installed_manifest_entry_count": 0, + "installed_manifest_task_body_count": 0, + "installed_manifest_has_task_body": False, + }, + "recommended_action": "fix upgrade-plan collection before default promotion", + } + print_payload(payload, output_format(args), render_upgrade_plan_markdown) + return 0 if payload.get("ok") else 1 + + return None diff --git a/tests/test_support_control_promotion_ownership.py b/tests/test_support_control_promotion_ownership.py new file mode 100644 index 0000000000..a5f3523a68 --- /dev/null +++ b/tests/test_support_control_promotion_ownership.py @@ -0,0 +1,70 @@ +"""Refs GH-C06: the promotion command group still belongs to support control. + +`promotion-gate`, `promotion-readiness`, and `upgrade-plan` were extracted from +`cli_commands/support_control.py` into +`cli_commands/support_control_promotion.py` so the shared support-control seam +stops owning three unrelated command groups at once. The extraction must not +change the public invocation, so these cases pin the two things that could +silently break it: the commands must still be part of the support control set, +and each must still be registered exactly once. +""" + +from __future__ import annotations + +import argparse +import re +from pathlib import Path + +from loopx.cli_commands import support_control +from loopx.cli_commands import support_control_promotion as promotion_module + +COMMANDS_DIR = Path(promotion_module.__file__).resolve().parent +ADD_PARSER_RE = re.compile( + r"subparsers\.add_parser\(\s*(?:\n\s*)?[\"'](?P[^\"']+)[\"']", + re.MULTILINE, +) + +PROMOTION_COMMANDS = ("promotion-gate", "promotion-readiness", "upgrade-plan") + + +def registered_commands() -> dict[str, list[str]]: + registrations: dict[str, list[str]] = {} + for path in sorted(COMMANDS_DIR.glob("*.py")): + for match in ADD_PARSER_RE.finditer(path.read_text(encoding="utf-8")): + registrations.setdefault(match.group("command"), []).append(path.name) + return registrations + + +def test_promotion_commands_are_still_support_control_commands() -> None: + for command in PROMOTION_COMMANDS: + assert command in support_control.SUPPORT_CONTROL_COMMANDS + + +def test_promotion_commands_are_registered_exactly_once() -> None: + registrations = registered_commands() + for command in PROMOTION_COMMANDS: + assert registrations[command] == ["support_control_promotion.py"] + + +def test_owner_module_exposes_both_halves() -> None: + """Registration and dispatch moved together, so both live in the new module.""" + assert callable(promotion_module.register_promotion_control_commands) + assert callable(promotion_module.handle_promotion_control_command) + + +def test_owner_module_owns_exactly_its_group() -> None: + assert promotion_module.PROMOTION_CONTROL_COMMANDS == set(PROMOTION_COMMANDS) + + +def test_dispatch_ignores_other_commands() -> None: + """A non-promotion command must fall through untouched, before any work.""" + args = argparse.Namespace(command="update") + assert ( + promotion_module.handle_promotion_control_command( + args, + registry_path=Path("/nonexistent-registry"), + print_payload=lambda *_args: None, + output_format=lambda *_args: "json", + ) + is None + )