#!/usr/bin/env python3
"""Swarm Visual — HTML generator for live agent communication graphs.
Generates a self-contained HTML page with hub-and-spoke layout showing
real team members, their connections via bezier curves, animated particles
with trail effects, and live-updating status.
"""
import json
import math
from datetime import datetime, timezone
from html import escape as _esc
from pathlib import Path
TEAMS_DIR = Path.home() / ".claude" / "teams"
TASKS_DIR = Path.home() / ".claude" / "tasks"
# Color mapping for member.color -> CSS hex
COLOR_MAP = {
"pink": "#f5a0c0",
"green": "#98c379",
"cyan": "#56b6c2",
"blue": "#61afef",
"purple": "#c678dd",
"yellow": "#e5c07b",
"red": "#e06c75",
"orange": "#d19a66",
}
MODEL_BADGE = {
"claude-opus-4-6": ("opus", "#c678dd"),
"opus": ("opus", "#c678dd"),
"claude-sonnet-4-6": ("sonnet", "#61afef"),
"sonnet": ("sonnet", "#61afef"),
"claude-haiku-4-5": ("haiku", "#98c379"),
"haiku": ("haiku", "#98c379"),
}
def _is_recent(ts_str: str, seconds: int = 120) -> bool:
if not ts_str:
return False
try:
ts = datetime.fromisoformat(ts_str.replace("Z", "+00:00"))
return (datetime.now(timezone.utc) - ts).total_seconds() < seconds
except (ValueError, AttributeError):
return False
def _load_inbox(team_dir: Path, member_name: str) -> list[dict]:
inbox = team_dir / "inboxes" / f"{member_name}.json"
if not inbox.exists():
return []
try:
data = json.loads(inbox.read_text(encoding="utf-8"))
if not isinstance(data, list):
return []
return [
m for m in data
if not any(k in m.get("text", "") for k in
['"type":"task_assignment"', '"type":"shutdown_request"',
'"type":"idle_notification"'])
]
except (json.JSONDecodeError, OSError):
return []
def _load_tasks(team_name: str) -> list[dict]:
tasks = []
task_dir = TASKS_DIR / team_name
if not task_dir.exists():
return tasks
for tf in sorted(task_dir.glob("*.json"), key=lambda f: f.stem.zfill(10)):
try:
data = json.loads(tf.read_text(encoding="utf-8"))
if data.get("metadata", {}).get("_internal"):
continue
tasks.append(data)
except (json.JSONDecodeError, OSError):
continue
return tasks
def generate_comm_graph(team_name: str) -> str:
"""Generate a complete HTML comm graph for a team."""
team_dir = TEAMS_DIR / team_name
config_path = team_dir / "config.json"
if not config_path.exists():
return f"Team '{_esc(team_name)}' nicht gefunden."
try:
config = json.loads(config_path.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError):
return f"Team '{_esc(team_name)}' config fehlerhaft."
members = config.get("members", [])
if not members:
return f"Team '{_esc(team_name)}' hat keine Members."
# Separate lead from other members
lead = None
others = []
for m in members:
if m.get("agentType") == "team-lead" or m.get("name") == "team-lead":
lead = m
else:
others.append(m)
if not lead and members:
lead = members[0]
others = members[1:]
# Collect inbox data per member
member_data = []
for m in [lead] + others:
name = m.get("name", "?")
msgs = _load_inbox(team_dir, name)
model_key = m.get("model", "")
model_info = MODEL_BADGE.get(model_key, (model_key, "#8b949e"))
color = COLOR_MAP.get(m.get("color", ""), "#8b949e")
last_ts = msgs[-1].get("timestamp", "") if msgs else ""
member_data.append({
"name": name,
"model": model_info[0],
"model_color": model_info[1],
"color": color,
"agent_type": m.get("agentType", ""),
"msg_count": len(msgs),
"status": "working" if _is_recent(last_ts, 120) else "idle",
"last_msgs": msgs[-3:] if msgs else [],
"is_lead": m == lead,
"has_recent": _is_recent(last_ts, 300),
})
# Load tasks
tasks = _load_tasks(team_name)
# Collect all messages chronologically
all_msgs = []
for m in members:
name = m.get("name", "?")
for msg in _load_inbox(team_dir, name):
msg["_to"] = name
all_msgs.append(msg)
all_msgs.sort(key=lambda x: x.get("timestamp", ""))
recent_msgs = all_msgs[-10:]
# Layout: lead in center, others in circle
cx, cy = 400, 270
radius = 190
n = len(others)
positions = {}
positions[lead.get("name", "?")] = (cx, cy)
for i, m in enumerate(others):
angle = -math.pi / 2 + (2 * math.pi * i / max(n, 1))
x = cx + radius * math.cos(angle)
y = cy + radius * math.sin(angle)
positions[m.get("name", "?")] = (x, y)
lead_name = lead.get("name", "?")
# Build SVG bezier connections
svg_paths = []
for md in member_data:
if md["is_lead"]:
continue
lx, ly = positions[lead_name]
mx, my = positions[md["name"]]
# Control point: offset perpendicular to midpoint
midx, midy = (lx + mx) / 2, (ly + my) / 2
dx, dy = mx - lx, my - ly
length = math.sqrt(dx * dx + dy * dy) or 1
# Perpendicular offset (30px)
px, py = -dy / length * 30, dx / length * 30
cpx, cpy = midx + px, midy + py
active_class = "conn-active" if md["has_recent"] else "conn"
svg_paths.append(
f''
)
# Build node HTML
node_html_parts = []
for md in member_data:
x, y = positions[md["name"]]
assigned_task = next((t for t in tasks if t.get("owner") == md["name"]), None)
task_html = ""
if assigned_task:
st = assigned_task.get("status", "pending")
st_color = {"completed": "#98c379", "in_progress": "#e5c07b"}.get(st, "#8b949e")
desc_preview = _esc(assigned_task.get("description", "")[:100].replace('\n', ' '))
subj = _esc(assigned_task.get("subject", "")[:60])
task_html = f'''
{st.upper().replace("_"," ")}
{subj}
{desc_preview}
'''
msg_html = ""
if md["last_msgs"]:
msg_items = ""
for msg in md["last_msgs"]:
sender = _esc(msg.get("from", "?"))
text = _esc(msg.get("text", "")[:80].replace("\n", " "))
msg_items += f'
{sender}: {text}
'
msg_html = f'
{msg_items}
'
status_class = "status-working" if md["status"] == "working" else "status-idle"
safe_name = _esc(md['name'])
# Use data attribute for JS to avoid quote-breaking in onclick
node_html_parts.append(f'''
{safe_name}{md['model']}
{md['agent_type']}{md['msg_count']}
{task_html}
{msg_html}
''')
# Task stream
order = {"in_progress": 0, "pending": 1, "completed": 2}
tasks.sort(key=lambda t: order.get(t.get("status", ""), 9))
task_stream_items = ""
for t in tasks[:8]:
st = t.get("status", "pending")
st_icon = {"completed": "✓", "in_progress": "●"}.get(st, "○")
st_color = {"completed": "#98c379", "in_progress": "#e5c07b"}.get(st, "#8b949e")
owner = _esc(t.get("owner", ""))
task_stream_items += f'''