ITWorx-Pulse-Public/tools/check_wiring.py

#!/usr/bin/env python3
"""Fail when a first-party Go package is implemented but unreachable from a binary.

The audit that motivated this tool found ~2,600 lines of fully implemented,
well-tested Go code with zero non-test importers outside its own package —
`cmd/worker` and `cmd/agent` never called it, so it shipped as dead weight
behind a "done" task. Acceptance checks had only ever exercised each
package's internal behaviour, never whether `go build ./cmd/...` would ever
pull it in.

This tool computes, using pure-Python source parsing (no `go list`, since the
Go toolchain may be unavailable in CI), the transitive set of first-party
packages reachable from the real entry points (`cmd/api`, `cmd/worker`,
`cmd/agent`, `cmd/migrate`) through non-test imports, and fails listing every
`internal/*` package that is implemented but unreachable and not explicitly
allowlisted with a reason and a tracking task id.
"""

from __future__ import annotations

import json
import re
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[1]
DEFAULT_ALLOWLIST = ROOT / "tools" / "wiring_allowlist.json"

GO_ROOTS = ("cmd", "internal")
SKIP_PARTS = {".git", "node_modules", "vendor"}

MODULE_RE = re.compile(r"(?m)^module\s+(\S+)")
PACKAGE_RE = re.compile(r"(?m)^package\s+\w+\b")
PACKAGE_MAIN_RE = re.compile(r"(?m)^package\s+main\b")
IGNORE_TAG_RE = re.compile(
    r"(?m)^//go:build\b.*\bignore\b|^//\s*\+build\b.*\bignore\b"
)
IMPORT_BLOCK_RE = re.compile(r"import\s*\(\s*(.*?)\n\)", re.S)
IMPORT_SINGLE_RE = re.compile(
    r'(?m)^\s*import\s+(?:(?:[A-Za-z_][A-Za-z0-9_]*|_|\.)\s+)?"([^"]+)"'
)
IMPORT_LINE_RE = re.compile(
    r'^\s*(?:(?:[A-Za-z_][A-Za-z0-9_]*|_|\.)\s+)?"([^"]+)"'
)


class AllowlistError(Exception):
    """Raised when the wiring allowlist file is malformed."""


@dataclass
class WiringReport:
    module: str
    entry_points: list[str]
    implemented_internal: set[str]
    reachable: set[str]
    unreachable: set[str]
    allowlist_entries: list[dict[str, Any]]
    allowlisted_active: dict[str, dict[str, Any]]
    stale_allowlist: dict[str, dict[str, Any]]
    unknown_allowlist: dict[str, dict[str, Any]]
    failing: set[str] = field(init=False)

    def __post_init__(self) -> None:
        self.failing = self.unreachable - set(self.allowlisted_active)

    @property
    def ok(self) -> bool:
        return not self.failing and not self.stale_allowlist and not self.unknown_allowlist


def read_module_path(root: Path) -> str:
    text = (root / "go.mod").read_text(encoding="utf-8")
    match = MODULE_RE.search(text)
    if not match:
        raise AllowlistError("go.mod has no module declaration")
    return match.group(1)


def discover_go_files(root: Path) -> list[Path]:
    files: list[Path] = []
    for go_root in GO_ROOTS:
        base = root / go_root
        if not base.is_dir():
            continue
        for path in base.rglob("*.go"):
            if any(part in SKIP_PARTS for part in path.parts):
                continue
            files.append(path)
    return files


def header_text(text: str) -> str:
    """Return the text preceding the package clause (build tags, doc comments)."""
    match = PACKAGE_RE.search(text)
    return text[: match.start()] if match else text


def has_ignore_build_tag(text: str) -> bool:
    return bool(IGNORE_TAG_RE.search(header_text(text)))


def parse_imports(text: str) -> set[str]:
    """Extract first-party-or-not import paths from grouped and single import forms.

    Handles grouped `import (...)` blocks, single `import "x"` statements,
    aliases (`alias "x"`), blank imports (`_ "x"`), and dot imports (`. "x"`).
    Blank and dot imports are still real reachability edges: they pull the
    package into the binary and run its `init()`.
    """
    imports: set[str] = set()
    for block in IMPORT_BLOCK_RE.finditer(text):
        for line in block.group(1).splitlines():
            match = IMPORT_LINE_RE.match(line)
            if match:
                imports.add(match.group(1))
    for match in IMPORT_SINGLE_RE.finditer(text):
        imports.add(match.group(1))
    return imports


def classify_files(
    files: list[Path],
) -> tuple[dict[str, list[Path]], dict[str, list[Path]]]:
    """Split files by package directory into (non-test, test) buckets.

    Files carrying a `//go:build ignore` (or legacy `// +build ignore`) tag
    are dropped entirely: the Go toolchain never compiles them, so they must
    not count as either "implemented" or as import edges.
    """
    prod_by_dir: dict[str, list[Path]] = {}
    test_by_dir: dict[str, list[Path]] = {}
    for path in files:
        text = path.read_text(encoding="utf-8", errors="replace")
        if has_ignore_build_tag(text):
            continue
        directory = path.parent
        key = directory.as_posix()
        if path.name.endswith("_test.go"):
            test_by_dir.setdefault(key, []).append(path)
        else:
            prod_by_dir.setdefault(key, []).append(path)
    return prod_by_dir, test_by_dir


def relativize(prod_by_dir: dict[str, list[Path]], root: Path) -> dict[str, list[Path]]:
    result: dict[str, list[Path]] = {}
    for key, paths in prod_by_dir.items():
        rel = Path(key).resolve().relative_to(root.resolve()).as_posix()
        result[rel] = paths
    return result


def build_graph(prod_by_dir: dict[str, list[Path]], module: str) -> dict[str, set[str]]:
    prefix = module + "/"
    graph: dict[str, set[str]] = {}
    for directory, paths in prod_by_dir.items():
        edges: set[str] = set()
        for path in paths:
            text = path.read_text(encoding="utf-8", errors="replace")
            for imp in parse_imports(text):
                if not imp.startswith(prefix):
                    continue
                target = imp[len(prefix):]
                if target in prod_by_dir:
                    edges.add(target)
        graph[directory] = edges
    return graph


def find_entry_points(prod_by_dir: dict[str, list[Path]]) -> set[str]:
    entries: set[str] = set()
    for directory, paths in prod_by_dir.items():
        if not directory.startswith("cmd/"):
            continue
        for path in paths:
            text = path.read_text(encoding="utf-8", errors="replace")
            if PACKAGE_MAIN_RE.search(text):
                entries.add(directory)
                break
    return entries


def reachable_from(graph: dict[str, set[str]], roots: set[str]) -> set[str]:
    seen: set[str] = set()
    stack = list(roots)
    while stack:
        current = stack.pop()
        if current in seen:
            continue
        seen.add(current)
        for target in graph.get(current, ()):
            if target not in seen:
                stack.append(target)
    return seen


def load_allowlist(path: Path) -> list[dict[str, Any]]:
    if not path.exists():
        return []
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as error:
        raise AllowlistError(f"{path}: invalid JSON ({error})") from error

    if isinstance(data, dict):
        entries = data.get("entries")
    else:
        entries = data
    if not isinstance(entries, list):
        raise AllowlistError(f"{path}: expected an 'entries' list")

    validated: list[dict[str, Any]] = []
    seen_packages: set[str] = set()
    for index, raw in enumerate(entries):
        if not isinstance(raw, dict):
            raise AllowlistError(f"{path}: entry {index} is not an object")
        package = raw.get("package")
        reason = raw.get("reason")
        task = raw.get("task")
        if not isinstance(package, str) or not package.strip():
            raise AllowlistError(f"{path}: entry {index} is missing a non-empty 'package'")
        if not isinstance(reason, str) or not reason.strip():
            raise AllowlistError(
                f"{path}: allowlist entry '{package}' is missing a required 'reason'"
            )
        if not isinstance(task, str) or not task.strip():
            raise AllowlistError(
                f"{path}: allowlist entry '{package}' is missing a required tracking 'task'"
            )
        if package in seen_packages:
            raise AllowlistError(f"{path}: duplicate allowlist entry for '{package}'")
        seen_packages.add(package)
        validated.append({"package": package, "reason": reason.strip(), "task": task.strip()})
    return validated


def analyze(root: Path = ROOT, allowlist_path: Path = DEFAULT_ALLOWLIST) -> WiringReport:
    module = read_module_path(root)
    files = discover_go_files(root)
    prod_by_dir_abs, _test_by_dir_abs = classify_files(files)
    prod_by_dir = relativize(prod_by_dir_abs, root)

    graph = build_graph(prod_by_dir, module)
    entry_points = find_entry_points(prod_by_dir)
    reachable = reachable_from(graph, entry_points)

    implemented_internal = {d for d in prod_by_dir if d.startswith("internal/")}
    unreachable = implemented_internal - reachable

    allowlist_entries = load_allowlist(allowlist_path)

    allowlisted_active: dict[str, dict[str, Any]] = {}
    stale_allowlist: dict[str, dict[str, Any]] = {}
    unknown_allowlist: dict[str, dict[str, Any]] = {}
    for entry in allowlist_entries:
        package = entry["package"]
        if package not in implemented_internal:
            unknown_allowlist[package] = entry
        elif package in reachable:
            stale_allowlist[package] = entry
        else:
            allowlisted_active[package] = entry

    return WiringReport(
        module=module,
        entry_points=sorted(entry_points),
        implemented_internal=implemented_internal,
        reachable=reachable,
        unreachable=unreachable,
        allowlist_entries=allowlist_entries,
        allowlisted_active=allowlisted_active,
        stale_allowlist=stale_allowlist,
        unknown_allowlist=unknown_allowlist,
    )


def format_report(report: WiringReport) -> str:
    lines: list[str] = []
    lines.append(f"module: {report.module}")
    lines.append(f"entry points: {', '.join(report.entry_points)}")
    reachable_internal = report.reachable & report.implemented_internal
    lines.append(
        f"reachable internal/* packages: {len(reachable_internal)} of {len(report.implemented_internal)}"
    )
    lines.append(f"allowlisted (tracked debt, still unreachable): {len(report.allowlisted_active)}")
    for package in sorted(report.allowlisted_active):
        entry = report.allowlisted_active[package]
        lines.append(f"  - {package} [{entry['task']}] {entry['reason']}")
    lines.append(f"unreachable and NOT allowlisted: {len(report.failing)}")
    for package in sorted(report.failing):
        lines.append(f"  - {package}")
    lines.append(f"stale allowlist entries (now reachable — remove from allowlist): {len(report.stale_allowlist)}")
    for package in sorted(report.stale_allowlist):
        lines.append(f"  - {package}")
    if report.unknown_allowlist:
        lines.append(f"allowlist entries for unknown/nonexistent packages: {len(report.unknown_allowlist)}")
        for package in sorted(report.unknown_allowlist):
            lines.append(f"  - {package}")
    return "\n".join(lines)


def main() -> int:
    try:
        report = analyze(ROOT, DEFAULT_ALLOWLIST)
    except AllowlistError as error:
        print(f"WIRING CHECK: ERROR — {error}", file=sys.stderr)
        return 2

    body = format_report(report)
    if report.ok:
        print(body)
        print(f"WIRING CHECK: PASS ({len(report.reachable & report.implemented_internal)} reachable, "
              f"{len(report.allowlisted_active)} allowlisted)")
        return 0

    print(body, file=sys.stderr)
    print(
        f"WIRING CHECK: FAIL ({len(report.failing)} unreachable, "
        f"{len(report.stale_allowlist)} stale, {len(report.unknown_allowlist)} unknown)",
        file=sys.stderr,
    )
    return 1


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