check_package_architecture_maps.py

#!/usr/bin/env python3
"""Enforce architecture maps for oversized source packages.

Existing mapless packages are grandfathered at their committed source-file
count. They emit warnings until documented. A new oversized package, or growth
past a grandfathered count, fails with guidance to add a package-root map.
"""

from __future__ import annotations

import argparse
import json
import os
import subprocess
from dataclasses import dataclass
from pathlib import Path
from typing import Mapping

ISSUE_URL = "https://github.com/BasedHardware/omi/issues/9446"
DEFAULT_THRESHOLD = 12
MAP_NAMES = ("ARCHITECTURE.md", "README.md")
SOURCE_EXTENSIONS = {
    ".c",
    ".cc",
    ".cpp",
    ".cxx",
    ".dart",
    ".h",
    ".hpp",
    ".js",
    ".jsx",
    ".kt",
    ".m",
    ".mm",
    ".py",
    ".rs",
    ".swift",
    ".ts",
    ".tsx",
}
SKIP_SUFFIXES = (".gen.dart", ".g.dart", ".lock", ".min.js")
SKIP_PARTS = {".git", ".next", ".venv", "__pycache__", "build", "dist", "node_modules", "target"}


@dataclass(frozen=True)
class Finding:
    package: str
    source_count: int
    baseline_count: int | None
    level: str
    message: str


def source_file(path: Path) -> bool:
    if path.suffix not in SOURCE_EXTENSIONS:
        return False
    if path.name.endswith(SKIP_SUFFIXES):
        return False
    return not any(part in SKIP_PARTS for part in path.parts)


def package_roots(repo_root: Path) -> list[Path]:
    """Return intentionally non-overlapping package roots covered by the ratchet."""

    backend_utils = repo_root / "backend" / "utils"
    packages = sorted(path for path in backend_utils.iterdir() if path.is_dir()) if backend_utils.is_dir() else []
    agent_source = repo_root / "desktop" / "macos" / "agent" / "src"
    if agent_source.is_dir():
        packages.append(agent_source)
    return packages


def source_count(package_root: Path) -> int:
    return sum(1 for path in package_root.rglob("*") if path.is_file() and source_file(path))


def has_architecture_map(package_root: Path) -> bool:
    return any((package_root / name).is_file() for name in MAP_NAMES)


def load_baseline(path: Path) -> dict[str, int]:
    try:
        raw = path.read_text(encoding="utf-8")
    except (OSError, json.JSONDecodeError) as exc:
        raise ValueError(f"cannot read package architecture baseline {path}: {exc}") from exc
    return parse_baseline(raw, label=str(path))


def parse_baseline(raw: str, *, label: str) -> dict[str, int]:
    try:
        payload = json.loads(raw)
    except json.JSONDecodeError as exc:
        raise ValueError(f"cannot parse package architecture baseline {label}: {exc}") from exc
    if not isinstance(payload, dict) or payload.get("version") != 1 or not isinstance(payload.get("packages"), dict):
        raise ValueError(f"invalid package architecture baseline schema: {label}")
    baseline: dict[str, int] = {}
    for raw_package, raw_count in payload["packages"].items():
        if not isinstance(raw_package, str) or isinstance(raw_count, bool) or not isinstance(raw_count, int):
            raise ValueError(f"invalid package architecture baseline entry: {raw_package!r}: {raw_count!r}")
        baseline[raw_package] = raw_count
    return baseline


def load_baseline_at_ref(repo_root: Path, ref: str) -> dict[str, int] | None:
    """Load the trusted baseline from a PR base; None means first-time bootstrap."""

    verify = subprocess.run(
        ["git", "rev-parse", "--verify", f"{ref}^{{commit}}"],
        cwd=repo_root,
        check=False,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.PIPE,
        text=True,
    )
    if verify.returncode:
        raise ValueError(f"cannot resolve package architecture baseline ref {ref}: {verify.stderr.strip()}")
    baseline_path = ".github/scripts/package_architecture_baseline.json"
    result = subprocess.run(
        ["git", "show", f"{ref}:{baseline_path}"],
        cwd=repo_root,
        check=False,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
    )
    if result.returncode:
        return None
    return parse_baseline(result.stdout, label=f"{ref}:{baseline_path}")


def baseline_change_findings(current: Mapping[str, int], previous: Mapping[str, int] | None) -> list[Finding]:
    """Reject self-grandfathering while allowing baseline shrinkage and removal."""

    if previous is None:
        return []
    findings: list[Finding] = []
    for package, current_count in sorted(current.items()):
        previous_count = previous.get(package)
        if previous_count is not None and current_count <= previous_count:
            continue
        if previous_count is None:
            message = (
                f"{package} was added to the architecture-map baseline at {current_count}. New oversized packages "
                f"must add ARCHITECTURE.md or README.md instead of grandfathering themselves. See {ISSUE_URL}"
            )
        else:
            message = (
                f"{package}'s architecture-map baseline increased from {previous_count} to {current_count}. "
                f"Add ARCHITECTURE.md or README.md instead of raising the baseline. See {ISSUE_URL}"
            )
        findings.append(Finding(package, current_count, previous_count, "error", message))
    return findings


def evaluate_packages(repo_root: Path, baseline: Mapping[str, int], threshold: int) -> list[Finding]:
    findings: list[Finding] = []
    for package_root in package_roots(repo_root):
        relative = package_root.relative_to(repo_root).as_posix()
        count = source_count(package_root)
        if count <= threshold or has_architecture_map(package_root):
            continue
        baseline_count = baseline.get(relative)
        if baseline_count is None:
            message = (
                f"{relative} has {count} source files (threshold: {threshold}) and no package-root "
                f"ARCHITECTURE.md or README.md. Add a map before this package grows further. See {ISSUE_URL}"
            )
            level = "error"
        elif count > baseline_count:
            message = (
                f"{relative} grew from its grandfathered {baseline_count} source files to {count} without a "
                f"package-root ARCHITECTURE.md or README.md. Add a map or revert the growth. See {ISSUE_URL}"
            )
            level = "error"
        else:
            message = (
                f"{relative} remains grandfathered without an architecture map at {count} source files "
                f"(baseline: {baseline_count}); add ARCHITECTURE.md or README.md opportunistically."
            )
            level = "warning"
        findings.append(Finding(relative, count, baseline_count, level, message))
    return findings


def annotation_escape(value: object) -> str:
    return str(value).replace("%", "%25").replace("\r", "%0D").replace("\n", "%0A")


def emit_finding(finding: Finding) -> None:
    title = "Package architecture map required" if finding.level == "error" else "Package needs architecture map"
    print(
        f"::{finding.level} file={annotation_escape(finding.package)},line=1,title={annotation_escape(title)}::"
        f"{annotation_escape(finding.message)}"
    )


def write_summary(findings: list[Finding], threshold: int) -> None:
    lines = [
        "## Package architecture maps",
        "",
        f"Packages over {threshold} source files require a package-root `ARCHITECTURE.md` or `README.md`.",
        "",
    ]
    if not findings:
        lines.append("All oversized packages have maps or remain below the threshold.")
    else:
        lines.extend(["| Level | Package | Source files |", "| --- | --- | ---: |"])
        lines.extend(f"| {item.level} | {item.package} | {item.source_count} |" for item in findings)
    summary = "\n".join(lines) + "\n"
    summary_path = os.environ.get("GITHUB_STEP_SUMMARY")
    if summary_path:
        with open(summary_path, "a", encoding="utf-8") as handle:
            handle.write(summary)
    else:
        print(summary, end="")


def run(
    *,
    repo_root: Path,
    baseline_path: Path,
    threshold: int = DEFAULT_THRESHOLD,
    previous_baseline: Mapping[str, int] | None = None,
) -> int:
    try:
        baseline = load_baseline(baseline_path)
    except ValueError as exc:
        print(f"::error title=Invalid package architecture baseline::{annotation_escape(exc)}")
        return 1
    findings = baseline_change_findings(baseline, previous_baseline)
    findings.extend(evaluate_packages(repo_root, baseline, threshold))
    for finding in findings:
        emit_finding(finding)
    write_summary(findings, threshold)
    return 1 if any(finding.level == "error" for finding in findings) else 0


def main() -> int:
    default_root = Path(__file__).resolve().parents[2]
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--root", type=Path, default=default_root)
    parser.add_argument("--baseline", type=Path)
    parser.add_argument("--base", help="Git revision whose baseline cannot be increased or extended")
    parser.add_argument("--threshold", type=int, default=DEFAULT_THRESHOLD)
    args = parser.parse_args()
    repo_root = args.root.resolve()
    baseline_path = args.baseline or repo_root / ".github" / "scripts" / "package_architecture_baseline.json"
    try:
        previous_baseline = load_baseline_at_ref(repo_root, args.base) if args.base else None
    except ValueError as exc:
        print(f"::error title=Invalid package architecture baseline::{annotation_escape(exc)}")
        return 1
    return run(
        repo_root=repo_root,
        baseline_path=baseline_path,
        threshold=args.threshold,
        previous_baseline=previous_baseline,
    )


if __name__ == "__main__":
    raise SystemExit(main())