rune/scripts/umbrella_impact.py

"""Umbrella impact checker — the change-propagation gate (#204, T32).

The umbrella manifest (config/umbrellas.json) is the change-propagation map:
when a PR touches a module, the connected areas (umbrella canaries, dependent
umbrellas, mirrors) must be addressed IN THE SAME PR or explicitly waived.

Subcommands:
  check --base <sha> --head <sha> [--body-file <path>] [--repo-root <path>]
      Resolve the diff's umbrellas; fail (exit 1) when:
        - a changed file is not in the manifest (unmapped surface), or
        - affected umbrellas have dependents and the PR neither touches
          dependent-umbrella files nor carries a Propagation statement.
  canaries --base <sha> --head <sha>
      Print the affected umbrellas' canary files (for targeted CI selection).

Exit codes: 0 = compliant; 1 = violations (actionable lines on stdout).
"""

from __future__ import annotations

import os

import argparse
import json
import subprocess
import sys
from pathlib import Path

MANIFEST = Path("config/umbrellas.json")


def load_manifest(root: Path) -> dict:
    return json.loads((root / MANIFEST).read_text())["umbrellas"]


def module_of(path: str) -> str | None:
    p = Path(path)
    if p.suffix != ".py":
        return None
    if p.parts and p.parts[0] in ("src", "shared") and "rune" in p.parts[:2] or path.startswith("shared/"):
        if path.startswith("src/rune/"):
            name = p.stem
            return None if name.startswith(("canary_", "test_", "__")) else name
        if path.startswith("shared/"):
            return p.stem
    return None


def resolve_umbrellas(changed: list[str], manifest: dict) -> tuple[dict[str, list[str]], list[str]]:
    """Map changed files → umbrellas. Returns ({umbrella: [changed modules]}, unmapped)."""
    index: dict[str, str] = {}
    for umb, data in manifest.items():
        for mod in data.get("modules", []):
            index.setdefault(mod, umb)
    affected: dict[str, list[str]] = {}
    unmapped: list[str] = []
    for f in changed:
        mod = module_of(f)
        if mod is None:
            continue
        umb = index.get(mod)
        if umb is None:
            unmapped.append(f)
        else:
            affected.setdefault(umb, []).append(mod)
    return affected, unmapped


def dependents_of(umbrellas: set[str], manifest: dict) -> set[str]:
    return {
        u
        for u, d in manifest.items()
        if d.get("depends_on") and set(d["depends_on"]) & umbrellas and u not in umbrellas
    }


def canaries_for(umbrellas: set[str], manifest: dict) -> list[str]:
    out: set[str] = set()
    for u in umbrellas:
        out.update(manifest[u].get("canaries", []))
        for dep in dependents_of({u}, manifest):
            out.update(manifest[dep].get("canaries", []))
    return sorted(out)


def changed_files(base: str, head: str, root: Path) -> list[str]:
    out = subprocess.run(
        ["git", "diff", "--name-only", f"{base}...{head}"],
        cwd=root,
        capture_output=True,
        text=True,
        check=True,
    ).stdout
    return [f for f in out.splitlines() if f.strip()]


def check(base: str, head: str, body_file: str | None, root: Path) -> int:
    manifest = load_manifest(root)
    changed = changed_files(base, head, root)
    affected, unmapped = resolve_umbrellas(changed, manifest)
    violations: list[str] = []

    if unmapped:
        violations.append(
            "UNMAPPED SURFACES — add these modules to config/umbrellas.json in this PR "
            "(the map must travel with the change): " + ", ".join(sorted(unmapped))
        )
    if not affected:
        print("umbrella-impact: no governed modules touched")
        return 1 if violations else 0

    deps = dependents_of(set(affected), manifest)
    touched_umbrellas_of_deps = {u for u, mods in affected.items() if u in deps}
    needs_propagation = bool(deps) and not touched_umbrellas_of_deps
    statement_ok = False
    if body_file:
        body = Path(body_file).read_text(errors="replace") if Path(body_file).exists() else ""
        for line in body.splitlines():
            if line.strip().lower().startswith(("propagation:", "propagation-waived:")):
                statement_ok = True
    skip_propagation = os.environ.get("UMBRELLA_SKIP_PROPAGATION") == "1"
    if needs_propagation and not statement_ok and not skip_propagation:
        violations.append(
            f"PROPAGATION REQUIRED — touched {sorted(affected)}; dependent umbrellas "
            f"{sorted(deps)} may need updates. Either include dependent-umbrella files in this "
            f"PR, or add a body line 'Propagation: <what you verified/updated>' or "
            f"'Propagation-waived: <reason>'."
        )
    print(
        f"umbrella-impact: affected={sorted(affected)} dependents={sorted(deps)} "
        f"canaries={canaries_for(set(affected), manifest)}"
    )
    for v in violations:
        print("umbrella-impact:", v)
    return 1 if violations else 0


def main(argv: list[str]) -> int:
    ap = argparse.ArgumentParser(prog="umbrella_impact")
    sub = ap.add_subparsers(dest="cmd", required=True)
    p_check = sub.add_parser("check")
    p_check.add_argument("--base", required=True)
    p_check.add_argument("--head", default="HEAD")
    p_check.add_argument("--body-file")
    p_check.add_argument("--repo-root", default=".")
    p_can = sub.add_parser("canaries")
    p_can.add_argument("--base", required=True)
    p_can.add_argument("--head", default="HEAD")
    p_can.add_argument("--repo-root", default=".")
    args = ap.parse_args(argv[1:])
    root = Path(args.repo_root)
    if args.cmd == "check":
        return check(args.base, args.head, args.body_file, root)
    manifest = load_manifest(root)
    changed = changed_files(args.base, args.head, root)
    affected, unmapped = resolve_umbrellas(changed, manifest)
    for c in canaries_for(set(affected), manifest):
        print(f"src/rune/{c}.py")
    return 0 if not unmapped else 1


if __name__ == "__main__":
    sys.exit(main(sys.argv))