digital-psychology/scripts/repo-governance-check.py

#!/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())