#!/usr/bin/env python3 """Render a Claude-style light-theme graph.html dashboard from a .graph_state state file. Usage: render_graph_html.py [.graph_state] [graph.html] Defaults to reading ./.graph_state and writing ./graph.html. Called by the /graph skill at every checkpoint (initial plan + each fan-in barrier), so opening graph.html in a browser (it self-refreshes) tracks execution live. No third-party dependencies — stdlib only. """ import html import json import sys from datetime import datetime STATUS = { "pending": ("Pending", "#8C8579", "#EFECE3"), "in_progress": ("In Progress", "#CC785C", "#F7E9E2"), "shipped": ("Shipped", "#3D7A5A", "#DFEEE4"), "failed": ("Failed", "#B54A3E", "#F6DEDA"), "blocked": ("Blocked", "#9A6C3A", "#F2E6D4"), "skipped": ("Skipped", "#8C8579", "#EFECE3"), } def esc(s): return html.escape(str(s if s is not None else "")) def node_card(nid, n): st = n.get("status", "pending") label, fg, bg = STATUS.get(st, STATUS["pending"]) deps = n.get("deps") or [] deps_str = ", ".join(f"#{d}" for d in deps) if deps else "no deps" meta = [] if n.get("pr"): meta.append(f'PR #{esc(n["pr"])}') if n.get("branch"): meta.append(f'{esc(n["branch"])}') if n.get("attempts"): meta.append(f'attempt {esc(n["attempts"])}') meta_html = " · ".join(meta) err = f'
{esc(n["error"])}
' if n.get("error") else "" return f"""
#{esc(nid)} {label}
{esc(n.get('title','(untitled)'))}
{esc(deps_str)}
{f'
{meta_html}
' if meta_html else ''} {err}
""" def mermaid(state): lines = ["graph LR"] nodes = state.get("nodes", {}) for nid, n in nodes.items(): t = n.get("title", "") lines.append(f' n{nid}["#{nid} {t}"]') for nid, n in nodes.items(): for d in (n.get("deps") or []): lines.append(f" n{d} --> n{nid}") # color by status for st, (_, fg, bg) in STATUS.items(): ids = [f"n{nid}" for nid, n in nodes.items() if n.get("status") == st] if ids: lines.append(f" classDef {st} fill:{bg},stroke:{fg},color:#33312B;") lines.append(f" class {','.join(ids)} {st};") return "\n".join(lines) def render(state): nodes = state.get("nodes", {}) total = len(nodes) counts = {k: 0 for k in STATUS} for n in nodes.values(): counts[n.get("status", "pending")] = counts.get(n.get("status", "pending"), 0) + 1 shipped = counts.get("shipped", 0) pct = int(shipped / total * 100) if total else 0 waves = state.get("waves", []) cur = state.get("current_wave", 0) legend = "".join( f'{label}' for label, fg, bg in STATUS.values() ) wave_html = "" for wi, wave in enumerate(waves): state_cls = "cur" if wi == cur else ("done" if wi < cur else "future") cards = "".join(node_card(str(nid), nodes.get(str(nid), {"title": f"#{nid}"})) for nid in wave) wave_html += f"""

Wave {wi} ×{len(wave)} parallel {'running' if wi == cur else ''}

{cards}
""" stat = lambda k: f'{counts.get(k,0)} {STATUS[k][0].lower()}' stats = " · ".join(stat(k) for k in ["shipped", "in_progress", "failed", "blocked", "skipped", "pending"]) return f""" graph · {esc(state.get('task','execution'))}

{esc(state.get('task','Task Graph Execution'))}

{esc(state.get('repo',''))} · wave {cur} of {max(len(waves)-1,0)} · updated {esc(state.get('updated_at',''))}
{shipped}/{total} shipped ({pct}%)  —  {stats}
{legend}
{esc(mermaid(state))}
{wave_html}
""" def main(): src = sys.argv[1] if len(sys.argv) > 1 else ".graph_state" dst = sys.argv[2] if len(sys.argv) > 2 else "graph.html" with open(src, encoding="utf-8") as f: state = json.load(f) state.setdefault("updated_at", datetime.now().isoformat(timespec="seconds")) with open(dst, "w", encoding="utf-8") as f: f.write(render(state)) print(f"wrote {dst} from {src}") if __name__ == "__main__": main()