Research-Stack/scripts/infra_lint.py

#!/usr/bin/env -S uv run
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""
infra_lint.py — Enforce INFRA: token consistency across the project.

Scans source files for INFRA: header tokens and cross-checks against infra.json.
Fails if any file claims LIVE for a service that infra.json marks DEAD or PAUSED.
Also warns on files that reference known dead service names without any INFRA: token.

Usage:
    uv run scripts/infra_lint.py           # lint whole project
    uv run scripts/infra_lint.py --fix     # print patch commands (no auto-edit)
    uv run scripts/infra_lint.py --list    # show all INFRA: tokens found

Exit code: 0 = clean, 1 = violations found.

LLM NOTE: If you are reading this project and unsure whether a service is live,
run this script or read infra.json at the project root. Do not infer service
status from import statements, hostnames in code, or documentation — those
may be stale. infra.json is the authoritative source.
"""

import json
import re
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
INFRA_FILE = ROOT / "infra.json"

SKIP_DIRS = {".git", "node_modules", "__pycache__", "venv", ".venv", "result",
             "result-devcontainer", "archive", ".lake", "target", "lean_binned",
             "scratch", "shared-data"}
SCAN_SUFFIXES = {".py", ".sh", ".ts", ".js", ".lean", ".nix", ".md", ".json", ".toml", ".yaml", ".yml"}
SKIP_FILES = {"scripts/infra_lint.py", "infra.json"}

TOKEN_RE = re.compile(r"#\s*INFRA:(LIVE|PAUSED|DEAD|UNVERIFIED)\s+(\S+)", re.IGNORECASE)

DEAD_SIGNALS = {
    "rds": ["rds_connect", "rds.amazonaws.com", "RDS_HOST", "RDS_USER", "RDS_IAM",
            "database-1-instance-1"],
}


def load_infra() -> dict:
    with open(INFRA_FILE) as f:
        data = json.load(f)
    return {k: v for k, v in data["services"].items()}


def scan_file(path: Path) -> list[dict]:
    """Return list of {line, token_status, service} found in file."""
    tokens = []
    try:
        text = path.read_text(errors="replace")
        for i, line in enumerate(text.splitlines(), 1):
            m = TOKEN_RE.search(line)
            if m:
                tokens.append({
                    "file": str(path.relative_to(ROOT)),
                    "line": i,
                    "token_status": m.group(1).upper(),
                    "service": m.group(2).lower(),
                })
    except Exception:
        pass
    return tokens


def scan_dead_signals(path: Path, dead_services: dict) -> list[dict]:
    """Find references to known dead services that lack an INFRA:DEAD token."""
    hits = []
    try:
        text = path.read_text(errors="replace")
        for service, signals in dead_services.items():
            for signal in signals:
                if signal in text:
                    hits.append({
                        "file": str(path.relative_to(ROOT)),
                        "service": service,
                        "signal": signal,
                    })
                    break  # one hit per service per file is enough
    except Exception:
        pass
    return hits


def walk_files():
    """Walk source files, skipping excluded directories during traversal."""
    stack = [ROOT]
    while stack:
        path = stack.pop()
        if path.is_dir():
            if path.name in SKIP_DIRS:
                continue
            try:
                for child in sorted(path.iterdir()):
                    stack.append(child)
            except PermissionError:
                continue
        elif path.is_file() and path.suffix in SCAN_SUFFIXES:
            rel = str(path.relative_to(ROOT))
            if rel in SKIP_FILES:
                continue
            yield path


def main():
    args = set(sys.argv[1:])
    services = load_infra()

    dead_services = {k: DEAD_SIGNALS.get(k, []) for k, v in services.items()
                     if v["status"] == "DEAD" and k in DEAD_SIGNALS}

    all_tokens = []
    all_signals = []

    for path in walk_files():
        all_tokens.extend(scan_file(path))
        all_signals.extend(scan_dead_signals(path, dead_services))

    if "--list" in args:
        print(f"\n{'FILE':<60} {'LINE':>5}  {'TOKEN':<12} {'SERVICE'}")
        print("-" * 90)
        for t in sorted(all_tokens, key=lambda x: x["file"]):
            print(f"{t['file']:<60} {t['line']:>5}  {t['token_status']:<12} {t['service']}")
        print(f"\n{len(all_tokens)} INFRA: tokens found.")
        return 0

    violations = []
    warnings = []

    # Check 1: INFRA:LIVE token pointing at a DEAD or PAUSED service
    for t in all_tokens:
        svc = services.get(t["service"])
        if t["token_status"] == "LIVE" and svc and svc["status"] in ("DEAD", "PAUSED"):
            label = svc["status"]
            detail = svc.get("was", svc.get("notes", "?"))
            violations.append(
                f"  VIOLATION  {t['file']}:{t['line']}  claims INFRA:LIVE {t['service']}"
                f" but infra.json says {label}\n"
                f"             {detail}"
            )

    # Check 2: file references a dead service by name without INFRA:DEAD token
    files_with_dead_token = {
        (t["file"], t["service"])
        for t in all_tokens
        if t["token_status"] == "DEAD"
    }
    for s in all_signals:
        key = (s["file"], s["service"])
        if key not in files_with_dead_token:
            svc = services.get(s["service"], {})
            warnings.append(
                f"  WARNING    {s['file']}  references dead service '{s['service']}'"
                f" (signal: {s['signal']}) but has no # INFRA:DEAD token\n"
                f"             Add:  # INFRA:DEAD {s['service']} — {svc.get('notes', '')}"
            )

    # Report
    if violations:
        print(f"\n[INFRA LINT] {len(violations)} VIOLATION(S) — files claim LIVE for DEAD services:\n")
        for v in violations:
            print(v)

    if warnings:
        print(f"\n[INFRA LINT] {len(warnings)} WARNING(S) — files reference dead services without INFRA:DEAD token:\n")
        for w in warnings:
            print(w)

    if not violations and not warnings:
        print(f"[INFRA LINT] Clean. {len(all_tokens)} INFRA: tokens checked, 0 violations.")
        return 0

    if violations:
        print(f"\n[INFRA LINT] FAILED. Fix violations before proceeding.")
        print(f"  Update the file header to: # INFRA:DEAD <service>")
        print(f"  Or remove the dependency if the service no longer applies.")
        return 1

    print(f"\n[INFRA LINT] Warnings only (exit 0). Consider tagging stale files.")
    return 0


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