"""CLI entry point ``rmgr``.""" from __future__ import annotations import argparse import json import os 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 _json_object(raw: str) -> dict: try: value = json.loads(raw) except json.JSONDecodeError as exc: raise argparse.ArgumentTypeError(str(exc)) from exc if not isinstance(value, dict): raise argparse.ArgumentTypeError("value must be a JSON object") return value 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_registrar = sub.add_parser( "registrar-reconcile", help="Assign missing State Hub UUIDs through a scoped on-demand registrar", ) p_registrar.add_argument("--path", default=".", help="Repository checkout path") p_registrar.add_argument( "--api-base", default=os.environ.get("STATE_HUB_API_BASE", "http://127.0.0.1:8000"), help="Authoritative State Hub API base", ) p_registrar.add_argument("--statehub-bin", default=None, help="Override statehub executable") p_registrar.add_argument( "--confirm-primary", action="store_true", help="Confirm that --api-base is the authoritative hub", ) p_registrar.add_argument("--push", action="store_true", help="Push the registrar commit") registrar_mode = p_registrar.add_mutually_exclusive_group() registrar_mode.add_argument( "--repair-workplan", default=None, metavar="ID", help="Rebuild and verify one already-identified workplan projection", ) registrar_mode.add_argument( "--bootstrap-empty-projection", action="store_true", help="Rebuild all authoritative UUIDs after proving the repo projection is empty", ) 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)", ) p_wp = sub.add_parser("workplan", help="Governed file-backed workplan mutations") wp_sub = p_wp.add_subparsers(dest="workplan_command") p_wp_create = wp_sub.add_parser("create", help="Create a workplan file") p_wp_create.add_argument("--path", default=".") p_wp_create.add_argument("--workplan-id", required=True) p_wp_create.add_argument("--title", required=True) p_wp_create.add_argument("--goal", required=True) p_wp_create.add_argument( "--status", choices=["proposed", "ready", "active", "blocked", "backlog", "finished", "archived"], default="proposed", ) p_wp_create.add_argument("--owner", default="codex") p_wp_create.add_argument("--domain", default=None) p_wp_create.add_argument("--topic-slug", default=None) p_wp_create.add_argument("--filename", default=None) p_wp_update = wp_sub.add_parser("update", help="Update workplan metadata or status") p_wp_update.add_argument("--path", default=".") p_wp_update.add_argument("--workplan-id", required=True) p_wp_update.add_argument("--title", default=None) p_wp_update.add_argument( "--status", choices=["proposed", "ready", "active", "blocked", "backlog", "finished", "archived"], default=None, ) p_wp_update.add_argument("--owner", default=None) p_wp_update.add_argument("--domain", default=None) p_wp_update.add_argument("--topic-slug", default=None) p_wp_archive = wp_sub.add_parser( "delete", help="Recoverable delete: set archived and move to workplans/archived", ) p_wp_archive.add_argument("--path", default=".") p_wp_archive.add_argument("--workplan-id", required=True) p_wp_archive.add_argument( "--confirm", action="store_true", help="Confirm moving the workplan to the dated archive", ) for wp_parser in (p_wp_create, p_wp_update, p_wp_archive): wp_parser.add_argument("--reason", default="rmgr CLI") wp_parser.add_argument("--correlation-id", default=None) wp_parser.add_argument("--idempotency-key", default=None) wp_parser.add_argument("--expected-head-sha", default=None) wp_parser.add_argument("--slug", default=None) wp_parser.add_argument("--push", action="store_true") wp_parser.add_argument("--no-commit", action="store_true") register_kinds = [ "sbom-inventory", "repo-goals", "upstream-contributions", "technical-debt", "extension-points", "register-entries", ] p_register = sub.add_parser("register", help="Repository-owned register spine") register_sub = p_register.add_subparsers(dest="register_command") p_reg_list = register_sub.add_parser("list", help="List indexed register entries") p_reg_list.add_argument("--path", default=".") p_reg_list.add_argument("--kind", choices=register_kinds, default=None) p_reg_list.add_argument("--slug", default=None) p_reg_put = register_sub.add_parser("put", help="Create or update a register entry") p_reg_put.add_argument("--path", default=".") p_reg_put.add_argument("--kind", required=True, choices=register_kinds) p_reg_put.add_argument("--entry-id", required=True) p_reg_put.add_argument("--title", default=None) p_reg_put.add_argument("--status", default=None) p_reg_put.add_argument("--data-json", type=_json_object, default=None) p_reg_defer = register_sub.add_parser("defer", help="Defer a register entry") p_reg_defer.add_argument("--path", default=".") p_reg_defer.add_argument("--kind", required=True, choices=register_kinds) p_reg_defer.add_argument("--entry-id", required=True) p_reg_defer.add_argument("--status", default="deferred") p_reg_note = register_sub.add_parser("note", help="Append a note to a register entry") p_reg_note.add_argument("--path", default=".") p_reg_note.add_argument("--kind", required=True, choices=register_kinds) p_reg_note.add_argument("--entry-id", required=True) p_reg_note.add_argument("--note", required=True) p_reg_note.add_argument("--author", default="codex") for register_parser in (p_reg_put, p_reg_defer, p_reg_note): register_parser.add_argument("--reason", default="rmgr CLI") register_parser.add_argument("--correlation-id", default=None) register_parser.add_argument("--idempotency-key", default=None) register_parser.add_argument("--expected-head-sha", default=None) register_parser.add_argument("--slug", default=None) register_parser.add_argument("--push", action="store_true") register_parser.add_argument("--no-commit", action="store_true") p_intake = sub.add_parser("intake", help="Repository-owned intake records") intake_sub = p_intake.add_subparsers(dest="record_command") p_intake_create = intake_sub.add_parser("create", help="Create an intake") p_intake_create.add_argument("--path", default=".") p_intake_create.add_argument("--record-id", required=True) p_intake_create.add_argument("--title", required=True) p_intake_create.add_argument("--status", default="open") p_intake_create.add_argument("--data-json", type=_json_object, default=None) p_intake_route = intake_sub.add_parser("route", help="Route an intake") p_intake_route.add_argument("--path", default=".") p_intake_route.add_argument("--record-id", required=True) p_intake_route.add_argument("--route-to", required=True) p_intake_note = intake_sub.add_parser("note", help="Append an intake note") p_intake_note.add_argument("--path", default=".") p_intake_note.add_argument("--record-id", required=True) p_intake_note.add_argument("--note", required=True) p_intake_note.add_argument("--author", default="codex") p_intake_close = intake_sub.add_parser("close", help="Close an intake") p_intake_close.add_argument("--path", default=".") p_intake_close.add_argument("--record-id", required=True) p_intake_close.add_argument("--outcome", default=None) p_decision = sub.add_parser("decision", help="Repository-owned decision records") decision_sub = p_decision.add_subparsers(dest="record_command") p_decision_create = decision_sub.add_parser("create", help="Create a decision") p_decision_create.add_argument("--path", default=".") p_decision_create.add_argument("--record-id", required=True) p_decision_create.add_argument("--title", required=True) p_decision_create.add_argument("--status", default="open") p_decision_create.add_argument("--data-json", type=_json_object, default=None) p_decision_update = decision_sub.add_parser("update", help="Update a decision") p_decision_update.add_argument("--path", default=".") p_decision_update.add_argument("--record-id", required=True) p_decision_update.add_argument("--title", default=None) p_decision_update.add_argument("--status", default=None) p_decision_update.add_argument("--data-json", type=_json_object, default=None) p_decision_resolve = decision_sub.add_parser("resolve", help="Resolve a decision") p_decision_resolve.add_argument("--path", default=".") p_decision_resolve.add_argument("--record-id", required=True) p_decision_resolve.add_argument("--rationale", required=True) p_decision_resolve.add_argument("--decided-by", required=True) for record_parser in ( p_intake_create, p_intake_route, p_intake_note, p_intake_close, p_decision_create, p_decision_update, p_decision_resolve, ): record_parser.add_argument("--reason", default="rmgr CLI") record_parser.add_argument("--correlation-id", default=None) record_parser.add_argument("--idempotency-key", default=None) record_parser.add_argument("--expected-head-sha", default=None) record_parser.add_argument("--slug", default=None) record_parser.add_argument("--push", action="store_true") record_parser.add_argument("--no-commit", action="store_true") add_rapp_parser(sub) p_conf = sub.add_parser("conform", help="Check a repository against flavor standards") p_conf.add_argument("--path", default=".", help="Repository checkout path") p_conf.add_argument("--slug", default=None) p_pref = sub.add_parser( "prefix-uniqueness", help="Detect shared workplan prefixes and reused identifiers (ADR-007)", ) p_pref.add_argument("--root", default=".", help="Fleet root or a single repository") p_pref.add_argument("--registry", default=None, help="Override prefix registry YAML") p_scaf = sub.add_parser("scaffold", help="Create flavor-correct repository baseline files") p_scaf.add_argument("--path", required=True) p_scaf.add_argument("--flavor", required=True, choices=["experimental", "research", "project", "tooling", "product", "business"]) p_scaf.add_argument("--slug", default=None) p_scaf.add_argument("--domain", default="infotech") p_scaf.add_argument("--wp-prefix", default=None) p_scaf.add_argument("--force", action="store_true") p_scaf.add_argument("--no-commit", action="store_true") p_provenance = sub.add_parser( "assistant-provenance", help="Install or report coding-assistant commit provenance", ) provenance_sub = p_provenance.add_subparsers(dest="provenance_command") p_prov_install = provenance_sub.add_parser("install", help="Set global core.hooksPath") p_prov_install.add_argument( "--hooks-path", default=str(Path(__file__).resolve().parents[2] / ".githooks"), ) p_prov_report = provenance_sub.add_parser("report", help="Report trailers from Git history") p_prov_report.add_argument("--path", default=".") p_prov_report.add_argument("--rev", default="HEAD") p_prov_report.add_argument("--max-count", type=int, default=None) p_identifier = sub.add_parser("identifier", help="Deterministic work-record identifiers") identifier_sub = p_identifier.add_subparsers(dest="identifier_command") p_id_derive = identifier_sub.add_parser("derive", help="Derive one UUIDv5") p_id_derive.add_argument("--namespace", default=None, help="Override declared fleet namespace") p_id_derive.add_argument("--record-id", required=True) p_id_preflight = identifier_sub.add_parser("preflight", help="Scan live identifier collisions") p_id_preflight.add_argument("--root", default=".") p_id_plan = identifier_sub.add_parser( "migration-plan", help="Emit a non-mutating, per-repository old-to-derived UUID plan", ) p_id_plan.add_argument("--root", default=".") p_id_plan.add_argument("--namespace", default=None, help="Override declared fleet namespace") p_id_plan.add_argument("--output", default=None, help="Write the provenance mapping as JSON") p_id_plan.add_argument("--force", action="store_true", help="Replace an existing --output file") p_id_verify = identifier_sub.add_parser( "migration-verify", help="Verify a sealed migration plan against current repository sources", ) p_id_verify.add_argument("--plan", required=True) p_id_verify.add_argument("--repo", default=None, help="Verify one repository atomic unit") p_id_batch = identifier_sub.add_parser( "migration-batch-plan", help="Pin a clean synchronized repository batch for explicit approval", ) p_id_batch.add_argument("--plan", required=True) p_id_batch.add_argument("--repo", action="append", required=True, dest="repos") p_id_batch.add_argument("--output", default=None) p_id_batch.add_argument("--force", action="store_true") p_id_batch_verify = identifier_sub.add_parser( "migration-batch-verify", help="Verify a saved batch seal and repeat source/Git preflight", ) p_id_batch_verify.add_argument("--plan", required=True) p_id_batch_verify.add_argument("--batch", required=True) p_id_files = identifier_sub.add_parser( "migration-files", help="Validate or execute one repository's sealed UUID file rewrite", ) p_id_files.add_argument("--plan", required=True) p_id_files.add_argument("--repo", required=True) p_id_files.add_argument("--confirm-plan-sha256", required=True) p_id_files.add_argument( "--direction", choices=["forward", "reverse"], default="forward" ) p_id_files.add_argument( "--execute", action="store_true", help="Write files; without this flag only validate and report", ) p_sbom = sub.add_parser("sbom", help="Derive SBOM snapshots and licence reports from repository files") sbom_sub = p_sbom.add_subparsers(dest="sbom_command") p_sbom_scan = sbom_sub.add_parser("scan", help="Scan recognised lockfiles and tool manifests") p_sbom_scan.add_argument("--path", default=".") p_sbom_scan.add_argument("--slug", default=None) p_sbom_scan.add_argument("--output", default=None, help="Write the derived snapshot as JSON") p_sbom_scan.add_argument("--force", action="store_true", help="Replace an existing --output file") p_sbom_report = sbom_sub.add_parser("licence-report", help="Report licences from a fresh file scan") p_sbom_report.add_argument("--path", default=".") p_sbom_report.add_argument("--slug", default=None) p_authority = sub.add_parser("authority", help="Resolve the one authoritative record owner") authority_sub = p_authority.add_subparsers(dest="authority_command") p_authority_route = authority_sub.add_parser("route", help="Resolve or verify an authority route") p_authority_route.add_argument("--record-type", required=True) p_authority_route.add_argument("--repo-slug", default=None) p_authority_route.add_argument("--domain-slug", default=None) p_authority_route.add_argument("--claimed-owner", default=None) p_cache = sub.add_parser("cache", help="Inspect and rebuild advisory repository caches") cache_sub = p_cache.add_subparsers(dest="cache_command") p_cache_status = cache_sub.add_parser("status", help="Report cache age and source drift") p_cache_status.add_argument("--path", default=".") p_cache_rebuild = cache_sub.add_parser("rebuild", help="Rebuild the local index from files") p_cache_rebuild.add_argument("--path", default=".") p_cache_rebuild.add_argument("--slug", default=None) p_cache_export = cache_sub.add_parser( "export-closed", help="Export unbound closed State Hub records before cache replacement", ) p_cache_export.add_argument( "--api-base", default=os.environ.get("STATE_HUB_API_BASE", "http://127.0.0.1:8000"), ) p_cache_export.add_argument("--output", required=True) p_cache_export.add_argument("--force", action="store_true") p_workload = sub.add_parser( "workload", help="Index or resolve authoritative rapp workload declarations", ) workload_sub = p_workload.add_subparsers(dest="workload_command") p_workload_index = workload_sub.add_parser( "index", help="Index workload identities from rapp declarations" ) p_workload_index.add_argument("--root", default=".", help="Fleet root or one rapp repo") p_workload_resolve = workload_sub.add_parser( "resolve", help="Resolve one explicit workload reference without inference" ) p_workload_resolve.add_argument("--root", default=".", help="Fleet root or one rapp repo") p_workload_resolve.add_argument("--rapp-id", required=True) p_workload_resolve.add_argument("--name", required=True) p_workload_resolve.add_argument("--deployable", default=None) p_owner_interface = sub.add_parser( "owner-interface", help="Inspect validated owner-consumable task interfaces", ) owner_interface_sub = p_owner_interface.add_subparsers(dest="owner_interface_command") p_owner_interface_validate = owner_interface_sub.add_parser( "validate", help="Validate and return owner task interfaces" ) p_owner_interface_validate.add_argument( "--path", default="interfaces", help="One interface YAML or an interface directory" ) p_owner_interface_validate.add_argument( "--owner", default=None, help="Return interfaces for this target repo or owner agent" ) 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 == "registrar-reconcile": from repo_manager.commands.registrar_reconcile import registrar_reconcile result = registrar_reconcile( Path(args.path), api_base=args.api_base, statehub_bin=args.statehub_bin, confirm_primary=args.confirm_primary, push=args.push, repair_workplan=args.repair_workplan, bootstrap_empty_projection=args.bootstrap_empty_projection, ) print(json.dumps(result.to_dict(), indent=2)) return 0 if result.status in {"applied", "noop"} else 1 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 == "workplan": from repo_manager.commands.workplan import mutate_workplan if not args.workplan_command: p_wp.print_help() return 2 operation = "archive" if args.workplan_command == "delete" else args.workplan_command result = mutate_workplan( Path(args.path), args.workplan_id, operation=operation, title=getattr(args, "title", None), goal=getattr(args, "goal", None), status=getattr(args, "status", None), owner=getattr(args, "owner", None), topic_slug=getattr(args, "topic_slug", None), domain=getattr(args, "domain", None), filename=getattr(args, "filename", None), confirm_archive=getattr(args, "confirm", False), 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 == "register": if not args.register_command: p_register.print_help() return 2 if args.register_command == "list": from repo_manager.observe import observe_repository _snapshot, index = observe_repository(Path(args.path), slug=args.slug) prefix = f"register:{args.kind}" if args.kind else "register:" entries = [r.__dict__ for r in index.work_records if r.kind.startswith(prefix)] print(json.dumps({"ok": True, "entries": entries}, indent=2)) return 0 from repo_manager.commands.register import mutate_register_entry result = mutate_register_entry( Path(args.path), args.kind, args.entry_id, operation="upsert" if args.register_command == "put" else args.register_command, title=getattr(args, "title", None), status=getattr(args, "status", None), data=getattr(args, "data_json", None), note=getattr(args, "note", None), note_author=getattr(args, "author", None), 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 in {"intake", "decision"}: selected = p_intake if args.command == "intake" else p_decision if not args.record_command: selected.print_help() return 2 from repo_manager.commands.record import mutate_record result = mutate_record( Path(args.path), args.command, args.record_id, operation=args.record_command, title=getattr(args, "title", None), status=getattr(args, "status", None), data=getattr(args, "data_json", None), route_to=getattr(args, "route_to", None), note=getattr(args, "note", None), author=getattr(args, "author", None), outcome=getattr(args, "outcome", None), rationale=getattr(args, "rationale", None), decided_by=getattr(args, "decided_by", None), 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 if args.command == "conform": from repo_manager.standards import check_repository report = check_repository(Path(args.path), slug=args.slug) print(json.dumps(report.to_dict(), indent=2)) return 0 if report.ok else 1 if args.command == "prefix-uniqueness": from repo_manager.prefix_registry import scan_prefixes registry = Path(args.registry) if args.registry else None report = scan_prefixes(Path(args.root), registry_path=registry) print(json.dumps(report, indent=2)) return 0 if report.get("ok") else 1 if args.command == "scaffold": from repo_manager.commands.scaffold import scaffold_repository result = scaffold_repository( Path(args.path), flavor=args.flavor, slug=args.slug, domain=args.domain, wp_prefix=args.wp_prefix, force=args.force, commit=not args.no_commit, ) print(json.dumps(result.to_dict(), indent=2)) return 0 if result.status == "applied" else 1 if args.command == "assistant-provenance": if not args.provenance_command: p_provenance.print_help() return 2 from repo_manager.provenance import assistant_report, install_hook if args.provenance_command == "install": result = install_hook(Path(args.hooks_path)) else: result = assistant_report(Path(args.path), rev=args.rev, max_count=args.max_count) print(json.dumps(result, indent=2)) return 0 if result.get("ok") else 1 if args.command == "identifier": if not args.identifier_command: p_identifier.print_help() return 2 from repo_manager.identifiers import ( derive_work_record_uuid, load_fleet_namespace, migrate_repository_identifier_files, plan_identifier_migration, plan_identifier_migration_batch, scan_live_identifier_collisions, verify_identifier_migration_batch, verify_identifier_migration_plan, ) namespace = getattr(args, "namespace", None) or load_fleet_namespace() if args.identifier_command == "derive": try: derived = derive_work_record_uuid(namespace, args.record_id) except ValueError as exc: print(json.dumps({"ok": False, "error": str(exc)}, indent=2)) return 1 result = { "ok": True, "namespace": namespace, "record_id": args.record_id, "uuid": str(derived), } elif args.identifier_command == "preflight": result = scan_live_identifier_collisions(Path(args.root)) elif args.identifier_command == "migration-verify": try: plan = json.loads(Path(args.plan).read_text(encoding="utf-8")) if not isinstance(plan, dict): raise TypeError("migration plan must be a JSON object") result = verify_identifier_migration_plan(plan, repo_slug=args.repo) except (OSError, TypeError, ValueError, json.JSONDecodeError) as exc: print(json.dumps({"ok": False, "error": str(exc)}, indent=2)) return 1 elif args.identifier_command == "migration-files": try: plan = json.loads(Path(args.plan).read_text(encoding="utf-8")) if not isinstance(plan, dict): raise TypeError("migration plan must be a JSON object") result = migrate_repository_identifier_files( plan, repo_slug=args.repo, confirm_plan_sha256=args.confirm_plan_sha256, direction=args.direction, execute=args.execute, ) except (OSError, TypeError, ValueError, json.JSONDecodeError) as exc: print(json.dumps({"ok": False, "error": str(exc)}, indent=2)) return 1 elif args.identifier_command == "migration-batch-plan": try: plan = json.loads(Path(args.plan).read_text(encoding="utf-8")) if not isinstance(plan, dict): raise TypeError("migration plan must be a JSON object") result = plan_identifier_migration_batch(plan, repo_slugs=args.repos) except (OSError, TypeError, ValueError, json.JSONDecodeError) as exc: print(json.dumps({"ok": False, "error": str(exc)}, indent=2)) return 1 if args.output: output = Path(args.output) if output.exists() and not args.force: print( json.dumps( {"ok": False, "error": f"output exists: {output}; use --force to replace"}, indent=2, ) ) return 1 output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8") elif args.identifier_command == "migration-batch-verify": try: plan = json.loads(Path(args.plan).read_text(encoding="utf-8")) batch = json.loads(Path(args.batch).read_text(encoding="utf-8")) if not isinstance(plan, dict) or not isinstance(batch, dict): raise TypeError("migration plan and batch must be JSON objects") result = verify_identifier_migration_batch(batch, plan=plan) except (OSError, TypeError, ValueError, json.JSONDecodeError) as exc: print(json.dumps({"ok": False, "error": str(exc)}, indent=2)) return 1 else: try: result = plan_identifier_migration(Path(args.root), namespace) except ValueError as exc: print(json.dumps({"ok": False, "error": str(exc)}, indent=2)) return 1 if args.output: output = Path(args.output) if output.exists() and not args.force: print( json.dumps( {"ok": False, "error": f"output exists: {output}; use --force to replace"}, indent=2, ) ) return 1 output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8") print(json.dumps(result, indent=2)) return 0 if result.get("ok") else 1 if args.command == "sbom": if not args.sbom_command: p_sbom.print_help() return 2 from repo_manager.sbom import scan_repository snapshot = scan_repository(Path(args.path), slug=args.slug) result = snapshot if args.sbom_command == "scan" else { "ok": snapshot["ok"], "repo_slug": snapshot["repo_slug"], "source_revision": snapshot["source_revision"], "generated_at": snapshot["generated_at"], "entry_count": snapshot["entry_count"], "licence_report": snapshot["licence_report"], "errors": snapshot["errors"], } if args.sbom_command == "scan" and args.output: output = Path(args.output) if output.exists() and not args.force: print( json.dumps( {"ok": False, "error": f"output exists: {output}; use --force to replace"}, indent=2, ) ) return 1 output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(snapshot, indent=2) + "\n", encoding="utf-8") print(json.dumps(result, indent=2)) return 0 if result.get("ok") else 1 if args.command == "authority": if not args.authority_command: p_authority.print_help() return 2 from repo_manager.authority import AuthorityError, resolve_record_authority try: result = resolve_record_authority( args.record_type, repo_slug=args.repo_slug, domain_slug=args.domain_slug, claimed_owner=args.claimed_owner, ) except AuthorityError as exc: print(json.dumps({"ok": False, "error": str(exc)}, indent=2)) return 1 print(json.dumps({"ok": True, **result}, indent=2)) return 0 if args.command == "cache": if not args.cache_command: p_cache.print_help() return 2 from repo_manager.cache import cache_status, fetch_closed_provenance if args.cache_command == "status": result = cache_status(Path(args.path)) elif args.cache_command == "rebuild": from repo_manager.index_store import save_index from repo_manager.observe import observe_repository snapshot, index = observe_repository(Path(args.path), slug=args.slug) index_path = save_index(index) result = { "ok": True, "rebuilt": True, "index_path": str(index_path), "record_count": snapshot["index"]["record_count"], "cache": cache_status(Path(args.path)), } else: output = Path(args.output) if output.exists() and not args.force: print( json.dumps( {"ok": False, "error": f"output exists: {output}; use --force to replace"}, indent=2, ) ) return 1 try: result = fetch_closed_provenance(args.api_base) except (OSError, TypeError, ValueError) as exc: print(json.dumps({"ok": False, "error": str(exc)}, indent=2)) return 1 output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8") result = {key: value for key, value in result.items() if key != "rows"} result["output"] = str(output) print(json.dumps(result, indent=2)) return 0 if result.get("ok") else 1 if args.command == "workload": if not args.workload_command: p_workload.print_help() return 2 from repo_manager.workloads import index_workloads, resolve_workload if args.workload_command == "index": result = index_workloads(Path(args.root)) else: result = resolve_workload( Path(args.root), rapp_id=args.rapp_id, name=args.name, deployable=args.deployable, ) print(json.dumps(result, indent=2)) return 0 if result.get("ok") else 1 if args.command == "owner-interface": if not args.owner_interface_command: p_owner_interface.print_help() return 2 from repo_manager.owner_interfaces import validate_owner_interfaces result = validate_owner_interfaces(Path(args.path), owner=args.owner) 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())