#!/usr/bin/env python3 """ Red Queen — C5 CLI (C5A + C5B + C5C): kontrollierte Inspektion & Dry-Run. KEIN produktiver Poll-Daemon. KEINE daemon/start-forever-Funktion. Nur explizite, begrenzte Befehle: * poll-once — ein kontrollierter Poll-Zyklus (read-only) * show-pending — ausstehende Commits anzeigen * show-commit — Details eines Commits anzeigen * dry-run-diff — Diff eines Commits anzeigen (read-only) * health — Health-Contract anzeigen * no-write-check — No-Downstream-Write-Guarantee pruefen * propagate-plan — Propagation-Plan fuer einen Commit (read-only, KEIN Write) * live-dry-run — Read-only Live-Dry-Run gegen realen Forgejo + Tolaria Stand * c5c-guarantees — No-Search- und No-Master-Write-Guarantee pruefen (statisch) """ from __future__ import annotations import argparse import json import os import sys from typing import Any, Dict, List, Optional from rq_c5a import C5AStore from rq_c5b import ( C5BPoller, GitReader, assert_no_downstream_write, DEFAULT_POLL_INTERVAL_SECONDS, ENV_POLL_INTERVAL, ) from rq_c5c import ( C5CPropagator, C5CDryRun, TolariaClient, assert_no_search_calls, assert_no_master_write, DEFAULT_TOLARIA_BASE, ENV_TOLARIA_BASE, ) def _store(db_path: str) -> C5AStore: return C5AStore(db_path) def _poller(store: C5AStore, repo_path: str, interval: Optional[int]) -> C5BPoller: return C5BPoller(store, repo_path, poll_interval_seconds=interval) def cmd_poll_once(args: argparse.Namespace) -> int: store = _store(args.db) poller = _poller(store, args.repo, args.interval) result = poller.poll_once() print(json.dumps(result, ensure_ascii=False, indent=2)) store.close() return 0 def cmd_show_pending(args: argparse.Namespace) -> int: store = _store(args.db) pending = store.list_commits() print(json.dumps(pending, ensure_ascii=False, indent=2)) store.close() return 0 def cmd_show_commit(args: argparse.Namespace) -> int: store = _store(args.db) commit = store.get_commit(args.sha) if commit is None: print(f"Commit nicht gefunden: {args.sha}") store.close() return 1 print(json.dumps(commit, ensure_ascii=False, indent=2)) objs = store.list_object_changes(args.sha) if objs: print("--- Object Changes ---") print(json.dumps(objs, ensure_ascii=False, indent=2)) store.close() return 0 def cmd_dry_run_diff(args: argparse.Namespace) -> int: reader = GitReader(args.repo) changes = reader.diff_name_status(args.sha) print(json.dumps(changes, ensure_ascii=False, indent=2)) return 0 def cmd_health(args: argparse.Namespace) -> int: store = _store(args.db) print(json.dumps(store.health(), ensure_ascii=False, indent=2)) store.close() return 0 def cmd_no_write_check(args: argparse.Namespace) -> int: result = assert_no_downstream_write() print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 if result["no_downstream_write"] else 1 def _c5c_client(args: argparse.Namespace) -> TolariaClient: return TolariaClient(base_url=args.tolaria_base) def cmd_propagate_plan(args: argparse.Namespace) -> int: """Erzeugt den Propagation-Plan fuer einen Commit (read-only, KEIN Write).""" store = _store(args.db) reader = GitReader(args.repo) dry = C5CDryRun(store, reader, client=_c5c_client(args)) plan = dry.plan_commit(args.sha) print(json.dumps(plan, ensure_ascii=False, indent=2)) store.close() return 0 def cmd_live_dry_run(args: argparse.Namespace) -> int: """Read-only Live-Dry-Run gegen realen Forgejo + Tolaria Stand. Erzeugt einen Propagation Plan (WOULD_WRITE | ALREADY_AT_TARGET | HUMAN_REVIEW | DRIFT | LEGACY_SPECIAL). Fuehrt KEINE produktive Propagation aus. KEINE Writes. """ store = _store(args.db) reader = GitReader(args.repo) dry = C5CDryRun(store, reader, client=_c5c_client(args)) commits = store.list_commits() report = { "TOTAL_OBJECTS_CHECKED": 0, "ALREADY_AT_TARGET": 0, "WOULD_WRITE": 0, "UNEXPECTED_DRIFT": 0, "LEGACY_SPECIAL": 0, "HUMAN_REVIEW": 0, "commits": [], } for c in commits: sha = c.get("commit_sha") or c.get("sha") if not sha: continue plan = dry.plan_commit(sha) for item in plan.get("plan", []): report["TOTAL_OBJECTS_CHECKED"] += 1 p = item.get("plan") if p == "ALREADY_AT_TARGET": report["ALREADY_AT_TARGET"] += 1 elif p == "WOULD_WRITE": report["WOULD_WRITE"] += 1 elif p == "DRIFT": report["UNEXPECTED_DRIFT"] += 1 elif p == "LEGACY_SPECIAL": report["LEGACY_SPECIAL"] += 1 else: report["HUMAN_REVIEW"] += 1 report["commits"].append({"commit_sha": sha, "plan": plan.get("plan", [])}) print(json.dumps(report, ensure_ascii=False, indent=2)) store.close() return 0 def cmd_c5c_guarantees(args: argparse.Namespace) -> int: """Prueft No-Search- und No-Master-Write-Guarantee (statisch).""" search = assert_no_search_calls() master = assert_no_master_write() result = {"no_search_calls": search, "no_master_write": master} print(json.dumps(result, ensure_ascii=False, indent=2)) ok = search["no_search_calls"] and master["no_master_write"] return 0 if ok else 1 def main(argv: Optional[List[str]] = None) -> int: parser = argparse.ArgumentParser( prog="rq_c5_cli", description="Red Queen C5 CLI (C5A + C5B) — kontrollierte Inspektion & Dry-Run", ) parser.add_argument( "--db", default=os.environ.get("C5A_DB", "c5a.db"), help="Pfad zur C5A-SQLite-DB (Env C5A_DB, Default c5a.db)", ) parser.add_argument( "--repo", default=os.environ.get("C5_REPO", "."), help="Pfad zum Forgejo-Clone (Env C5_REPO, Default .)", ) parser.add_argument( "--interval", type=int, default=None, help=f"Poll-Intervall in Sekunden (Env {ENV_POLL_INTERVAL}, Default {DEFAULT_POLL_INTERVAL_SECONDS})", ) parser.add_argument( "--tolaria-base", default=os.environ.get(ENV_TOLARIA_BASE, DEFAULT_TOLARIA_BASE), help=f"Tolaria-Vault-API-Basis (Env {ENV_TOLARIA_BASE}, Default {DEFAULT_TOLARIA_BASE})", ) sub = parser.add_subparsers(dest="command", required=True) p = sub.add_parser("poll-once", help="Ein kontrollierter Poll-Zyklus (read-only)") p.set_defaults(func=cmd_poll_once) p = sub.add_parser("show-pending", help="Ausstehende Commits anzeigen") p.set_defaults(func=cmd_show_pending) p = sub.add_parser("show-commit", help="Details eines Commits anzeigen") p.add_argument("sha", help="Commit-SHA") p.set_defaults(func=cmd_show_commit) p = sub.add_parser("dry-run-diff", help="Diff eines Commits anzeigen (read-only)") p.add_argument("sha", help="Commit-SHA") p.set_defaults(func=cmd_dry_run_diff) p = sub.add_parser("health", help="Health-Contract anzeigen") p.set_defaults(func=cmd_health) p = sub.add_parser("no-write-check", help="No-Downstream-Write-Guarantee pruefen") p.set_defaults(func=cmd_no_write_check) p = sub.add_parser("propagate-plan", help="Propagation-Plan fuer einen Commit (read-only, KEIN Write)") p.add_argument("sha", help="Commit-SHA") p.set_defaults(func=cmd_propagate_plan) p = sub.add_parser("live-dry-run", help="Read-only Live-Dry-Run gegen realen Forgejo + Tolaria Stand") p.set_defaults(func=cmd_live_dry_run) p = sub.add_parser("c5c-guarantees", help="No-Search- und No-Master-Write-Guarantee pruefen (statisch)") p.set_defaults(func=cmd_c5c_guarantees) args = parser.parse_args(argv) return args.func(args) if __name__ == "__main__": sys.exit(main())