feat: runnersok

frontend/src/hooks/useRunStream.ts

@@ -0,0 +1,109 @@
+/**
+ * SSE subscriber hook for graph run execution.
+ * Connects to GET /api/runs/:id/stream and updates runStore.
+ */
+
+import { useEffect, useRef, useCallback } from 'react'
+import { useRunStore } from '@/lib/graph/runStore'
+import { dispatchCanvasCommand } from '@/app/canvas/canvasStore'
+
+export function useRunStream() {
+  const eventSourceRef = useRef<EventSource | null>(null)
+  const { startRun, setRunStatus, setNodeStatus, appendChunk } = useRunStore()
+
+  const disconnect = useCallback(() => {
+    if (eventSourceRef.current) {
+      eventSourceRef.current.close()
+      eventSourceRef.current = null
+    }
+  }, [])
+
+  const connectToRun = useCallback(
+    (runId: string) => {
+      disconnect()
+      startRun(runId)
+
+      const es = new EventSource(`/api/runs/${runId}/stream`)
+      eventSourceRef.current = es
+
+      es.addEventListener('run/started', (e) => {
+        const data = JSON.parse(e.data)
+        setRunStatus('running')
+        // Initialize all nodes as pending so overlays appear immediately
+        const nodeIds = data.nodeIds as string[] | undefined
+        if (nodeIds) {
+          for (const nodeId of nodeIds) {
+            setNodeStatus(nodeId, { status: 'pending' })
+          }
+        }
+      })
+
+      es.addEventListener('run/completed', () => {
+        setRunStatus('completed')
+        es.close()
+      })
+
+      es.addEventListener('run/failed', (e) => {
+        const data = JSON.parse(e.data)
+        setRunStatus('failed', data.error)
+        es.close()
+      })
+
+      es.addEventListener('step/started', (e) => {
+        const data = JSON.parse(e.data)
+        setNodeStatus(data.nodeId, { status: 'running' })
+        // Fire trail animation along edges leading to this node
+        dispatchCanvasCommand({ type: 'path/addTrigger', payload: data.nodeId })
+      })
+
+      es.addEventListener('step/completed', (e) => {
+        const data = JSON.parse(e.data)
+        setNodeStatus(data.nodeId, { status: 'completed', output: data.output })
+      })
+
+      es.addEventListener('step/failed', (e) => {
+        const data = JSON.parse(e.data)
+        setNodeStatus(data.nodeId, { status: 'failed', error: data.error })
+      })
+
+      es.addEventListener('step/chunk', (e) => {
+        const data = JSON.parse(e.data)
+        appendChunk(data.nodeId, data.chunk)
+      })
+
+      es.onerror = () => {
+        setRunStatus('failed', 'Connection lost')
+        es.close()
+      }
+    },
+    [disconnect, startRun, setRunStatus, setNodeStatus, appendChunk]
+  )
+
+  // Cleanup on unmount
+  useEffect(() => disconnect, [disconnect])
+
+  return { connectToRun, disconnect }
+}
+
+/**
+ * Trigger a new run: POST the graph, then connect SSE.
+ */
+export async function createAndStreamRun(
+  recollectionId: string,
+  graph: { nodes: unknown[]; edges: unknown[] },
+  connectToRun: (runId: string) => void
+): Promise<void> {
+  const res = await fetch('/api/runs', {
+    method: 'POST',
+    headers: { 'Content-Type': 'application/json' },
+    body: JSON.stringify({ recollectionId, graph }),
+  })
+
+  if (!res.ok) {
+    const err = await res.json().catch(() => ({ error: 'Failed to create run' }))
+    throw new Error(err.error ?? 'Failed to create run')
+  }
+
+  const { id } = await res.json()
+  connectToRun(id)
+}