"""CLI entry point ``rmgr``.""" from __future__ import annotations import argparse import json from pathlib import Path from repo_manager.commands.rapp import add_rapp_parser from repo_manager.commands.rapp import init as rapp_init from repo_manager.commands.rapp import pin_image as rapp_pin_image from repo_manager.commands.rapp import place as rapp_place from repo_manager.commands.rapp import skeleton as rapp_skeleton from repo_manager.commands.rapp import validate as rapp_validate from repo_manager.commands.rapp import wrap as rapp_wrap def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( prog="rmgr", description="Repo Manager CLI (helixforge.repo-manager)", ) parser.add_argument("--version", action="store_true", help="Print package version") sub = parser.add_subparsers(dest="command") sub.add_parser("version", help="Print version") sub.add_parser("dual-run-status", help="Show dual-run flags and mutation meter") p_obs = sub.add_parser("observe", help="Observe repository + print snapshot JSON") p_obs.add_argument("--path", default=".", help="Repository checkout path") p_obs.add_argument("--slug", default=None, help="Override repo slug") p_rec = sub.add_parser("reconcile", help="Rebuild local work-record index from files") p_rec.add_argument("--path", default=".", help="Repository checkout path") p_rec.add_argument("--slug", default=None) p_rec.add_argument("--no-write-index", action="store_true", help="Do not write index file") p_cmd = sub.add_parser( "update-task-status", help="Command repo.work.update_task_status (file + git commit)", ) p_cmd.add_argument("--path", default=".", help="Repository checkout path") p_cmd.add_argument( "--task-id", required=True, help="Canonical task id or State Hub task UUID", ) p_cmd.add_argument( "--status", required=True, choices=["wait", "todo", "progress", "done", "cancel"], ) p_cmd.add_argument("--reason", default="rmgr CLI") p_cmd.add_argument("--correlation-id", default=None) p_cmd.add_argument("--idempotency-key", default=None) p_cmd.add_argument("--expected-head-sha", default=None) p_cmd.add_argument("--slug", default=None) p_cmd.add_argument("--push", action="store_true", help="git push after commit (push-seal)") p_cmd.add_argument( "--no-commit", action="store_true", help="Patch file only (invalid as full applied evidence; for tests)", ) add_rapp_parser(sub) args = parser.parse_args(argv) if args.version or args.command in (None, "version"): from repo_manager import __version__ print(__version__) return 0 if args.command == "dual-run-status": from repo_manager.dual_run import flags_status print(json.dumps(flags_status(), indent=2)) return 0 if args.command == "observe": from repo_manager.observe import observe_repository snap, _idx = observe_repository(Path(args.path), slug=args.slug) print(json.dumps(snap, indent=2)) return 0 if args.command == "reconcile": from repo_manager.dual_run import record_mutation from repo_manager.index_store import append_event, save_index from repo_manager.observe import observe_repository root = Path(args.path) snap, index = observe_repository(root, slug=args.slug) append_event( index, { "type": "repo.reconciled", "workplan_count": snap["index"]["workplan_count"], "task_count": snap["index"]["task_count"], "source": "repo-manager", }, ) record_mutation( source="repo-manager", kind="reconcile", repo_slug=snap.get("slug"), detail=snap["index"], ) if not args.no_write_index: path = save_index(index) print(json.dumps({"ok": True, "index_path": str(path), "snapshot": snap}, indent=2)) else: print(json.dumps({"ok": True, "snapshot": snap, "index": index.to_dict()}, indent=2)) return 0 if args.command == "update-task-status": from repo_manager.commands.task_status import update_task_status result = update_task_status( Path(args.path), args.task_id, args.status, correlation_id=args.correlation_id, reason=args.reason, commit=not args.no_commit, push=args.push, expected_head_sha=args.expected_head_sha, idempotency_key=args.idempotency_key, repo_slug=args.slug, ) print(json.dumps(result.to_dict(), indent=2)) return 0 if result.status == "applied" else 1 if args.command == "rapp": if args.rapp_command == "init": result = rapp_init( Path(args.path), app=args.app, ownership_repo=args.ownership_repo, rail=args.rail, classification=args.classification, criticality=args.criticality, package_type=args.package_type, purpose=args.purpose, force=args.force, ) elif args.rapp_command == "validate": result = rapp_validate( Path(args.path), family_root=Path(args.family_root) if args.family_root else None, ) elif args.rapp_command == "skeleton": result = rapp_skeleton( Path(args.path), from_app=Path(args.from_app), app=args.app, package_type=args.package_type, force=args.force, dedicated_postgres=args.dedicated_postgres, ) elif args.rapp_command == "wrap": result = rapp_wrap( Path(args.path), app=args.app, ownership_repo=args.ownership_repo, from_app=Path(args.from_app), rail=args.rail, classification=args.classification, criticality=args.criticality, package_type=args.package_type, purpose=args.purpose, force=args.force, dedicated_postgres=args.dedicated_postgres, family_root=Path(args.family_root) if args.family_root else None, ) elif args.rapp_command == "place": result = rapp_place( Path(args.path), reef=args.reef, family_root=Path(args.family_root) if args.family_root else None, ) elif args.rapp_command == "pin-image": result = rapp_pin_image(Path(args.path), args.digest) else: parser.print_help() return 2 print(json.dumps(result, indent=2)) return 0 if result.get("ok") else 1 parser.print_help() return 0 if __name__ == "__main__": raise SystemExit(main())