Initial commit: codemap skill (architecture map + per-module code-quality audit)

scripts/apply_audit.py

@@ -0,0 +1,75 @@
+#!/usr/bin/env python3
+"""apply_audit.py — merge one subagent's audit result into modules.json.
+
+A module audit is produced by an INDEPENDENT subagent (see reference/STANDARDS.md)
+and returned as a small JSON object:
+
+    {"score": 72, "grade": "C",
+     "tags": ["duplication","legacy"],
+     "findings": [{"sev":"HIGH","loc":"path/file.py:120","text":"..."}, ...]}
+
+This script writes that result onto the module and stamps `auditedHash` =
+current `contentHash` (so scan.py will treat the module as fresh until its code
+changes again), plus `auditedAt` / `auditedRev`. Run scan.py --write FIRST so the
+current contentHash is present.
+
+Accepts the result inline (--json '...') or from a file (--json-file path).
+Stdlib only.
+"""
+import argparse, datetime, json, sys
+
+VALID_GRADES = {"A", "B", "C", "D", "F"}
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--state", required=True)
+    ap.add_argument("--id", required=True, help="module id to update")
+    ap.add_argument("--json", help="audit result as an inline JSON string")
+    ap.add_argument("--json-file", help="audit result JSON file")
+    ap.add_argument("--rev", default="", help="git rev being audited (optional)")
+    args = ap.parse_args()
+
+    if args.json_file:
+        result = json.load(open(args.json_file, encoding="utf-8"))
+    elif args.json:
+        result = json.loads(args.json)
+    else:
+        result = json.load(sys.stdin)
+
+    state = json.load(open(args.state, encoding="utf-8"))
+    mod = next((m for m in state.get("modules", []) if m["id"] == args.id), None)
+    if mod is None:
+        sys.exit(f"module id not found: {args.id}")
+
+    score = int(result["score"])
+    grade = str(result["grade"]).strip().upper()[:1]
+    if grade not in VALID_GRADES:
+        sys.exit(f"invalid grade: {result['grade']}")
+    if not (0 <= score <= 100):
+        sys.exit(f"score out of range: {score}")
+
+    findings = []
+    for f in result.get("findings", []):
+        sev = str(f.get("sev", "LOW")).upper()
+        if sev not in {"HIGH", "MED", "LOW"}:
+            sev = "LOW"
+        findings.append({"sev": sev, "loc": str(f.get("loc", "")),
+                         "text": str(f.get("text", ""))})
+
+    mod["score"] = score
+    mod["grade"] = grade
+    mod["tags"] = list(result.get("tags", [])) or ["clean"]
+    mod["findings"] = findings
+    mod["auditedHash"] = mod.get("contentHash", "")
+    mod["auditedAt"] = datetime.datetime.now().strftime("%Y-%m-%d")
+    mod["auditedRev"] = args.rev
+
+    json.dump(state, open(args.state, "w", encoding="utf-8"),
+              ensure_ascii=False, indent=1)
+    print(f"applied: {args.id}  score={score} grade={grade} "
+          f"findings={len(findings)} tags={mod['tags']}")
+
+
+if __name__ == "__main__":
+    main()

scripts/query.py

@@ -0,0 +1,122 @@
+#!/usr/bin/env python3
+"""query.py — token-cheap module selector over modules.json.
+
+An agent uses this to find exactly the modules it must act on WITHOUT reading the
+whole state file. Filter by grade/score level, by smell tag, by finding severity,
+by band/coupling, or by staleness; emit just ids, paths, a compact table, the
+findings, or filtered JSON.
+
+Examples:
+  # all C-and-below modules → just their ids (pipe into a fix loop)
+  python query.py --state s.json --max-grade C --format ids
+  # modules with a dual-format problem, worst first, as a table
+  python query.py --state s.json --tag dual-format
+  # the file globs to read for every module that has a HIGH finding
+  python query.py --state s.json --sev HIGH --format paths
+  # the actual findings to fix for one tag, as text
+  python query.py --state s.json --tag glue --format findings
+  # what needs re-auditing (stale or never scored)
+  python query.py --state s.json --needs-audit --format ids
+
+Filters combine with AND. --tag may be repeated (ANY by default, --match-all for AND).
+Stdlib only.
+"""
+import argparse, json, sys
+
+# include a module if its score is strictly below this bound (grade and worse).
+GRADE_BOUND = {"A": 101, "B": 90, "C": 75, "D": 60, "F": 40}
+
+
+def main():
+    ap = argparse.ArgumentParser(description="filter modules.json for agents")
+    ap.add_argument("--state", required=True)
+    ap.add_argument("--max-grade", choices=list(GRADE_BOUND),
+                    help="include this grade AND worse (e.g. C → C,D,F)")
+    ap.add_argument("--min-score", type=int)
+    ap.add_argument("--max-score", type=int)
+    ap.add_argument("--tag", action="append", default=[],
+                    help="smell tag; repeatable (ANY unless --match-all)")
+    ap.add_argument("--match-all", action="store_true", help="require ALL --tag")
+    ap.add_argument("--sev", choices=["HIGH", "MED", "LOW"],
+                    help="has at least one finding of this severity")
+    ap.add_argument("--band")
+    ap.add_argument("--coupling", choices=["low", "med", "high", "core"])
+    ap.add_argument("--needs-audit", action="store_true",
+                    help="stale (contentHash != auditedHash) or never scored")
+    ap.add_argument("--sort", choices=["score", "loc", "label", "band"], default="score")
+    ap.add_argument("--desc", action="store_true", help="sort descending")
+    ap.add_argument("--limit", type=int)
+    ap.add_argument("--format", choices=["table", "ids", "paths", "findings", "json", "count"],
+                    default="table")
+    args = ap.parse_args()
+
+    state = json.load(open(args.state, encoding="utf-8"))
+    mods = state.get("modules", [])
+
+    bound = GRADE_BOUND[args.max_grade] if args.max_grade else None
+    tags = set(args.tag)
+
+    def keep(m):
+        s = m.get("score")
+        if bound is not None and not (s is not None and s < bound):
+            return False
+        if args.min_score is not None and not (s is not None and s >= args.min_score):
+            return False
+        if args.max_score is not None and not (s is not None and s <= args.max_score):
+            return False
+        if tags:
+            mt = set(m.get("tags") or [])
+            if args.match_all and not tags <= mt:
+                return False
+            if not args.match_all and not (tags & mt):
+                return False
+        if args.sev and not any(f.get("sev") == args.sev for f in (m.get("findings") or [])):
+            return False
+        if args.band and m.get("band") != args.band:
+            return False
+        if args.coupling and m.get("coupling") != args.coupling:
+            return False
+        if args.needs_audit:
+            stale = (m.get("score") is None or not m.get("auditedHash")
+                     or m.get("auditedHash") != m.get("contentHash"))
+            if not stale:
+                return False
+        return True
+
+    sel = [m for m in mods if keep(m)]
+    key = {"score": lambda m: (m.get("score") if m.get("score") is not None else 999),
+           "loc": lambda m: m.get("loc") or 0,
+           "label": lambda m: m.get("label", ""),
+           "band": lambda m: m.get("band", "")}[args.sort]
+    sel.sort(key=key, reverse=args.desc)
+    if args.limit:
+        sel = sel[:args.limit]
+
+    if args.format == "count":
+        print(len(sel))
+    elif args.format == "ids":
+        print(" ".join(m["id"] for m in sel))
+    elif args.format == "paths":
+        seen, out = set(), []
+        for m in sel:
+            for p in (m.get("paths") or []):
+                if p not in seen:
+                    seen.add(p); out.append(p)
+        print("\n".join(out))
+    elif args.format == "findings":
+        for m in sel:
+            for f in (m.get("findings") or []):
+                print(f"{m['id']}\t{f.get('sev')}\t{f.get('loc','')}\t{f.get('text','')}")
+    elif args.format == "json":
+        print(json.dumps(sel, ensure_ascii=False, indent=1))
+    else:  # table
+        print(f"# {len(sel)} of {len(mods)} modules match", file=sys.stderr)
+        for m in sel:
+            s = m.get("score")
+            sg = f"{s if s is not None else '--'}/{m.get('grade','-')}"
+            tg = ",".join(t for t in (m.get("tags") or []) if t != "clean") or "-"
+            print(f"{m['id']:<22} {sg:>7} {str(m.get('loc') or 0):>6}  {m.get('band',''):<12} {tg}")
+
+
+if __name__ == "__main__":
+    main()

scripts/render.py

@@ -0,0 +1,152 @@
+#!/usr/bin/env python3
+"""render.py — regenerate architecture-map.html + architecture-audit.md from modules.json.
+
+modules.json is the single source of truth. The HTML and MD are pure projections
+of it and must never be hand-edited. Run scan.py --write before rendering so LoC /
+content hashes are current.
+
+Usage:
+  python render.py --state modules.json \
+      --template assets/template.html \
+      --out-html architecture-map.html --out-md architecture-audit.md
+
+Stdlib only.
+"""
+import argparse, json, os
+
+
+def health_color(s):
+    if s is None: return "#6b7280"
+    if s < 50:  return "#e0524b"
+    if s < 65:  return "#e0804a"
+    if s < 75:  return "#d9a441"
+    if s < 85:  return "#8f969d"
+    return "#5d6b63"
+
+
+def band_order(state):
+    return [b["id"] for b in state.get("bands", []) if not b.get("wire")]
+
+
+def render_html(state, template):
+    data = {
+        "meta": state.get("meta", {}),
+        "bands": state.get("bands", []),
+        "spine": state.get("spine", []),
+        "reportThemes": state.get("reportThemes", []),
+        "modules": [
+            {k: m.get(k) for k in (
+                "id", "label", "band", "path", "desc", "coupling", "deps",
+                "loc", "score", "grade", "tags", "findings")}
+            for m in state.get("modules", [])
+        ],
+    }
+    blob = json.dumps(data, ensure_ascii=False).replace("</", "<\\/")
+    return template.replace("__ARCH_DATA__", blob)
+
+
+def render_md(state):
+    meta = state.get("meta", {})
+    mods = [m for m in state.get("modules", []) if m.get("score") is not None]
+    bands = {b["id"]: b for b in state.get("bands", [])}
+    out = []
+    proj = meta.get("project", "Project")
+    out.append("<!--")
+    out.append(f"  This file:        {meta.get('mdPath', 'architecture-audit.md')}   (written report)")
+    out.append(f"  Interactive map:  {meta.get('htmlPath', 'architecture-map.html')}")
+    out.append("-->\n")
+    out.append(f"# {proj} — Functional Module Quality Audit\n")
+    out.append(f"> **Interactive view:** [`{meta.get('htmlPath','architecture-map.html')}`]"
+               f"({os.path.basename(meta.get('htmlPath','architecture-map.html'))}) — "
+               "per-module scores, findings, LoC, and the dependency graph. This file is the written report.\n")
+    gen = meta.get("generatedAt", "")
+    loc_line = meta.get("locLine") or (
+        f"{meta.get('tracked_loc','?')} tracked LoC across {meta.get('tracked_files','?')} files")
+    out.append(f"**Generated:** {gen} · **Modules:** {len(mods)} · **Size:** {loc_line}\n")
+
+    # per-layer averages
+    out.append("## Health by layer\n")
+    out.append("| Layer | Modules | Avg score |")
+    out.append("|---|--:|--:|")
+    for b in state.get("bands", []):
+        if b.get("wire"):
+            continue
+        grp = [m for m in mods if m["band"] == b["id"]]
+        if not grp:
+            continue
+        avg = round(sum(m["score"] for m in grp) / len(grp))
+        out.append(f"| {b.get('t', b['id'])} | {len(grp)} | {avg} |")
+    out.append("")
+
+    # per-module LoC + score, grouped by band, sorted by loc desc
+    out.append("## Per-module lines of code & score\n")
+    out.append("_LoC is the representative file/folder per module; folder-level modules overlap "
+               "and are not additive._\n")
+    for b in state.get("bands", []):
+        if b.get("wire"):
+            continue
+        grp = sorted([m for m in mods if m["band"] == b["id"]],
+                     key=lambda m: -(m.get("loc") or 0))
+        if not grp:
+            continue
+        out.append(f"### {b.get('t', b['id'])}\n")
+        out.append("| Module | LoC | Score | Tags |")
+        out.append("|---|--:|:--|:--|")
+        for m in grp:
+            tags = ", ".join(t for t in (m.get("tags") or []) if t != "clean") or "—"
+            loc = f"{m.get('loc',0):,}"
+            out.append(f"| {m['label']} | {loc} | {m['score']} {m['grade']} | {tags} |")
+        out.append("")
+
+    # worst offenders
+    out.append("## Worst offenders\n")
+    worst = sorted(mods, key=lambda m: m["score"])[:10]
+    for m in worst:
+        fnd = m.get("findings") or []
+        top = next((f for f in fnd if f["sev"] == "HIGH"), fnd[0] if fnd else None)
+        ev = f" — {top['loc']}: {top['text']}" if top else ""
+        out.append(f"- **{m['label']} ({m['score']}/{m['grade']})**{ev}")
+    out.append("")
+
+    # all findings, by severity
+    out.append("## All findings\n")
+    for sev in ("HIGH", "MED", "LOW"):
+        rows = [(m, f) for m in mods for f in (m.get("findings") or []) if f["sev"] == sev]
+        if not rows:
+            continue
+        out.append(f"### {sev} ({len(rows)})\n")
+        for m, f in rows:
+            out.append(f"- **{m['label']}** · `{f['loc']}` — {f['text']}")
+        out.append("")
+
+    # cross-cutting themes
+    themes = state.get("reportThemes", [])
+    if themes:
+        out.append("## Cross-cutting themes\n")
+        for h, body in themes:
+            out.append(f"- **{h}.** {body}")
+        out.append("")
+
+    return "\n".join(out) + "\n"
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--state", required=True)
+    ap.add_argument("--template", required=True)
+    ap.add_argument("--out-html", required=True)
+    ap.add_argument("--out-md", required=True)
+    args = ap.parse_args()
+
+    state = json.load(open(args.state, encoding="utf-8"))
+    template = open(args.template, encoding="utf-8").read()
+
+    open(args.out_html, "w", encoding="utf-8").write(render_html(state, template))
+    open(args.out_md, "w", encoding="utf-8").write(render_md(state))
+    n = len(state.get("modules", []))
+    scored = sum(1 for m in state.get("modules", []) if m.get("score") is not None)
+    print(f"rendered {n} modules ({scored} scored) -> {args.out_html} + {args.out_md}")
+
+
+if __name__ == "__main__":
+    main()

scripts/scan.py

@@ -0,0 +1,129 @@
+#!/usr/bin/env python3
+"""scan.py — compute per-module LoC + content hash and report staleness.
+
+The architecture state file (modules.json) is the source of truth. Each module
+declares `paths` (a list of globs, relative to the project root). This script:
+
+  * resolves each module's files, counts lines (LoC) and computes a content hash
+    (sha256 over sorted "relpath:sha256(bytes)" pairs) that is stable across
+    checkouts (depends on content, not mtime);
+  * compares the fresh content hash to `auditedHash` (the hash captured the last
+    time the module was audited) to classify each module as:
+        fresh      — code unchanged since last audit
+        stale      — code changed since last audit (needs re-audit)
+        unaudited  — never audited (no auditedHash / no score)
+        empty      — paths match no files (likely deleted / moved)
+  * with --write, writes the fresh `loc` and `contentHash` back into the state.
+
+Output (stdout): a JSON report the orchestrator uses to decide what to re-audit.
+Stdlib only.
+"""
+import argparse, glob, hashlib, json, os, sys
+
+DEFAULT_EXCLUDES = [
+    "__pycache__", "/node_modules/", "/dist/", "/build/", "/.git/",
+    "/vendor/", ".min.js", ".min.css", "/.venv/", "/venv/",
+    ".pytest", "/coverage/", ".map",
+    # tests are the regression net, not part of a module's audit scope:
+    "/tests/", "/test/", "/__tests__/", ".test.", ".spec.", "_test.py",
+    "conftest.py", ".stories.",
+]
+
+
+def iter_files(root, patterns, excludes):
+    seen = set()
+    for pat in patterns:
+        for p in glob.glob(os.path.join(root, pat), recursive=True):
+            if not os.path.isfile(p):
+                continue
+            rp = os.path.relpath(p, root).replace("\\", "/")
+            low = "/" + rp.lower()
+            if any(e in low for e in excludes):
+                continue
+            if rp in seen:
+                continue
+            seen.add(rp)
+            yield p, rp
+
+
+def module_stats(root, module, excludes):
+    pats = module.get("paths") or []
+    if isinstance(pats, str):
+        pats = [pats]
+    excl = list(excludes) + list(module.get("exclude", []))
+    loc = 0
+    parts = []
+    nfiles = 0
+    for p, rp in sorted(iter_files(root, pats, excl), key=lambda x: x[1]):
+        try:
+            data = open(p, "rb").read()
+        except OSError:
+            continue
+        loc += data.count(b"\n") + (1 if data and not data.endswith(b"\n") else 0)
+        parts.append(rp + ":" + hashlib.sha256(data).hexdigest())
+        nfiles += 1
+    chash = hashlib.sha256("\n".join(parts).encode()).hexdigest() if parts else ""
+    return loc, chash, nfiles
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--root", default=".", help="project root")
+    ap.add_argument("--state", required=True, help="path to modules.json")
+    ap.add_argument("--write", action="store_true",
+                    help="write fresh loc + contentHash back into the state")
+    args = ap.parse_args()
+
+    state = json.load(open(args.state, encoding="utf-8"))
+    excludes = state.get("excludes", DEFAULT_EXCLUDES)
+    root = os.path.abspath(args.root)
+
+    buckets = {"fresh": [], "stale": [], "unaudited": [], "empty": []}
+    union_files = {}
+    for m in state.get("modules", []):
+        loc, chash, nfiles = module_stats(root, m, excludes)
+        m["loc"] = loc
+        m["contentHash"] = chash
+        # union for an accurate, non-double-counted repo total
+        for p, rp in iter_files(root, (m.get("paths") or []),
+                                list(excludes) + list(m.get("exclude", []))):
+            union_files[rp] = p
+        if nfiles == 0:
+            buckets["empty"].append(m["id"])
+        elif not m.get("auditedHash") or m.get("score") is None:
+            buckets["unaudited"].append(m["id"])
+        elif m.get("auditedHash") != chash:
+            buckets["stale"].append(m["id"])
+        else:
+            buckets["fresh"].append(m["id"])
+
+    tracked_loc = 0
+    for rp, p in union_files.items():
+        try:
+            data = open(p, "rb").read()
+            tracked_loc += data.count(b"\n") + (1 if data and not data.endswith(b"\n") else 0)
+        except OSError:
+            pass
+
+    if args.write:
+        meta = state.setdefault("meta", {})
+        meta["tracked_loc"] = tracked_loc
+        meta["tracked_files"] = len(union_files)
+        json.dump(state, open(args.state, "w", encoding="utf-8"),
+                  ensure_ascii=False, indent=1)
+
+    needs = buckets["stale"] + buckets["unaudited"]
+    report = {
+        "modules": len(state.get("modules", [])),
+        "tracked_loc": tracked_loc,
+        "tracked_files": len(union_files),
+        "needs_audit": needs,
+        "needs_audit_count": len(needs),
+        "up_to_date": len(needs) == 0 and not buckets["empty"],
+        **buckets,
+    }
+    print(json.dumps(report, ensure_ascii=False, indent=1))
+
+
+if __name__ == "__main__":
+    main()