#!/usr/bin/env python3 """Repository governance checks — outside ESS Loop. Validates: 1. Migration stem uniqueness under apps/api/migrations/ 2. ECR identity uniqueness (H1 / Canonical-ID) under docs/ECR/ Usage: python scripts/repo-governance-check.py python scripts/repo-governance-check.py --print-anchors Exit 0 = PASS; non-zero = FAIL (merge/gate blocker). """ from __future__ import annotations import argparse import re import sys from collections import defaultdict from pathlib import Path ROOT = Path(__file__).resolve().parents[1] MIG_DIR = ROOT / "apps" / "api" / "migrations" ECR_DIR = ROOT / "docs" / "ECR" # Full stem: 000018_admin_rbac (with or without .up/.down) MIG_STEM_RE = re.compile(r"^(\d{6}_.+?)(?:\.(?:up|down))?\.sql$", re.I) # Numeric prefix only (collision surface across branches) MIG_NUM_RE = re.compile(r"^(\d{6})_") ECR_ID_RE = re.compile(r"^ECR-\d+[A-Z]?(?:-[a-z0-9]+(?:-[a-z0-9]+)*)?$", re.I) H1_RE = re.compile(r"^#\s+(ECR-\S+)", re.M) CANON_RE = re.compile(r"(?im)^\*\*?Canonical-ID:\*\*?\s*`?(ECR-[A-Za-z0-9-]+)`?\s*$") def migration_stems(paths: list[Path]) -> dict[str, list[Path]]: by_stem: dict[str, list[Path]] = defaultdict(list) for p in paths: m = MIG_STEM_RE.match(p.name) if not m: continue by_stem[m.group(1)].append(p) return by_stem def migration_numbers(paths: list[Path]) -> dict[str, list[str]]: """Map 6-digit version → distinct full stems (duplicate numbers = FAIL).""" by_num: dict[str, set[str]] = defaultdict(set) for p in paths: m = MIG_STEM_RE.match(p.name) if not m: continue stem = m.group(1) num_m = MIG_NUM_RE.match(stem) if not num_m: continue by_num[num_m.group(1)].add(stem) return {k: sorted(v) for k, v in by_num.items()} def ecr_identity(path: Path) -> str | None: text = path.read_text(encoding="utf-8") canon = CANON_RE.search(text) if canon: return canon.group(1) h1 = H1_RE.search(text) if h1: # Strip trailing punctuation from H1 token return h1.group(1).rstrip(".:)") return None def max_migration_num(paths: list[Path]) -> int: n = 0 for p in paths: m = MIG_STEM_RE.match(p.name) if not m: continue num_m = MIG_NUM_RE.match(m.group(1)) if num_m: n = max(n, int(num_m.group(1))) return n def next_ecr_suggestion(identities: list[str]) -> str: """Next unused bare ECR-NNN (ignores letter/suffix variants for suggestion only).""" nums: list[int] = [] for i in identities: m = re.match(r"ECR-(\d+)", i, re.I) if m: nums.append(int(m.group(1))) nxt = (max(nums) + 1) if nums else 1 return f"ECR-{nxt:03d}" def check() -> int: errors: list[str] = [] warnings: list[str] = [] if not MIG_DIR.is_dir(): errors.append(f"missing migrations dir: {MIG_DIR}") mig_files: list[Path] = [] else: mig_files = sorted(MIG_DIR.glob("*.sql")) if not ECR_DIR.is_dir(): errors.append(f"missing ECR dir: {ECR_DIR}") ecr_files: list[Path] = [] else: ecr_files = sorted(ECR_DIR.glob("ECR-*.md")) # --- migrations --- by_stem = migration_stems(mig_files) for stem, files in sorted(by_stem.items()): # Same stem with both .up/.down is OK if only those; identical duplicate names shouldn't happen names = [f.name for f in files] if len(files) > 2 or (len(files) == 2 and not _up_down_pair(names)): if len(set(names)) != len(names): errors.append(f"migration duplicate filename: {names}") elif len(files) > 2: errors.append(f"migration stem over-shared ({stem}): {names}") by_num = migration_numbers(mig_files) for num, stems in sorted(by_num.items()): if len(stems) > 1: errors.append( f"migration number collision {num}: {', '.join(stems)} " f"(allocate next from TRACEABILITY Max Migration; never reuse)" ) # --- ECR identity --- by_id: dict[str, list[Path]] = defaultdict(list) missing: list[Path] = [] invalid: list[tuple[Path, str]] = [] for p in ecr_files: ident = ecr_identity(p) if not ident: missing.append(p) continue if not ECR_ID_RE.match(ident): invalid.append((p, ident)) continue by_id[ident].append(p) for p in missing: errors.append(f"ECR missing identity (H1 or Canonical-ID): {p.name}") for p, ident in invalid: errors.append(f"ECR invalid identity {ident!r} in {p.name}") for ident, files in sorted(by_id.items()): if len(files) > 1: errors.append( f"ECR identity collision {ident}: " + ", ".join(f.name for f in files) + " (ECR-NNN is globally unique; use ECR-NNN-suffix or ECR-NNNA)" ) # Soft: bare number reserved — if ECR-012 and ECR-012-star both exist, OK; # if two different files both declare ECR-012, caught above. identities = list(by_id.keys()) max_mig = max_migration_num(mig_files) next_ecr = next_ecr_suggestion(identities) if errors: print("REPO GOVERNANCE: FAIL") for e in errors: print(f" ERROR: {e}") for w in warnings: print(f" WARN: {w}") return 1 print("REPO GOVERNANCE: PASS") print(f" migrations: {len(by_num)} unique numbers (max {max_mig:06d})") print(f" ECR identities: {len(by_id)} unique") print(f" suggested Next ECR: {next_ecr}") print(f" suggested Max Migration: {max_mig:06d}") for w in warnings: print(f" WARN: {w}") return 0 def _up_down_pair(names: list[str]) -> bool: if len(names) != 2: return False a, b = sorted(names) return a.endswith(".up.sql") and b.endswith(".down.sql") and a[: -len(".up.sql")] == b[: -len(".down.sql")] def print_anchors() -> int: mig_files = sorted(MIG_DIR.glob("*.sql")) if MIG_DIR.is_dir() else [] ecr_files = sorted(ECR_DIR.glob("ECR-*.md")) if ECR_DIR.is_dir() else [] identities = [] for p in ecr_files: i = ecr_identity(p) if i: identities.append(i) max_mig = max_migration_num(mig_files) print(f"Next ECR: {next_ecr_suggestion(identities)}") print(f"Max Migration: {max_mig:06d}") return 0 def main() -> int: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument( "--print-anchors", action="store_true", help="Print Next ECR / Max Migration suggestions and exit 0", ) args = ap.parse_args() if args.print_anchors: return print_anchors() return check() if __name__ == "__main__": sys.exit(main())