svrnty-hermes-webui-plugin/scripts/ast-connection-map.py

#!/usr/bin/env python3
"""ast-connection-map.py — regenerate CONNECTION-MAP.md from plugin source.

Walks plugin.py + routes/*.py via the Python AST. Categorizes every reference
that crosses into hermes-webui internals as:

  - PUBLIC API:   calls on the `api` parameter exposed by the loader hook
                  (api.register_route, api.register_static, api.inject_script,
                  api.inject_stylesheet, api.config_get, api.logger).
  - FORCED INTERNAL: any `import hermes_webui...` or `from hermes_webui...`
                  not in the public set. Each row needs a `# CONNECTION:` source
                  comment justifying the escape hatch + naming the risk.

Also scans static/*.js for hardcoded /api/* URL references → "frontend
dependencies" table. (DOM selectors not yet scanned — v1.1.)

Modes:
  python ast-connection-map.py             regenerate CONNECTION-MAP.md
  python ast-connection-map.py --check     exit 1 if regen != committed (CI)
  python ast-connection-map.py --diff REF  show what changed since REF

Lives at the plugin repo root via scripts/ — run from anywhere; uses Path
relative to __file__.
"""
import argparse
import ast
import re
import subprocess
import sys
from pathlib import Path

REPO = Path(__file__).resolve().parent.parent
MAP_PATH = REPO / "CONNECTION-MAP.md"
PUBLIC_API = {
    "register_route", "register_static", "inject_script",
    "inject_stylesheet", "config_get", "logger",
    "register_audio_attachment_processor",
}


def _python_sources():
    """Every Python source file in the plugin (plugin.py + routes/*.py)."""
    files = [REPO / "plugin.py"]
    routes_dir = REPO / "routes"
    if routes_dir.exists():
        files.extend(sorted(p for p in routes_dir.rglob("*.py") if "__pycache__" not in p.parts))
    return [f for f in files if f.exists()]


def _walk_python(path):
    """Return (public_calls, forced_internals) for one file.

    public_calls   : list[(lineno, method, snippet)]   — api.<method>(...) calls
    forced_internals: list[(lineno, target, justification)] — hermes_webui imports
    """
    src = path.read_text(encoding="utf-8")
    tree = ast.parse(src)
    lines = src.splitlines()

    public = []
    internal = []

    for node in ast.walk(tree):
        # api.METHOD(...) calls
        if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute):
            if isinstance(node.func.value, ast.Name) and node.func.value.id == "api":
                method = node.func.attr
                if method in PUBLIC_API:
                    snippet = lines[node.lineno - 1].strip()[:90]
                    public.append((node.lineno, method, snippet))

        # Imports targeting hermes-webui internals (anything not in PUBLIC_API)
        if isinstance(node, (ast.Import, ast.ImportFrom)):
            mods = []
            if isinstance(node, ast.Import):
                mods = [a.name for a in node.names]
            else:
                if node.module:
                    mods = [node.module]
            for m in mods:
                if m.startswith("hermes_webui") or m.startswith("api.") or m == "api":
                    # api is the loader param, not a module import — skip if local.
                    # But anything else under hermes_webui or api.* = forced internal.
                    if m == "api":
                        continue
                    # Look for nearby `# CONNECTION: <text>` comment for justification
                    just = _comment_for(lines, node.lineno)
                    internal.append((node.lineno, m, just))

    return public, internal


def _comment_for(lines, lineno):
    """Find a `# CONNECTION: <text>` comment within 2 lines above or on same line."""
    pat = re.compile(r"#\s*CONNECTION:\s*(.+)")
    for offset in (0, -1, -2):
        idx = lineno - 1 + offset
        if 0 <= idx < len(lines):
            m = pat.search(lines[idx])
            if m:
                return m.group(1).strip()
    return "(no justification — add `# CONNECTION: <reason>` above the import)"


def _scan_frontend():
    """Scan static/*.js for /api/* URL references. Returns list[(file, lineno, url)]."""
    rows = []
    static = REPO / "static"
    if not static.exists():
        return rows
    pat = re.compile(r"['\"`](/api/[a-zA-Z0-9_/-]+)['\"`]")
    for js in static.rglob("*.js"):
        for i, line in enumerate(js.read_text(encoding="utf-8", errors="ignore").splitlines(), 1):
            for m in pat.finditer(line):
                rows.append((js.relative_to(REPO), i, m.group(1)))
    return rows


def _upstream_version():
    """Best-effort upstream version pin from manifest.yaml."""
    mf = REPO / "manifest.yaml"
    if not mf.exists():
        return "unknown"
    for line in mf.read_text().splitlines():
        m = re.match(r"\s*current_local:\s*(\S+)", line)
        if m:
            return m.group(1)
    return "unknown"


def _plugin_version():
    mf = REPO / "manifest.yaml"
    if not mf.exists():
        return "unknown"
    for line in mf.read_text().splitlines():
        m = re.match(r"\s*plugin_version:\s*(\S+)", line)
        if m:
            return m.group(1)
    return "unknown"


def generate():
    """Build CONNECTION-MAP.md content as a string."""
    public_rows = []
    internal_rows = []
    for f in _python_sources():
        rel = f.relative_to(REPO)
        pub, fi = _walk_python(f)
        for ln, method, snippet in pub:
            public_rows.append((str(rel), ln, f"api.{method}", snippet))
        for ln, target, just in fi:
            internal_rows.append((str(rel), ln, target, just))

    frontend_rows = _scan_frontend()

    total = len(public_rows) + len(internal_rows) + len(frontend_rows)
    out = [
        "# CONNECTION MAP — svrnty-hermes-webui-plugin → nesquena/hermes-webui",
        "",
        f"**Upstream version:** {_upstream_version()}  ",
        f"**Plugin version:** {_plugin_version()}  ",
        f"**Total dependencies:** {total} ({len(public_rows)} public API · "
        f"{len(internal_rows)} forced internal · {len(frontend_rows)} frontend)",
        "",
        "> **Auto-generated by `scripts/ast-connection-map.py`. Do not hand-edit.**",
        "> To change a justification, edit the `# CONNECTION:` comment above the",
        "> relevant import and re-run the script.",
        "",
        "---",
        "",
        "## Public API dependencies (via loader-provided `api`)",
        "",
    ]
    if public_rows:
        out.append("| Plugin location | Upstream symbol | Snippet |")
        out.append("|---|---|---|")
        for rel, ln, sym, snip in public_rows:
            out.append(f"| `{rel}:{ln}` | `{sym}` | `{snip}` |")
    else:
        out.append("_None yet (plugin scaffold pre-Phase-2)._")
    out.append("")
    out.append("---")
    out.append("")
    out.append("## Forced internal dependencies (Rule 2 escape hatch)")
    out.append("")
    out.append("Each row requires a `# CONNECTION: <reason>` comment in source.")
    out.append("")
    if internal_rows:
        out.append("| Plugin location | Upstream symbol | Justification |")
        out.append("|---|---|---|")
        for rel, ln, sym, just in internal_rows:
            out.append(f"| `{rel}:{ln}` | `{sym}` | {just} |")
    else:
        out.append("_None. Plugin uses only the public API._ ✓")
    out.append("")
    out.append("---")
    out.append("")
    out.append("## Frontend dependencies (static/*.js → /api/* URLs)")
    out.append("")
    if frontend_rows:
        out.append("| File | Line | URL |")
        out.append("|---|---|---|")
        for f, ln, url in frontend_rows:
            out.append(f"| `{f}` | {ln} | `{url}` |")
    else:
        out.append("_None yet._")
    out.append("")
    return "\n".join(out) + "\n"


def main():
    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    ap.add_argument("--check", action="store_true",
                    help="Exit 1 if regenerated content differs from committed CONNECTION-MAP.md")
    ap.add_argument("--diff", metavar="REF",
                    help="Show diff vs another git ref (read-only)")
    args = ap.parse_args()

    new_content = generate()

    if args.check:
        old = MAP_PATH.read_text(encoding="utf-8") if MAP_PATH.exists() else ""
        if old == new_content:
            print(f"OK — CONNECTION-MAP.md is fresh ({MAP_PATH.name})")
            sys.exit(0)
        print("DRIFT — CONNECTION-MAP.md is stale.", file=sys.stderr)
        print("Run: python3 scripts/ast-connection-map.py", file=sys.stderr)
        # Quick line-diff for the CI log
        import difflib
        diff = difflib.unified_diff(old.splitlines(keepends=True),
                                    new_content.splitlines(keepends=True),
                                    fromfile="CONNECTION-MAP.md (committed)",
                                    tofile="CONNECTION-MAP.md (regenerated)",
                                    n=2)
        sys.stderr.writelines(diff)
        sys.exit(1)

    if args.diff:
        try:
            old = subprocess.check_output(
                ["git", "show", f"{args.diff}:CONNECTION-MAP.md"],
                cwd=REPO, stderr=subprocess.DEVNULL, text=True)
        except subprocess.CalledProcessError:
            print(f"could not read CONNECTION-MAP.md at ref {args.diff}", file=sys.stderr)
            sys.exit(2)
        import difflib
        diff = difflib.unified_diff(old.splitlines(keepends=True),
                                    new_content.splitlines(keepends=True),
                                    fromfile=f"CONNECTION-MAP.md @{args.diff}",
                                    tofile="CONNECTION-MAP.md (working)",
                                    n=3)
        sys.stdout.writelines(diff)
        sys.exit(0)

    MAP_PATH.write_text(new_content, encoding="utf-8")
    print(f"wrote {MAP_PATH.name} ({len(new_content)} bytes)")


if __name__ == "__main__":
    main()