From 59f92f18dffbf9f8ccb50c1be3e95d06db01dc44 Mon Sep 17 00:00:00 2001 From: Yahya Alhinai Date: Sun, 28 Jun 2026 03:50:56 +0000 Subject: [PATCH] feat: add realtime pod activity streams --- backend/src/activity/store.ts | 196 +++++++++++++++++++++++++ backend/src/server.ts | 46 ++++++ frontend/src/components/PodView.tsx | 166 ++++++++++++++++++++- frontend/src/hooks/use-pod-activity.ts | 63 ++++++++ frontend/src/lib/api.ts | 12 +- scripts/verify-frontend.mjs | 33 +++-- shared/src/activity.ts | 19 +++ shared/src/index.ts | 6 + 8 files changed, 526 insertions(+), 15 deletions(-) create mode 100644 backend/src/activity/store.ts create mode 100644 frontend/src/hooks/use-pod-activity.ts create mode 100644 shared/src/activity.ts diff --git a/backend/src/activity/store.ts b/backend/src/activity/store.ts new file mode 100644 index 0000000..946674a --- /dev/null +++ b/backend/src/activity/store.ts @@ -0,0 +1,196 @@ +import type { + Collision, + EngineerContext, + Intervention, + InterventionOutcome, + PodActivityEvent, +} from '@podman/shared'; +import { getDb } from '../memory/db.js'; + +interface EngineerStateDoc { + _id: string; + podId: string; + name: string; + changedFiles?: string[]; + diffStat?: string | null; + recentCommit?: string | null; + branch?: string | null; + gitUpdatedAt?: Date | string; + updatedAt?: Date | string; +} + +function toIso(value: Date | string | undefined): string { + if (value instanceof Date) return value.toISOString(); + if (value) return new Date(value).toISOString(); + return new Date(0).toISOString(); +} + +function clean(value: string | undefined): string | undefined { + const trimmed = value?.trim(); + return trimmed || undefined; +} + +function shortFiles(files: string[] | undefined): string { + if (!files?.length) return 'clean working tree'; + const sample = files.slice(0, 3).join(', '); + return files.length > 3 ? `${sample}, +${files.length - 3} more` : sample; +} + +function observationEvent(doc: EngineerContext): PodActivityEvent { + const file = clean(doc.currentFile); + const symbol = clean(doc.currentSymbol); + return { + id: `observation:${doc.engineerId}:${doc.observedAt}`, + podId: doc.podId, + kind: 'observation', + source: 'vision', + actor: doc.engineerId, + actors: [doc.engineerId], + file, + title: file ? `Working in ${file}` : 'Screen context updated', + detail: [ + symbol ? `symbol ${symbol}` : undefined, + clean(doc.activity), + doc.hasUnpushedChanges ? 'unpushed changes visible' : undefined, + `confidence ${Math.round(doc.confidence * 100)}%`, + ] + .filter(Boolean) + .join(' · '), + severity: doc.hasUnpushedChanges ? 'warn' : 'info', + at: doc.observedAt, + }; +} + +function gitEvent(doc: EngineerStateDoc): PodActivityEvent { + const changedFiles = doc.changedFiles ?? []; + return { + id: `git:${doc._id}:${toIso(doc.gitUpdatedAt ?? doc.updatedAt)}`, + podId: doc.podId, + kind: 'git', + source: 'git', + actor: doc.name, + actors: [doc.name], + title: changedFiles.length ? `${changedFiles.length} local file changes` : 'Git state is clean', + detail: [ + doc.branch ? `branch ${doc.branch}` : undefined, + shortFiles(changedFiles), + doc.recentCommit ? `head ${doc.recentCommit}` : undefined, + ] + .filter(Boolean) + .join(' · '), + severity: changedFiles.length ? 'warn' : 'info', + at: toIso(doc.gitUpdatedAt ?? doc.updatedAt), + }; +} + +function collisionEvent(doc: Collision): PodActivityEvent { + return { + id: `collision:${doc.id}`, + podId: doc.podId, + kind: 'collision', + source: 'memory', + actor: doc.engineers[0], + actors: doc.engineers, + file: doc.file, + title: `${doc.engineers.join(' + ')} conflict on ${doc.file}`, + detail: [ + doc.symbol ? `symbol ${doc.symbol}` : undefined, + doc.githubState?.unpushed ? 'unpushed local changes involved' : undefined, + doc.githubState?.openPrs?.length + ? `open PRs ${doc.githubState.openPrs.join(', ')}` + : undefined, + ] + .filter(Boolean) + .join(' · '), + severity: doc.severity, + at: doc.detectedAt, + }; +} + +function interventionEvent(doc: Intervention): PodActivityEvent { + return { + id: `intervention:${doc.id}`, + podId: doc.podId, + kind: 'intervention', + source: 'hermes', + title: `Hermes ${doc.status} ${doc.suggestedAction.kind.replaceAll('_', ' ')}`, + detail: doc.message, + severity: doc.status === 'accepted' ? 'success' : doc.status === 'dismissed' ? 'info' : 'warn', + at: doc.createdAt, + }; +} + +function outcomeEvent(doc: InterventionOutcome): PodActivityEvent { + return { + id: `outcome:${doc.interventionId}:${doc.recordedAt}`, + podId: doc.podId, + kind: 'outcome', + source: 'policy', + title: doc.accepted ? 'Intervention accepted' : 'Intervention dismissed', + detail: doc.wasRealCollision ? 'confirmed real collision' : 'marked as false positive', + severity: doc.accepted ? 'success' : 'info', + at: doc.recordedAt, + }; +} + +export async function listPodActivity(podId: string, limit = 80): Promise { + const db = await getDb(); + const [observations, gitStates, collisions, interventions, outcomes] = await Promise.all([ + db + .collection('observations') + .find({ podId }, { projection: { _id: 0 } }) + .sort({ observedAt: -1 }) + .limit(limit) + .toArray(), + db + .collection('engineer_states') + .find( + { podId }, + { + projection: { + _id: 1, + podId: 1, + name: 1, + changedFiles: 1, + diffStat: 1, + recentCommit: 1, + branch: 1, + gitUpdatedAt: 1, + updatedAt: 1, + }, + }, + ) + .sort({ gitUpdatedAt: -1 }) + .limit(limit) + .toArray(), + db + .collection('collisions') + .find({ podId }, { projection: { _id: 0 } }) + .sort({ detectedAt: -1 }) + .limit(limit) + .toArray(), + db + .collection('interventions') + .find({ podId }, { projection: { _id: 0 } }) + .sort({ createdAt: -1 }) + .limit(limit) + .toArray(), + db + .collection('outcomes') + .find({ podId }, { projection: { _id: 0 } }) + .sort({ recordedAt: -1 }) + .limit(limit) + .toArray(), + ]); + + return [ + ...observations.map(observationEvent), + ...gitStates.map(gitEvent), + ...collisions.map(collisionEvent), + ...interventions.map(interventionEvent), + ...outcomes.map(outcomeEvent), + ] + .filter((event) => event.at !== new Date(0).toISOString()) + .sort((a, b) => Date.parse(b.at) - Date.parse(a.at)) + .slice(0, limit); +} diff --git a/backend/src/server.ts b/backend/src/server.ts index 7f4dc59..c1e1212 100644 --- a/backend/src/server.ts +++ b/backend/src/server.ts @@ -19,6 +19,7 @@ import { } from './pods/store.js'; import { getPresence, closeRoom } from './livekit/rooms.js'; import { loadPodGraph, reachFrom } from './graph/store.js'; +import { listPodActivity } from './activity/store.js'; import type { InterventionOutcome } from '@podman/shared'; const app = express(); @@ -159,6 +160,51 @@ app.get('/api/pods/:id/graph/reach/:node', async (req, res) => { } }); +app.get('/api/pods/:id/activity', async (req, res) => { + try { + const limit = Math.min(Number(req.query.limit ?? 80) || 80, 200); + res.json(await listPodActivity(req.params.id, limit)); + } catch (e) { + res.status(500).json({ error: (e as Error).message }); + } +}); + +app.get('/api/pods/:id/activity/stream', async (req, res) => { + res.setHeader('Content-Type', 'text/event-stream'); + res.setHeader('Cache-Control', 'no-cache, no-transform'); + res.setHeader('Connection', 'keep-alive'); + res.flushHeaders?.(); + + let closed = false; + let lastPayload = ''; + + const send = async () => { + if (closed) return; + try { + const events = await listPodActivity(req.params.id, 80); + const payload = JSON.stringify(events); + if (payload !== lastPayload) { + lastPayload = payload; + res.write(`event: snapshot\n`); + res.write(`data: ${payload}\n\n`); + } else { + res.write(`: keepalive ${Date.now()}\n\n`); + } + } catch (e) { + res.write(`event: error\n`); + res.write(`data: ${JSON.stringify({ error: (e as Error).message })}\n\n`); + } + }; + + await send(); + const interval = setInterval(() => void send(), 1500); + + req.on('close', () => { + closed = true; + clearInterval(interval); + }); +}); + const http = createServer(app); // ws relay: the agent pushes collision/intervention JSON here; PWAs subscribed by pod receive it. diff --git a/frontend/src/components/PodView.tsx b/frontend/src/components/PodView.tsx index 64123cb..95834fe 100644 --- a/frontend/src/components/PodView.tsx +++ b/frontend/src/components/PodView.tsx @@ -4,18 +4,22 @@ import { ArrowLeftIcon, CheckIcon, CircleDotIcon, + GitBranchIcon, ExternalLinkIcon, + FileTextIcon, MessageSquareIcon, MonitorUpIcon, RadioTowerIcon, SparklesIcon, + TriangleAlertIcon, Volume2Icon, XIcon, } from 'lucide-react'; import type { Room, RemoteTrack, RemoteTrackPublication, RemoteParticipant } from 'livekit-client'; -import type { Pod } from '@podman/shared'; +import type { Pod, PodActivityEvent, PodActivityKind } from '@podman/shared'; import { startBeat, type BeatHandle } from '../lib/beat.js'; import { useInterventions } from '../livekit/useInterventions.js'; +import { usePodActivity } from '../hooks/use-pod-activity.js'; import LiveWaveform from '@/components/ruixen/live-waveform'; import { Alert, AlertDescription, AlertTitle } from '@/components/ui/alert'; import { Avatar, AvatarBadge, AvatarFallback } from '@/components/ui/avatar'; @@ -83,6 +87,7 @@ export function PodView({ const [playingBeat, setPlayingBeat] = useState(false); const [note, setNote] = useState(null); const { active, hermes, voiceCue, actionUrl, respond } = useInterventions(room); + const activity = usePodActivity(team.id, me); const audioRef = useRef(null); const beatRef = useRef(null); @@ -285,6 +290,26 @@ export function PodView({

+ +
+ + +
+ {activity.error &&

{activity.error}

}