diff --git a/packages/web/package.json b/packages/web/package.json index e290378b9..8e0074306 100644 --- a/packages/web/package.json +++ b/packages/web/package.json @@ -28,6 +28,7 @@ "@composio/ao-plugin-workspace-worktree": "workspace:*", "@xterm/addon-fit": "^0.11.0", "@xterm/addon-web-links": "^0.12.0", + "d3-force": "^3.0.0", "next": "^15.1.0", "node-pty": "^1.1.0", "react": "^19.0.0", @@ -39,6 +40,7 @@ "@tailwindcss/postcss": "^4.0.0", "@testing-library/jest-dom": "^6.9.1", "@testing-library/react": "^16.1.0", + "@types/d3-force": "^3.0.10", "@types/react": "^19.0.0", "@types/react-dom": "^19.0.0", "@types/ws": "^8.18.1", diff --git a/packages/web/src/app/activity/page.tsx b/packages/web/src/app/activity/page.tsx new file mode 100644 index 000000000..352ad6f06 --- /dev/null +++ b/packages/web/src/app/activity/page.tsx @@ -0,0 +1,165 @@ +"use client"; + +import { useState, useEffect, useCallback, useRef } from "react"; +import { AgentMap } from "@/components/activity/AgentMap"; +import { BranchTimeline } from "@/components/activity/BranchTimeline"; +import { ActivityFeed } from "@/components/activity/ActivityFeed"; +import type { ActivityEvent, SessionGitActivity } from "@/lib/git-activity"; + +interface SessionSnapshot { + id: string; + branch: string | null; + status: string; + activity: string; + pr: { + number: number; + url: string; + state: string; + ciStatus: string; + } | null; +} + +interface ActivityState { + sessions: SessionSnapshot[]; + gitActivity: Record; + events: ActivityEvent[]; +} + +export default function ActivityPage() { + const [state, setState] = useState({ + sessions: [], + gitActivity: {}, + events: [], + }); + const [connected, setConnected] = useState(false); + const eventSourceRef = useRef(null); + + const connectToSSE = useCallback(() => { + if (eventSourceRef.current) { + eventSourceRef.current.close(); + } + + const es = new EventSource("/api/activity"); + eventSourceRef.current = es; + + es.onopen = () => { + setConnected(true); + }; + + es.onmessage = (event) => { + try { + const data = JSON.parse(event.data); + + if (data.type === "snapshot") { + setState((prev) => ({ + ...prev, + sessions: data.sessions, + gitActivity: data.gitActivity ?? {}, + })); + } else if (data.type === "activity") { + setState((prev) => ({ + ...prev, + events: [data.event, ...prev.events].slice(0, 100), // Keep last 100 events + })); + } else if (data.type === "state_update") { + setState((prev) => ({ + ...prev, + sessions: data.sessions, + gitActivity: { + ...prev.gitActivity, + ...data.gitActivity, + }, + })); + } + } catch { + // Ignore parse errors + } + }; + + es.onerror = () => { + setConnected(false); + es.close(); + // Reconnect after 3 seconds + setTimeout(connectToSSE, 3000); + }; + }, []); + + useEffect(() => { + connectToSSE(); + return () => { + if (eventSourceRef.current) { + eventSourceRef.current.close(); + } + }; + }, [connectToSSE]); + + return ( +
+ {/* Header */} +
+
+ + + + + +

+ Mission Control +

+
+
+
+ + + {connected ? "Live" : "Reconnecting..."} + +
+ + {state.sessions.length} agent{state.sessions.length !== 1 ? "s" : ""} active + +
+
+ + {/* Main content — three-panel layout */} +
+ {/* Left panel — Agent Map (Gource-inspired visualization) */} +
+

+ Agent Map +

+ +
+ + {/* Center panel — Branch Timeline */} +
+

+ Branch Timeline +

+ +
+ + {/* Right panel — Activity Feed */} +
+

+ Live Activity +

+ +
+
+
+ ); +} diff --git a/packages/web/src/app/api/activity/route.ts b/packages/web/src/app/api/activity/route.ts new file mode 100644 index 000000000..735a83d09 --- /dev/null +++ b/packages/web/src/app/api/activity/route.ts @@ -0,0 +1,195 @@ +import { getServices } from "@/lib/services"; +import { sessionToDashboard } from "@/lib/serialize"; +import { + pollSessionsActivity, + onActivity, + type ActivityEvent, + type SessionGitActivity, +} from "@/lib/git-activity"; + +export const dynamic = "force-dynamic"; + +/** + * GET /api/activity — SSE stream for Mission Control activity events + * + * Sends: + * - Initial snapshot of all session git activity + * - Real-time commit/file change events as they happen + * - Session state updates (PR opened, CI status, etc.) + */ +export async function GET(): Promise { + const encoder = new TextEncoder(); + let heartbeat: ReturnType | undefined; + let poller: ReturnType | undefined; + let unsubscribe: (() => void) | undefined; + + const stream = new ReadableStream({ + start(controller) { + // Buffer for events to send + const eventBuffer: ActivityEvent[] = []; + let isSendingInitial = true; + + // Subscribe to real-time activity events + unsubscribe = onActivity((event) => { + if (isSendingInitial) { + eventBuffer.push(event); + } else { + try { + controller.enqueue( + encoder.encode(`data: ${JSON.stringify({ type: "activity", event })}\n\n`), + ); + } catch { + // Stream closed + } + } + }); + + // Send initial snapshot + void (async () => { + try { + const { config, sessionManager } = await getServices(); + const sessions = await sessionManager.list(); + const activeSessions = sessions.filter( + (s) => !["done", "terminated", "killed"].includes(s.status), + ); + const dashboardSessions = activeSessions.map(sessionToDashboard); + + // Get default branch from first project config + const firstProject = Object.values(config.projects)[0]; + const defaultBranch = firstProject?.defaultBranch ?? "main"; + + // Poll initial activity + const activity = await pollSessionsActivity(dashboardSessions, defaultBranch); + + // Convert to serializable format + const activitySnapshot: Record = {}; + for (const [id, act] of activity) { + activitySnapshot[id] = { + ...act, + lastPolledAt: act.lastPolledAt, + commits: act.commits.map((c) => ({ + ...c, + timestamp: c.timestamp, + })), + }; + } + + // Send initial snapshot + const initialEvent = { + type: "snapshot", + sessions: dashboardSessions.map((s) => ({ + id: s.id, + branch: s.branch, + status: s.status, + activity: s.activity, + pr: s.pr + ? { + number: s.pr.number, + url: s.pr.url, + state: s.pr.state, + ciStatus: s.pr.ciStatus, + } + : null, + })), + gitActivity: activitySnapshot, + }; + controller.enqueue(encoder.encode(`data: ${JSON.stringify(initialEvent)}\n\n`)); + + // Send any buffered events + isSendingInitial = false; + for (const event of eventBuffer) { + try { + controller.enqueue( + encoder.encode(`data: ${JSON.stringify({ type: "activity", event })}\n\n`), + ); + } catch { + break; + } + } + eventBuffer.length = 0; + } catch { + // If services aren't available, send empty snapshot + controller.enqueue( + encoder.encode( + `data: ${JSON.stringify({ type: "snapshot", sessions: [], gitActivity: {} })}\n\n`, + ), + ); + isSendingInitial = false; + } + })(); + + // Send periodic heartbeat + heartbeat = setInterval(() => { + try { + controller.enqueue(encoder.encode(`: heartbeat\n\n`)); + } catch { + clearInterval(heartbeat); + clearInterval(poller); + } + }, 15000); + + // Poll for git activity every 10 seconds + poller = setInterval(() => { + void (async () => { + try { + const { config, sessionManager } = await getServices(); + const sessions = await sessionManager.list(); + const activeSessions = sessions.filter( + (s) => !["done", "terminated", "killed"].includes(s.status), + ); + const dashboardSessions = activeSessions.map(sessionToDashboard); + + const firstProject = Object.values(config.projects)[0]; + const defaultBranch = firstProject?.defaultBranch ?? "main"; + + // Poll activity — this will emit events via the subscription + const activity = await pollSessionsActivity(dashboardSessions, defaultBranch); + + // Send session state update + const stateUpdate = { + type: "state_update", + sessions: dashboardSessions.map((s) => ({ + id: s.id, + status: s.status, + activity: s.activity, + pr: s.pr + ? { + number: s.pr.number, + state: s.pr.state, + ciStatus: s.pr.ciStatus, + } + : null, + })), + gitActivity: Object.fromEntries( + Array.from(activity.entries()).map(([id, act]) => [ + id, + { + commitsAhead: act.commitsAhead, + recentFiles: act.recentFiles.slice(0, 10), + }, + ]), + ), + }; + controller.enqueue(encoder.encode(`data: ${JSON.stringify(stateUpdate)}\n\n`)); + } catch { + // Transient error — skip this poll + } + })(); + }, 10000); + }, + cancel() { + clearInterval(heartbeat); + clearInterval(poller); + if (unsubscribe) unsubscribe(); + }, + }); + + return new Response(stream, { + headers: { + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache", + Connection: "keep-alive", + "X-Accel-Buffering": "no", + }, + }); +} diff --git a/packages/web/src/components/Dashboard.tsx b/packages/web/src/components/Dashboard.tsx index fbee50abc..f93bc7b69 100644 --- a/packages/web/src/components/Dashboard.tsx +++ b/packages/web/src/components/Dashboard.tsx @@ -88,6 +88,12 @@ export function Dashboard({ sessions, stats, orchestratorId }: DashboardProps) { Agent Orchestrator
+ + mission control + {orchestratorId && ( { + if (events.length > 0) return events; + + // Create placeholder events from session states + return sessions.map((session) => ({ + id: `synthetic-${session.id}`, + type: "commit" as const, + sessionId: session.id, + branch: session.branch ?? "unknown", + message: session.pr + ? `PR #${session.pr.number} ${session.pr.state}` + : `Working on ${session.branch ?? "branch"}`, + timestamp: new Date(), + data: {}, + })); + }, [events, sessions]); + + if (displayEvents.length === 0) { + return ( +
+
+
+ + + +
+

Waiting for activity...

+

+ Events will appear here as agents work +

+
+
+ ); + } + + return ( +
+
+ {displayEvents.map((event) => ( + + ))} +
+
+ ); +} + +function ActivityEventRow({ event }: { event: ActivityEvent }) { + const color = getSessionColor(event.sessionId); + + return ( +
+ {/* Session indicator */} +
+
+ + + +
+
+ + {/* Content */} +
+
+ + {event.sessionId} + + + {formatRelativeTime(event.timestamp)} + +
+

+ {event.message} +

+ {event.data.files && event.data.files.length > 0 && ( +
+ {event.data.files.slice(0, 3).map((file) => ( + + {file.split("/").pop()} + + ))} + {event.data.files.length > 3 && ( + + +{event.data.files.length - 3} + + )} +
+ )} +
+
+ ); +} diff --git a/packages/web/src/components/activity/AgentMap.tsx b/packages/web/src/components/activity/AgentMap.tsx new file mode 100644 index 000000000..c58c7dbb6 --- /dev/null +++ b/packages/web/src/components/activity/AgentMap.tsx @@ -0,0 +1,442 @@ +"use client"; + +import { useEffect, useRef, useMemo, useState } from "react"; +import { + forceSimulation, + forceLink, + forceManyBody, + forceCenter, + forceCollide, + type Simulation, + type SimulationNodeDatum, + type SimulationLinkDatum, +} from "d3-force"; +import type { ActivityEvent, SessionGitActivity } from "@/lib/git-activity"; + +interface SessionSnapshot { + id: string; + branch: string | null; + status: string; + activity: string; + pr: { + number: number; + state: string; + ciStatus: string; + } | null; +} + +interface AgentMapProps { + sessions: SessionSnapshot[]; + gitActivity: Record; + events: ActivityEvent[]; +} + +interface Node extends SimulationNodeDatum { + id: string; + type: "main" | "agent" | "file"; + label: string; + color: string; + radius: number; + pulse?: boolean; + lastPulse?: number; +} + +interface Link extends SimulationLinkDatum { + source: string | Node; + target: string | Node; + type: "branch" | "file"; + strength: number; + pulse?: boolean; + lastPulse?: number; +} + +/** Get a color for a session based on its ID (consistent hashing) */ +function getSessionColor(sessionId: string): string { + const colors = [ + "#58a6ff", // blue + "#3fb950", // green + "#a371f7", // purple + "#f0883e", // orange + "#f778ba", // pink + "#79c0ff", // light blue + "#7ee787", // light green + "#d2a8ff", // light purple + ]; + let hash = 0; + for (let i = 0; i < sessionId.length; i++) { + hash = (hash << 5) - hash + sessionId.charCodeAt(i); + hash = hash & hash; + } + return colors[Math.abs(hash) % colors.length]; +} + +/** Get file color based on extension */ +function getFileColor(filename: string): string { + const ext = filename.split(".").pop()?.toLowerCase(); + switch (ext) { + case "ts": + case "tsx": + return "#3178c6"; + case "js": + case "jsx": + return "#f7df1e"; + case "css": + case "scss": + return "#264de4"; + case "json": + return "#cbcb41"; + case "md": + return "#083fa1"; + case "yaml": + case "yml": + return "#cb171e"; + default: + return "#6e7681"; + } +} + +export function AgentMap({ sessions, gitActivity, events }: AgentMapProps) { + const canvasRef = useRef(null); + const simulationRef = useRef | null>(null); + const nodesRef = useRef([]); + const linksRef = useRef([]); + const animationRef = useRef(0); + const [dimensions, setDimensions] = useState({ width: 340, height: 400 }); + + // Build graph data from sessions and activity + const graphData = useMemo(() => { + const nodes: Node[] = []; + const links: Link[] = []; + const fileMap = new Map>(); // file -> agent IDs + + // Central "main" node + nodes.push({ + id: "main", + type: "main", + label: "main", + color: "#484f58", + radius: 20, + }); + + // Agent nodes + for (const session of sessions) { + const color = getSessionColor(session.id); + const activity = gitActivity[session.id]; + const isActive = session.activity === "active"; + + nodes.push({ + id: session.id, + type: "agent", + label: session.id, + color, + radius: isActive ? 14 : 10, + pulse: isActive, + }); + + // Link agent to main + links.push({ + source: "main", + target: session.id, + type: "branch", + strength: 0.8, + }); + + // Track files touched by this agent + if (activity?.recentFiles) { + for (const file of activity.recentFiles.slice(0, 5)) { + const existing = fileMap.get(file); + if (existing) { + existing.add(session.id); + } else { + fileMap.set(file, new Set([session.id])); + } + } + } + } + + // File nodes (shared files are larger) + for (const [file, agentIds] of fileMap) { + const shortName = file.split("/").pop() ?? file; + nodes.push({ + id: `file:${file}`, + type: "file", + label: shortName, + color: getFileColor(file), + radius: 4 + Math.min(agentIds.size * 2, 6), + }); + + // Link file to all agents that touched it + for (const agentId of agentIds) { + links.push({ + source: agentId, + target: `file:${file}`, + type: "file", + strength: 0.3, + }); + } + } + + return { nodes, links }; + }, [sessions, gitActivity]); + + // Mark nodes/links as pulsing based on recent events + useEffect(() => { + const now = Date.now(); + for (const event of events.slice(0, 10)) { + const eventTime = + typeof event.timestamp === "string" + ? new Date(event.timestamp).getTime() + : event.timestamp.getTime(); + if (now - eventTime < 5000) { + // Within last 5 seconds + const node = nodesRef.current.find((n) => n.id === event.sessionId); + if (node) { + node.pulse = true; + node.lastPulse = now; + } + + // Pulse links to files + if (event.data.files) { + for (const file of event.data.files) { + const link = linksRef.current.find( + (l) => + (typeof l.source === "object" ? l.source.id : l.source) === event.sessionId && + (typeof l.target === "object" ? l.target.id : l.target) === `file:${file}`, + ); + if (link) { + link.pulse = true; + link.lastPulse = now; + } + } + } + } + } + }, [events]); + + // Handle resize + useEffect(() => { + const canvas = canvasRef.current; + if (!canvas) return; + + const updateDimensions = () => { + const parent = canvas.parentElement; + if (parent) { + setDimensions({ + width: parent.clientWidth, + height: Math.max(parent.clientHeight - 30, 300), + }); + } + }; + + updateDimensions(); + const resizeObserver = new ResizeObserver(updateDimensions); + if (canvas.parentElement) { + resizeObserver.observe(canvas.parentElement); + } + + return () => resizeObserver.disconnect(); + }, []); + + // Initialize and update simulation + useEffect(() => { + const { nodes, links } = graphData; + + // Copy nodes and links (D3 mutates them) + nodesRef.current = nodes.map((n) => ({ ...n })); + linksRef.current = links.map((l) => ({ ...l })); + + // Create or update simulation + if (!simulationRef.current) { + simulationRef.current = forceSimulation(); + } + + const sim = simulationRef.current; + sim.nodes(nodesRef.current); + sim + .force( + "link", + forceLink(linksRef.current) + .id((d) => d.id) + .distance((d) => (d.type === "branch" ? 60 : 40)) + .strength((d) => d.strength), + ) + .force("charge", forceManyBody().strength(-80)) + .force("center", forceCenter(dimensions.width / 2, dimensions.height / 2)) + .force( + "collide", + forceCollide().radius((d) => d.radius + 3), + ) + .alpha(0.5) + .restart(); + + return () => { + sim.stop(); + }; + }, [graphData, dimensions]); + + // Animation loop + useEffect(() => { + const canvas = canvasRef.current; + if (!canvas) return; + + const ctx = canvas.getContext("2d"); + if (!ctx) return; + + const dpr = window.devicePixelRatio || 1; + canvas.width = dimensions.width * dpr; + canvas.height = dimensions.height * dpr; + ctx.scale(dpr, dpr); + + const render = () => { + const now = Date.now(); + + // Clear canvas + ctx.fillStyle = "#0d1117"; + ctx.fillRect(0, 0, dimensions.width, dimensions.height); + + // Draw links + for (const link of linksRef.current) { + const source = link.source as Node; + const target = link.target as Node; + if (!source.x || !source.y || !target.x || !target.y) continue; + + // Calculate pulse effect + const pulseAge = link.lastPulse ? now - link.lastPulse : Infinity; + const isPulsing = pulseAge < 2000; + const pulseAlpha = isPulsing ? Math.max(0, 1 - pulseAge / 2000) : 0; + + ctx.beginPath(); + ctx.moveTo(source.x, source.y); + ctx.lineTo(target.x, target.y); + + if (isPulsing) { + ctx.strokeStyle = `rgba(88, 166, 255, ${0.3 + pulseAlpha * 0.7})`; + ctx.lineWidth = 1 + pulseAlpha * 2; + } else { + ctx.strokeStyle = + link.type === "branch" ? "rgba(72, 79, 88, 0.6)" : "rgba(72, 79, 88, 0.3)"; + ctx.lineWidth = link.type === "branch" ? 2 : 1; + } + ctx.stroke(); + + // Clear pulse after fade + if (pulseAge > 2000) { + link.pulse = false; + } + } + + // Draw nodes + for (const node of nodesRef.current) { + if (!node.x || !node.y) continue; + + // Calculate pulse effect + const pulseAge = node.lastPulse ? now - node.lastPulse : Infinity; + const isPulsing = node.pulse || pulseAge < 3000; + const pulsePhase = (now % 1500) / 1500; + const pulseScale = isPulsing ? 1 + Math.sin(pulsePhase * Math.PI * 2) * 0.15 : 1; + + // Draw glow for active nodes + if (isPulsing && node.type === "agent") { + const gradient = ctx.createRadialGradient( + node.x, + node.y, + node.radius * pulseScale, + node.x, + node.y, + node.radius * pulseScale * 2.5, + ); + gradient.addColorStop(0, `${node.color}40`); + gradient.addColorStop(1, "transparent"); + ctx.fillStyle = gradient; + ctx.beginPath(); + ctx.arc(node.x, node.y, node.radius * pulseScale * 2.5, 0, Math.PI * 2); + ctx.fill(); + } + + // Draw node + ctx.beginPath(); + ctx.arc(node.x, node.y, node.radius * pulseScale, 0, Math.PI * 2); + ctx.fillStyle = node.color; + ctx.fill(); + + // Draw border for main node + if (node.type === "main") { + ctx.strokeStyle = "#30363d"; + ctx.lineWidth = 2; + ctx.stroke(); + } + + // Draw label for agents + if (node.type === "agent" || node.type === "main") { + ctx.font = node.type === "main" ? "bold 11px system-ui" : "10px system-ui"; + ctx.fillStyle = "#e6edf3"; + ctx.textAlign = "center"; + ctx.textBaseline = "middle"; + + const label = node.type === "main" ? "main" : (node.label.split("-").pop() ?? node.label); + ctx.fillText(label, node.x, node.y + node.radius + 12); + } + + // Clear pulse after some time (but keep for active agents) + if (pulseAge > 3000 && !node.pulse) { + node.lastPulse = undefined; + } + } + + animationRef.current = requestAnimationFrame(render); + }; + + animationRef.current = requestAnimationFrame(render); + + return () => { + cancelAnimationFrame(animationRef.current); + }; + }, [dimensions]); + + if (sessions.length === 0) { + return ( +
+
+
+ + + + + +
+

No agents active

+

+ Agents will appear as orbiting nodes +

+
+
+ ); + } + + return ( +
+ +
+
+ + main +
+
+ + agent +
+
+ + file +
+
+
+ ); +} diff --git a/packages/web/src/components/activity/BranchTimeline.tsx b/packages/web/src/components/activity/BranchTimeline.tsx new file mode 100644 index 000000000..e5f0e88c0 --- /dev/null +++ b/packages/web/src/components/activity/BranchTimeline.tsx @@ -0,0 +1,271 @@ +"use client"; + +import { useMemo } from "react"; +import type { SessionGitActivity } from "@/lib/git-activity"; + +interface SessionSnapshot { + id: string; + branch: string | null; + status: string; + activity: string; + pr: { + number: number; + state: string; + ciStatus: string; + } | null; +} + +interface BranchTimelineProps { + sessions: SessionSnapshot[]; + gitActivity: Record; +} + +/** Get a color for a session based on its ID (consistent hashing) */ +function getSessionColor(sessionId: string): string { + const colors = [ + "#58a6ff", // blue + "#3fb950", // green + "#a371f7", // purple + "#f0883e", // orange + "#f778ba", // pink + "#79c0ff", // light blue + "#7ee787", // light green + "#d2a8ff", // light purple + ]; + let hash = 0; + for (let i = 0; i < sessionId.length; i++) { + hash = (hash << 5) - hash + sessionId.charCodeAt(i); + hash = hash & hash; + } + return colors[Math.abs(hash) % colors.length]; +} + +/** Get status indicator color */ +function getStatusColor(status: string): string { + switch (status) { + case "working": + case "spawning": + return "var(--color-accent-green)"; + case "pr_open": + case "review_pending": + return "var(--color-accent-blue)"; + case "ci_failed": + case "changes_requested": + return "var(--color-accent-red)"; + case "approved": + case "mergeable": + return "var(--color-accent-purple)"; + case "merged": + case "done": + return "var(--color-text-muted)"; + default: + return "var(--color-accent-yellow)"; + } +} + +/** Get CI status indicator */ +function getCIIndicator(ciStatus: string | undefined): { color: string; label: string } { + switch (ciStatus) { + case "passing": + return { color: "var(--color-accent-green)", label: "CI passing" }; + case "failing": + return { color: "var(--color-accent-red)", label: "CI failing" }; + case "pending": + return { color: "var(--color-accent-yellow)", label: "CI running" }; + default: + return { color: "var(--color-text-muted)", label: "No CI" }; + } +} + +export function BranchTimeline({ sessions, gitActivity }: BranchTimelineProps) { + const sortedSessions = useMemo(() => { + return [...sessions].sort((a, b) => { + // Active sessions first + const aActive = a.activity === "active"; + const bActive = b.activity === "active"; + if (aActive !== bActive) return bActive ? 1 : -1; + + // Then by commits ahead + const aCommits = gitActivity[a.id]?.commitsAhead ?? 0; + const bCommits = gitActivity[b.id]?.commitsAhead ?? 0; + return bCommits - aCommits; + }); + }, [sessions, gitActivity]); + + if (sortedSessions.length === 0) { + return ( +
+
+
+ + + +
+

No active branches

+

+ Spawn agents to see branch activity +

+
+
+ ); + } + + return ( +
+ {/* Main branch reference line */} +
+
+ main +
+
+
+
+ + + +
+
+
+ + {/* Branch swimlanes */} + {sortedSessions.map((session) => ( + + ))} +
+ ); +} + +function BranchLane({ + session, + activity, +}: { + session: SessionSnapshot; + activity: SessionGitActivity | undefined; +}) { + const color = getSessionColor(session.id); + const statusColor = getStatusColor(session.status); + const commits = activity?.commits ?? []; + const commitsAhead = activity?.commitsAhead ?? 0; + const ci = session.pr ? getCIIndicator(session.pr.ciStatus) : null; + + return ( +
+ {/* Session label */} +
+
+ + + {session.id} + +
+ {session.branch && ( +
+ {session.branch} +
+ )} +
+ + {/* Timeline */} +
+ {/* Branch line */} +
+ + {/* Diverge point from main */} +
+ + {/* Commits as nodes */} +
+ {commits.length > 0 ? ( + <> + {commits.slice(0, 8).map((commit, idx) => ( +
+
+ {/* Tooltip */} +
+
+ {commit.shortHash} +
+
+ {commit.message} +
+
+
+ ))} + + ) : ( + + No recent commits + + )} +
+ + {/* PR indicator if exists */} + {session.pr && ( +
+ + PR #{session.pr.number} + + {ci && ( + + )} +
+ )} +
+ + {/* Commits ahead badge */} +
+ {commitsAhead > 0 && ( + + +{commitsAhead} + + )} +
+
+ ); +} diff --git a/packages/web/src/components/activity/index.ts b/packages/web/src/components/activity/index.ts new file mode 100644 index 000000000..209b22159 --- /dev/null +++ b/packages/web/src/components/activity/index.ts @@ -0,0 +1,3 @@ +export { ActivityFeed } from "./ActivityFeed.js"; +export { AgentMap } from "./AgentMap.js"; +export { BranchTimeline } from "./BranchTimeline.js"; diff --git a/packages/web/src/lib/git-activity.ts b/packages/web/src/lib/git-activity.ts new file mode 100644 index 000000000..532522f91 --- /dev/null +++ b/packages/web/src/lib/git-activity.ts @@ -0,0 +1,284 @@ +/** + * Git Activity Polling — Tracks commits, file changes, and branch activity. + * + * Polls git log on each session's worktree to detect: + * - New commits + * - Files modified in recent commits + * - Branch divergence from main + */ + +import { execFile } from "node:child_process"; +import { promisify } from "node:util"; +import type { DashboardSession } from "./types.js"; + +const execFileAsync = promisify(execFile); +const EXEC_TIMEOUT = 10_000; + +/** A single commit from git log */ +export interface GitCommit { + hash: string; + shortHash: string; + message: string; + author: string; + timestamp: Date; + files: string[]; +} + +/** Git activity for a single session */ +export interface SessionGitActivity { + sessionId: string; + branch: string; + commits: GitCommit[]; + /** How many commits ahead of main branch */ + commitsAhead: number; + /** Files changed in recent commits */ + recentFiles: string[]; + lastPolledAt: Date; +} + +/** A real-time activity event for the event stream */ +export interface ActivityEvent { + id: string; + type: "commit" | "push" | "file_change" | "pr_opened" | "ci_status" | "review"; + sessionId: string; + branch: string; + message: string; + timestamp: Date; + data: { + files?: string[]; + commitHash?: string; + prNumber?: number; + ciStatus?: string; + }; +} + +/** Cached git activity by session ID */ +const activityCache = new Map(); + +/** Track seen commit hashes to detect new commits */ +const seenCommits = new Map>(); + +/** Event listeners for new activity */ +type ActivityListener = (event: ActivityEvent) => void; +const listeners: ActivityListener[] = []; + +/** Subscribe to activity events */ +export function onActivity(listener: ActivityListener): () => void { + listeners.push(listener); + return () => { + const idx = listeners.indexOf(listener); + if (idx >= 0) listeners.splice(idx, 1); + }; +} + +/** Emit an activity event to all listeners */ +function emitActivity(event: ActivityEvent): void { + for (const listener of listeners) { + try { + listener(event); + } catch { + // Ignore listener errors + } + } +} + +/** Generate a unique event ID */ +function generateEventId(): string { + return `${Date.now()}-${Math.random().toString(36).slice(2, 9)}`; +} + +/** + * Parse git log output into commits. + * Format: hash|short|author|timestamp|subject + */ +function parseGitLog(output: string, files: Map): GitCommit[] { + const commits: GitCommit[] = []; + const lines = output.trim().split("\n").filter(Boolean); + + for (const line of lines) { + const [hash, shortHash, author, timestamp, ...messageParts] = line.split("|"); + if (!hash) continue; + + commits.push({ + hash, + shortHash: shortHash ?? hash.slice(0, 7), + author: author ?? "unknown", + timestamp: new Date(timestamp ?? Date.now()), + message: messageParts.join("|"), + files: files.get(hash) ?? [], + }); + } + + return commits; +} + +/** + * Get recent git activity for a session's worktree. + * @param workspacePath - Path to the git worktree + * @param branch - Branch name to check + * @param defaultBranch - Main branch to compare against (e.g., "main") + * @param sinceMinutes - How far back to look for commits + */ +export async function getGitActivity( + workspacePath: string, + branch: string, + defaultBranch: string = "main", + sinceMinutes: number = 30, +): Promise<{ commits: GitCommit[]; commitsAhead: number }> { + try { + // Get recent commits with custom format + const logFormat = "%H|%h|%an|%aI|%s"; + const { stdout: logOutput } = await execFileAsync( + "git", + ["log", `--since=${sinceMinutes} minutes ago`, `--format=${logFormat}`, "-n", "50"], + { cwd: workspacePath, timeout: EXEC_TIMEOUT }, + ); + + // Get files changed per commit + const files = new Map(); + if (logOutput.trim()) { + const hashes = logOutput + .trim() + .split("\n") + .map((l) => l.split("|")[0]) + .filter(Boolean); + + for (const hash of hashes.slice(0, 10)) { + // Only get files for most recent 10 commits + try { + const { stdout: filesOutput } = await execFileAsync( + "git", + ["diff-tree", "--no-commit-id", "--name-only", "-r", hash], + { cwd: workspacePath, timeout: EXEC_TIMEOUT }, + ); + files.set(hash, filesOutput.trim().split("\n").filter(Boolean)); + } catch { + // Skip if we can't get files + } + } + } + + const commits = parseGitLog(logOutput, files); + + // Count commits ahead of default branch + let commitsAhead = 0; + try { + const { stdout: aheadOutput } = await execFileAsync( + "git", + ["rev-list", "--count", `origin/${defaultBranch}..HEAD`], + { cwd: workspacePath, timeout: EXEC_TIMEOUT }, + ); + commitsAhead = parseInt(aheadOutput.trim(), 10) || 0; + } catch { + // Branch comparison might fail if remote doesn't exist + } + + return { commits, commitsAhead }; + } catch { + // Git command failed — return empty + return { commits: [], commitsAhead: 0 }; + } +} + +/** + * Poll git activity for multiple sessions. + * Emits events for new commits detected. + */ +export async function pollSessionsActivity( + sessions: DashboardSession[], + defaultBranch: string = "main", +): Promise> { + const results = new Map(); + + await Promise.all( + sessions.map(async (session) => { + if (!session.metadata?.["worktree"] || !session.branch) return; + + const workspacePath = session.metadata["worktree"]; + const { commits, commitsAhead } = await getGitActivity( + workspacePath, + session.branch, + defaultBranch, + ); + + // Check for new commits and emit events + const sessionSeen = seenCommits.get(session.id) ?? new Set(); + const recentFiles = new Set(); + + for (const commit of commits) { + // Collect recent files + for (const file of commit.files) { + recentFiles.add(file); + } + + // Emit event for new commits + if (!sessionSeen.has(commit.hash)) { + sessionSeen.add(commit.hash); + emitActivity({ + id: generateEventId(), + type: "commit", + sessionId: session.id, + branch: session.branch, + message: `committed: "${commit.message}"`, + timestamp: commit.timestamp, + data: { + files: commit.files, + commitHash: commit.shortHash, + }, + }); + + // Also emit file change events for significant changes + if (commit.files.length > 0) { + emitActivity({ + id: generateEventId(), + type: "file_change", + sessionId: session.id, + branch: session.branch, + message: `modified ${commit.files.slice(0, 3).join(", ")}${commit.files.length > 3 ? ` +${commit.files.length - 3} more` : ""}`, + timestamp: commit.timestamp, + data: { files: commit.files }, + }); + } + } + } + + seenCommits.set(session.id, sessionSeen); + + const activity: SessionGitActivity = { + sessionId: session.id, + branch: session.branch, + commits, + commitsAhead, + recentFiles: Array.from(recentFiles), + lastPolledAt: new Date(), + }; + + results.set(session.id, activity); + activityCache.set(session.id, activity); + }), + ); + + return results; +} + +/** + * Get cached activity for a session. + */ +export function getCachedActivity(sessionId: string): SessionGitActivity | undefined { + return activityCache.get(sessionId); +} + +/** + * Get all cached activity. + */ +export function getAllCachedActivity(): Map { + return new Map(activityCache); +} + +/** + * Clear activity cache for a session. + */ +export function clearActivityCache(sessionId: string): void { + activityCache.delete(sessionId); + seenCommits.delete(sessionId); +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 6a0bc9134..d34769b08 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -466,6 +466,9 @@ importers: '@xterm/addon-web-links': specifier: ^0.12.0 version: 0.12.0 + d3-force: + specifier: ^3.0.0 + version: 3.0.0 next: specifier: ^15.1.0 version: 15.5.12(@babel/core@7.29.0)(@opentelemetry/api@1.9.0)(react-dom@19.2.4(react@19.2.4))(react@19.2.4) @@ -494,6 +497,9 @@ importers: '@testing-library/react': specifier: ^16.1.0 version: 16.3.2(@testing-library/dom@10.4.1)(@types/react-dom@19.2.3(@types/react@19.2.14))(@types/react@19.2.14)(react-dom@19.2.4(react@19.2.4))(react@19.2.4) + '@types/d3-force': + specifier: ^3.0.10 + version: 3.0.10 '@types/react': specifier: ^19.0.0 version: 19.2.14 @@ -1718,6 +1724,9 @@ packages: '@types/chai@5.2.3': resolution: {integrity: sha512-Mw558oeA9fFbv65/y4mHtXDs9bPnFMZAL/jxdPFUpOHHIXX91mcgEHbS5Lahr+pwZFR8A7GQleRWeI6cGFC2UA==} + '@types/d3-force@3.0.10': + resolution: {integrity: sha512-ZYeSaCF3p73RdOKcjj+swRlZfnYpK1EbaDiYICEEp5Q6sUiqFaFQ9qgoshp5CzIyyb/yD09kD9o2zEltCexlgw==} + '@types/deep-eql@4.0.2': resolution: {integrity: sha512-c9h9dVVMigMPc4bwTvC5dxqtqJZwQPePsWjPlpSOnojbor6pGqdk541lfA7AqFQr5pB1BRdq0juY9db81BwyFw==} @@ -2166,6 +2175,22 @@ packages: csstype@3.2.3: resolution: {integrity: sha512-z1HGKcYy2xA8AGQfwrn0PAy+PB7X/GSj3UVJW9qKyn43xWa+gl5nXmU4qqLMRzWVLFC8KusUX8T/0kCiOYpAIQ==} + d3-dispatch@3.0.1: + resolution: {integrity: sha512-rzUyPU/S7rwUflMyLc1ETDeBj0NRuHKKAcvukozwhshr6g6c5d8zh4c2gQjY2bZ0dXeGLWc1PF174P2tVvKhfg==} + engines: {node: '>=12'} + + d3-force@3.0.0: + resolution: {integrity: sha512-zxV/SsA+U4yte8051P4ECydjD/S+qeYtnaIyAs9tgHCqfguma/aAQDjo85A9Z6EKhBirHRJHXIgJUlffT4wdLg==} + engines: {node: '>=12'} + + d3-quadtree@3.0.1: + resolution: {integrity: sha512-04xDrxQTDTCFwP5H6hRhsRcb9xxv2RzkcsygFzmkSIOJy3PeRJP7sNk3VRIbKXcog561P9oU0/rVH6vDROAgUw==} + engines: {node: '>=12'} + + d3-timer@3.0.1: + resolution: {integrity: sha512-ndfJ/JxxMd3nw31uyKoY2naivF+r29V+Lc0svZxe1JvvIRmi8hUsrMvdOwgS1o6uBHmiz91geQ0ylPP0aj1VUA==} + engines: {node: '>=12'} + data-urls@5.0.0: resolution: {integrity: sha512-ZYP5VBHshaDAiVZxjbRVcFJpc+4xGgT0bK3vzy1HLN8jTO975HEbuYzZJcHoQEY5K1a0z8YayJkyVETa08eNTg==} engines: {node: '>=18'} @@ -4922,6 +4947,8 @@ snapshots: '@types/deep-eql': 4.0.2 assertion-error: 2.0.1 + '@types/d3-force@3.0.10': {} + '@types/deep-eql@4.0.2': {} '@types/esrecurse@4.3.1': {} @@ -5432,6 +5459,18 @@ snapshots: csstype@3.2.3: {} + d3-dispatch@3.0.1: {} + + d3-force@3.0.0: + dependencies: + d3-dispatch: 3.0.1 + d3-quadtree: 3.0.1 + d3-timer: 3.0.1 + + d3-quadtree@3.0.1: {} + + d3-timer@3.0.1: {} + data-urls@5.0.0: dependencies: whatwg-mimetype: 4.0.0