fold/scripts/check_source_structure.py

#!/usr/bin/env python3
"""Enforce Fold's source-size ratchet and report decomposition targets."""

from __future__ import annotations

import argparse
import fnmatch
import json
import re
import sys
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[1]
DEFAULT_BUDGET = ROOT / "tests" / "architecture" / "source_budgets.json"
SOURCE_SUFFIXES = {".cc", ".cpp", ".cxx", ".h", ".hh", ".hpp", ".hxx"}


def source_files(root: Path) -> list[Path]:
    return sorted(
        path
        for directory in (root / "src", root / "include" / "fold")
        for path in directory.rglob("*")
        if path.is_file() and path.suffix in SOURCE_SUFFIXES
    )


def line_count(path: Path) -> int:
    return len(path.read_text(encoding="utf-8").splitlines())


def require_integer(value: object, name: str) -> int:
    if isinstance(value, bool) or not isinstance(value, int):
        raise ValueError(f"{name} must be an integer")
    return value


def validate_budget(
    root: Path, config: dict[str, Any]
) -> tuple[list[dict[str, Any]], list[str]]:
    default_limit = require_integer(
        config.get("new_file_max_lines"), "new_file_max_lines"
    )
    tracked_rows = config.get("tracked_monoliths", [])
    if default_limit < 1 or not isinstance(tracked_rows, list):
        raise ValueError("invalid source budget schema")

    tracked: dict[str, dict[str, Any]] = {}
    for row in tracked_rows:
        if not isinstance(row, dict) or not isinstance(row.get("path"), str):
            raise ValueError("tracked_monoliths entries require a string path")
        tracked[row["path"]] = row
    observations: list[dict[str, Any]] = []
    errors: list[str] = []
    seen: set[str] = set()

    for path in source_files(root):
        relative = path.relative_to(root).as_posix()
        lines = line_count(path)
        budget = tracked.get(relative)
        if budget is None:
            if lines > default_limit:
                errors.append(
                    f"untracked source monolith {relative}: {lines} > {default_limit} lines"
                )
            continue

        seen.add(relative)
        maximum = require_integer(budget.get("max_lines"), f"{relative}.max_lines")
        target = require_integer(budget.get("target_lines"), f"{relative}.target_lines")
        if target < 1 or maximum < target:
            errors.append(f"invalid max/target line budget for {relative}")
        if lines > maximum:
            errors.append(f"{relative} grew to {lines} lines (ratchet {maximum})")
        observations.append(
            {
                "path": relative,
                "lines": lines,
                "max_lines": maximum,
                "target_lines": target,
                "excess_over_target": max(0, lines - target),
                "next_extraction": str(budget.get("next_extraction", "")),
            }
        )

    for missing in sorted(set(tracked) - seen):
        errors.append(f"tracked source path is missing: {missing}")
    observations.sort(key=lambda row: (-row["excess_over_target"], row["path"]))
    return observations, errors


def validate_architecture(root: Path, config: dict[str, Any]) -> list[str]:
    """Apply narrow, explicit source-boundary rules alongside size budgets."""

    architecture = config.get("architecture", {})
    if not isinstance(architecture, dict):
        raise ValueError("architecture must be an object")
    if "required_content" in architecture:
        raise ValueError(
            "architecture.required_content is brittle; enforce the contract with a symbol, "
            "AST, or behavior test"
        )

    errors: list[str] = []
    for rule in architecture.get("forbidden_patterns", []):
        if not isinstance(rule, dict):
            raise ValueError("architecture.forbidden_patterns entries must be objects")
        name = str(rule["name"])
        pattern = re.compile(str(rule["regex"]))
        paths = [
            str(value) for value in rule.get("paths", ["src/**", "include/fold/**"])
        ]
        allowed = {str(value) for value in rule.get("allow", [])}
        for path in source_files(root):
            relative = path.relative_to(root).as_posix()
            if relative in allowed or not any(
                fnmatch.fnmatch(relative, glob) for glob in paths
            ):
                continue
            for line_number, line in enumerate(
                path.read_text(encoding="utf-8").splitlines(), 1
            ):
                if pattern.search(line):
                    errors.append(f"{name}: {relative}:{line_number}")
    return errors


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--root", type=Path, default=ROOT)
    parser.add_argument("--budget", type=Path, default=DEFAULT_BUDGET)
    parser.add_argument(
        "--json", action="store_true", help="emit a machine-readable report"
    )
    args = parser.parse_args()

    root = args.root.resolve()
    try:
        config = json.loads(args.budget.read_text(encoding="utf-8"))
        observations, errors = validate_budget(root, config)
        errors.extend(validate_architecture(root, config))
    except (OSError, ValueError, KeyError, TypeError, json.JSONDecodeError) as error:
        print(f"source-structure configuration error: {error}", file=sys.stderr)
        return 2

    if args.json:
        print(
            json.dumps(
                {
                    "status": "failed" if errors else "ok",
                    "files": observations,
                    "errors": errors,
                },
                indent=2,
                sort_keys=True,
            )
        )
    else:
        for row in observations:
            print(
                f"{row['path']}: {row['lines']} lines; target {row['target_lines']}; "
                f"next: {row['next_extraction']}"
            )
        for error in errors:
            print(f"error: {error}", file=sys.stderr)
    return 1 if errors else 0


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