feat(plugin): initial scaffold — plugin loader entry + AST + CI workflows (P1.B/C/D)
THE single repo holding every Svrnty modification to nesquena/hermes-webui per
the SVRNTY-HERMES Plugin Protocol PRD (hermes/docs/SVRNTY-PLUGIN-PROTOCOL.md).
This scaffold is the empty vessel; Phase 2 migrates the existing fork
modifications (STT, vault status, brand skin) into it.
Contents:
plugin.py register(api) entry; reads 6-method
contract; wires static + routes
svrnty_hermes_webui_plugin/ package wrapper for pip install
manifest.yaml plugin name · version · upstream pin
pyproject.toml pip-installable
routes/__init__.py placeholder until Phase 2 migration
static/ placeholder for brand skin migration
CONNECTION-MAP.md AST-generated; 4 public API calls, 0 forced
scripts/ast-connection-map.py AST walker; modes: regen · --check · --diff
tests/{unit,integration,evals}/ skeleton for Phase 3 eval suite
.github/workflows/
plugin-tests.yml push: unit + integration + map check
connection-map-check.yml PR: regen + diff vs committed
upstream-drift.yml daily cron: detect new upstream tags +
run matrix (activates when Gitea runner
registered on Svrnty infra)
Karpathy 4 rules in CLAUDE.md; every subagent inherits them via the workspace
contract.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Executable
+262
@@ -0,0 +1,262 @@
|
||||
#!/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
|
||||
from datetime import datetime, timezone
|
||||
|
||||
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",
|
||||
}
|
||||
|
||||
|
||||
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)
|
||||
now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
out = [
|
||||
"# CONNECTION MAP — svrnty-hermes-webui-plugin → nesquena/hermes-webui",
|
||||
"",
|
||||
f"**Generated:** {now} ",
|
||||
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()
|
||||
Reference in New Issue
Block a user