#!/usr/bin/env python3 """Inventory and ratchet code bloat and obvious pass-through wrappers.""" import argparse import re import sys from collections import Counter, namedtuple from pathlib import Path from typing import List, Tuple REPO = Path(__file__).resolve().parents[2] PRODUCTION_ROOTS = ("internal", "cmd", "desktop/internal", "web/src") SOURCE_SUFFIXES = {".go", ".ts", ".tsx", ".js", ".jsx", ".mjs"} EXCLUDED_PREFIXES = ( "cmd/agentteam-e2e-server/", "internal/api/testkit/", "internal/coredb/testkit/", "internal/store/testkit/", "desktop/internal/appbuild/templates/", ) # Test fixtures under production package roots are not shipped source and must # not affect production ratchets. This is path-based so arbitrary CJS/JS files # outside testdata remain covered. EXCLUDED_PARTS = {".git", "node_modules", "dist", "coverage", ".data", "testdata"} GO_FUNC_RE = re.compile(r"^\s*func\b") THIN_WRAPPER_RE = re.compile(r"^(return\s+)?[A-Za-z_][\w.]*\([^;]*\);?$") FileHit = namedtuple("FileHit", "file lines") FunctionHit = namedtuple("FunctionHit", "file line lines") WrapperHit = namedtuple("WrapperHit", "file line body") def rel(path: Path) -> str: return path.relative_to(REPO).as_posix() def is_excluded(path: Path) -> bool: path_rel = rel(path) if any(path_rel.startswith(prefix) for prefix in EXCLUDED_PREFIXES): return True return bool(EXCLUDED_PARTS.intersection(path.parts)) def is_test_file(path: Path) -> bool: name = path.name return name.endswith("_test.go") or ".test." in name or ".spec." in name def source_files() -> List[Path]: files: List[Path] = [] for root_name in PRODUCTION_ROOTS: root = REPO / root_name if not root.exists(): continue for path in root.rglob("*"): if not path.is_file() or is_excluded(path) or is_test_file(path): continue if path.suffix in SOURCE_SUFFIXES: files.append(path) return sorted(files) def go_functions(path: Path) -> List[Tuple[int, int, List[str]]]: lines = path.read_text(errors="ignore").splitlines() functions: List[Tuple[int, int, List[str]]] = [] index = 0 while index < len(lines): if not GO_FUNC_RE.match(lines[index]): index += 1 continue start = index depth = 0 seen_open = False cursor = index while cursor < len(lines): opens = lines[cursor].count("{") closes = lines[cursor].count("}") if opens: seen_open = True depth += opens - closes if seen_open and depth == 0: functions.append((start + 1, cursor + 1, lines[start : cursor + 1])) index = cursor break cursor += 1 index += 1 return functions def logical_body_lines(function_lines: List[str]) -> List[str]: body: List[str] = [] for raw in function_lines[1:-1]: line = raw.strip() if not line or line.startswith("//") or line in {"{", "}"}: continue body.append(line) return body def scan(max_file_line_soft: int, max_file_line_hard: int, max_go_func_lines: int) -> Tuple[List[FileHit], List[FileHit], List[FunctionHit], List[WrapperHit]]: soft_files: List[FileHit] = [] hard_files: List[FileHit] = [] large_functions: List[FunctionHit] = [] thin_wrappers: List[WrapperHit] = [] for path in source_files(): lines = path.read_text(errors="ignore").splitlines() line_count = len(lines) if line_count > max_file_line_soft: soft_files.append(FileHit(rel(path), line_count)) if line_count > max_file_line_hard: hard_files.append(FileHit(rel(path), line_count)) if path.suffix != ".go": continue for start, end, function_lines in go_functions(path): function_len = end - start + 1 if function_len > max_go_func_lines: large_functions.append(FunctionHit(rel(path), start, function_len)) body = logical_body_lines(function_lines) if len(body) != 1: continue body_text = body[0] if THIN_WRAPPER_RE.match(body_text): thin_wrappers.append(WrapperHit(rel(path), start, body_text[:160])) soft_files.sort(key=lambda hit: (-hit.lines, hit.file)) hard_files.sort(key=lambda hit: (-hit.lines, hit.file)) large_functions.sort(key=lambda hit: (-hit.lines, hit.file, hit.line)) return soft_files, hard_files, large_functions, thin_wrappers def print_file_inventory(name: str, hits: List[FileHit], *, max_examples: int = 10) -> None: print(f"{name}: {len(hits)}") for hit in hits[:max_examples]: print(f" {hit.lines:5d} {hit.file}") def print_function_inventory(name: str, hits: List[FunctionHit], *, max_examples: int = 10) -> None: print(f"{name}: {len(hits)}") for hit in hits[:max_examples]: print(f" {hit.lines:5d} {hit.file}:{hit.line}") def print_wrapper_inventory(name: str, hits: List[WrapperHit], *, max_examples: int = 10) -> None: print(f"{name}: {len(hits)}") by_file = Counter(hit.file for hit in hits) top_files = by_file.most_common(max_examples) for file, count in top_files: print(f" {count:5d} {file}") if hits: print(" examples:") for file, _ in top_files: hit = next(hit for hit in hits if hit.file == file) print(f" {hit.file}:{hit.line}: {hit.body}") def fail_if_over(name: str, actual: int, maximum: int, failures: List[str]) -> None: print(f"{name}: {actual} (max {maximum})") if actual > maximum: failures.append(f"{name} grew above baseline: {actual} > {maximum}") def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--file-line-soft-threshold", type=int, default=1000) parser.add_argument("--file-line-hard-threshold", type=int, default=2000) parser.add_argument("--go-function-line-threshold", type=int, default=120) parser.add_argument("--max-files-over-soft-threshold", type=int, required=True) parser.add_argument("--max-files-over-hard-threshold", type=int, required=True) parser.add_argument("--max-go-functions-over-threshold", type=int, required=True) parser.add_argument("--max-go-thin-wrappers", type=int, required=True) args = parser.parse_args() soft_files, hard_files, large_functions, thin_wrappers = scan( args.file_line_soft_threshold, args.file_line_hard_threshold, args.go_function_line_threshold, ) print("== Code shape inventory ==") print_file_inventory(f"production_files_over_{args.file_line_soft_threshold}_lines", soft_files) print_file_inventory(f"production_files_over_{args.file_line_hard_threshold}_lines", hard_files) print_function_inventory(f"go_functions_over_{args.go_function_line_threshold}_lines", large_functions) print_wrapper_inventory("go_thin_wrappers", thin_wrappers) failures: List[str] = [] print("\n== Ratchet baselines ==") fail_if_over("production_files_over_soft_threshold", len(soft_files), args.max_files_over_soft_threshold, failures) fail_if_over("production_files_over_hard_threshold", len(hard_files), args.max_files_over_hard_threshold, failures) fail_if_over("go_functions_over_threshold", len(large_functions), args.max_go_functions_over_threshold, failures) fail_if_over("go_thin_wrappers", len(thin_wrappers), args.max_go_thin_wrappers, failures) if failures: for failure in failures: print(f"FAIL: {failure}", file=sys.stderr) print(f"\nCode shape guardrail failed with {len(failures)} issue(s).", file=sys.stderr) return 1 print("\nCode shape guardrail passed.") return 0 if __name__ == "__main__": raise SystemExit(main())