CUB-125: implement real-time SSE/WebSocket in React frontend #46
1129
frontend/package-lock.json
generated
1129
frontend/package-lock.json
generated
File diff suppressed because it is too large
Load Diff
@@ -7,7 +7,9 @@
|
||||
"dev": "vite",
|
||||
"build": "tsc -b && vite build",
|
||||
"lint": "eslint .",
|
||||
"preview": "vite preview"
|
||||
"preview": "vite preview",
|
||||
"test": "vitest run",
|
||||
"test:watch": "vitest"
|
||||
},
|
||||
"dependencies": {
|
||||
"@tanstack/react-query": "^5.100.9",
|
||||
@@ -20,6 +22,8 @@
|
||||
"devDependencies": {
|
||||
"@eslint/js": "^10.0.1",
|
||||
"@tailwindcss/vite": "^4.2.4",
|
||||
"@testing-library/jest-dom": "^6.9.1",
|
||||
"@testing-library/react": "^16.3.2",
|
||||
"@types/node": "^24.12.2",
|
||||
"@types/react": "^19.2.14",
|
||||
"@types/react-dom": "^19.2.3",
|
||||
@@ -29,10 +33,12 @@
|
||||
"eslint-plugin-react-hooks": "^7.1.1",
|
||||
"eslint-plugin-react-refresh": "^0.5.2",
|
||||
"globals": "^17.5.0",
|
||||
"jsdom": "^29.1.1",
|
||||
"postcss": "^8.5.14",
|
||||
"tailwindcss": "^4.2.4",
|
||||
"typescript": "~6.0.2",
|
||||
"typescript-eslint": "^8.58.2",
|
||||
"vite": "^8.0.10"
|
||||
"vite": "^8.0.10",
|
||||
"vitest": "^4.1.7"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
import { useState } from 'react'
|
||||
import { NavLink } from 'react-router-dom'
|
||||
import { Command, Activity, FolderKanban, Monitor, Settings, Menu, X } from 'lucide-react'
|
||||
import { Command, Activity, FolderKanban, Monitor, Settings, Menu, X, Wifi, WifiOff, Loader } from 'lucide-react'
|
||||
import { useSSEContext } from '../contexts/SSEContext'
|
||||
import type { SSEStatus } from '../hooks/useSSE'
|
||||
|
||||
const navItems = [
|
||||
{ to: '/', icon: Command, label: 'Hub' },
|
||||
@@ -10,9 +12,29 @@ const navItems = [
|
||||
{ to: '/settings', icon: Settings, label: 'Settings' },
|
||||
]
|
||||
|
||||
/** Small status pill shown in the sidebar footer and mobile header. */
|
||||
function SSEStatusBadge({ status, showLabel = false }: { status: SSEStatus; showLabel?: boolean }) {
|
||||
const cfg = {
|
||||
connected: { icon: Wifi, color: 'text-green-500', label: 'Live' },
|
||||
connecting: { icon: Loader, color: 'text-yellow-500 animate-spin', label: 'Connecting' },
|
||||
reconnecting: { icon: Loader, color: 'text-yellow-500 animate-spin', label: 'Reconnecting' },
|
||||
error: { icon: WifiOff, color: 'text-red-500', label: 'Disconnected' },
|
||||
}[status]
|
||||
|
||||
const Icon = cfg.icon
|
||||
|
||||
return (
|
||||
<div className="flex items-center gap-1.5" title={cfg.label}>
|
||||
<Icon size={14} className={cfg.color} />
|
||||
{showLabel && <span className={`text-xs ${cfg.color}`}>{cfg.label}</span>}
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
export default function Layout({ children }: { children: React.ReactNode }) {
|
||||
const [expanded, setExpanded] = useState(false)
|
||||
const [mobileOpen, setMobileOpen] = useState(false)
|
||||
const { sseStatus } = useSSEContext()
|
||||
|
||||
return (
|
||||
<div className="flex min-h-screen bg-surface-darkest text-on-surface">
|
||||
@@ -46,6 +68,15 @@ export default function Layout({ children }: { children: React.ReactNode }) {
|
||||
</NavLink>
|
||||
))}
|
||||
</nav>
|
||||
{/* SSE connection status — footer of sidebar */}
|
||||
<div className="px-4 py-3 border-t border-surface-light flex items-center gap-2">
|
||||
<SSEStatusBadge status={sseStatus} />
|
||||
{expanded && (
|
||||
<span className="text-xs text-on-surface-muted whitespace-nowrap">
|
||||
{sseStatus === 'connected' ? 'Live updates on' : sseStatus}
|
||||
</span>
|
||||
)}
|
||||
</div>
|
||||
</aside>
|
||||
|
||||
{/* Mobile Header + Bottom Nav */}
|
||||
@@ -54,6 +85,7 @@ export default function Layout({ children }: { children: React.ReactNode }) {
|
||||
<div className="flex items-center gap-2">
|
||||
<Command size={22} className="text-primary" />
|
||||
<span className="font-bold">Control Center</span>
|
||||
<SSEStatusBadge status={sseStatus} />
|
||||
</div>
|
||||
<button onClick={() => setMobileOpen(!mobileOpen)} className="p-2">
|
||||
{mobileOpen ? <X size={22} /> : <Menu size={22} />}
|
||||
|
||||
23
frontend/src/contexts/SSEContext.tsx
Normal file
23
frontend/src/contexts/SSEContext.tsx
Normal file
@@ -0,0 +1,23 @@
|
||||
/**
|
||||
* SSEContext — provides SSE connection status throughout the component tree.
|
||||
* Mount <SSEProvider> once inside QueryClientProvider.
|
||||
*/
|
||||
import { createContext, useContext, type ReactNode } from 'react'
|
||||
import { useRealtimeSync } from '../hooks/useRealtimeSync'
|
||||
import type { SSEStatus } from '../hooks/useSSE'
|
||||
|
||||
interface SSEContextValue {
|
||||
sseStatus: SSEStatus
|
||||
}
|
||||
|
||||
const SSEContext = createContext<SSEContextValue>({ sseStatus: 'connecting' })
|
||||
|
||||
export function SSEProvider({ children }: { children: ReactNode }) {
|
||||
const { sseStatus } = useRealtimeSync()
|
||||
return <SSEContext.Provider value={{ sseStatus }}>{children}</SSEContext.Provider>
|
||||
}
|
||||
|
||||
/** Access the SSE connection status from any component. */
|
||||
export function useSSEContext(): SSEContextValue {
|
||||
return useContext(SSEContext)
|
||||
}
|
||||
129
frontend/src/hooks/useRealtimeSync.test.tsx
Normal file
129
frontend/src/hooks/useRealtimeSync.test.tsx
Normal file
@@ -0,0 +1,129 @@
|
||||
/**
|
||||
* Tests for useRealtimeSync — event → query invalidation mapping.
|
||||
*
|
||||
* Uses .tsx extension so Vite/OXC can parse JSX in the wrapper component.
|
||||
*/
|
||||
import { renderHook } from '@testing-library/react'
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import { QueryClient, QueryClientProvider } from '@tanstack/react-query'
|
||||
import * as useSSEModule from './useSSE'
|
||||
import { useRealtimeSync } from './useRealtimeSync'
|
||||
import React from 'react'
|
||||
import type { SSEMessage } from '../services/sse'
|
||||
|
||||
describe('useRealtimeSync', () => {
|
||||
let queryClient: QueryClient
|
||||
let mockSSEOnMessage: ((msg: { type: string; data: unknown }) => void) | null = null
|
||||
|
||||
beforeEach(() => {
|
||||
queryClient = new QueryClient({
|
||||
defaultOptions: { queries: { retry: false } },
|
||||
})
|
||||
mockSSEOnMessage = null
|
||||
|
||||
// Spy on useSSE to capture the onMessage callback
|
||||
vi.spyOn(useSSEModule, 'useSSE').mockImplementation((opts) => {
|
||||
mockSSEOnMessage = opts?.onMessage ?? null
|
||||
return { status: 'connected' }
|
||||
})
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks()
|
||||
})
|
||||
|
||||
function render() {
|
||||
return renderHook(() => useRealtimeSync(), {
|
||||
wrapper: ({ children }: { children: React.ReactNode }) => (
|
||||
React.createElement(QueryClientProvider, { client: queryClient }, children)
|
||||
),
|
||||
})
|
||||
}
|
||||
|
||||
it('invalidates ["agents"] on agent.status event', async () => {
|
||||
const invalidateSpy = vi.spyOn(queryClient, 'invalidateQueries')
|
||||
render()
|
||||
|
||||
const msg: SSEMessage = {
|
||||
type: 'agent.status',
|
||||
data: { agentId: 'a1', status: 'active' },
|
||||
}
|
||||
mockSSEOnMessage!(msg)
|
||||
|
||||
expect(invalidateSpy).toHaveBeenCalledWith({ queryKey: ['agents'] })
|
||||
expect(invalidateSpy).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('invalidates ["tasks"] and ["agents"] on agent.task event', async () => {
|
||||
const invalidateSpy = vi.spyOn(queryClient, 'invalidateQueries')
|
||||
render()
|
||||
|
||||
const msg: SSEMessage = {
|
||||
type: 'agent.task',
|
||||
data: { agentId: 'a1', taskId: 't1', title: 'Test', action: 'assigned' },
|
||||
}
|
||||
mockSSEOnMessage!(msg)
|
||||
|
||||
expect(invalidateSpy).toHaveBeenCalledWith({ queryKey: ['tasks'] })
|
||||
expect(invalidateSpy).toHaveBeenCalledWith({ queryKey: ['agents'] })
|
||||
expect(invalidateSpy).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
|
||||
it('invalidates ["tasks"] and ["agents"] on agent.progress event', async () => {
|
||||
const invalidateSpy = vi.spyOn(queryClient, 'invalidateQueries')
|
||||
render()
|
||||
|
||||
const msg: SSEMessage = {
|
||||
type: 'agent.progress',
|
||||
data: { agentId: 'a1', taskId: 't1', progress: 50, message: 'working' },
|
||||
}
|
||||
mockSSEOnMessage!(msg)
|
||||
|
||||
expect(invalidateSpy).toHaveBeenCalledWith({ queryKey: ['tasks'] })
|
||||
expect(invalidateSpy).toHaveBeenCalledWith({ queryKey: ['agents'] })
|
||||
expect(invalidateSpy).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
|
||||
it('invalidates ["agents"], ["sessions"], ["tasks"] on fleet.update event', async () => {
|
||||
const invalidateSpy = vi.spyOn(queryClient, 'invalidateQueries')
|
||||
render()
|
||||
|
||||
const msg: SSEMessage = {
|
||||
type: 'fleet.update',
|
||||
data: { timestamp: '2026-05-20T12:00:00Z', agentCount: 5 },
|
||||
}
|
||||
mockSSEOnMessage!(msg)
|
||||
|
||||
expect(invalidateSpy).toHaveBeenCalledWith({ queryKey: ['agents'] })
|
||||
expect(invalidateSpy).toHaveBeenCalledWith({ queryKey: ['sessions'] })
|
||||
expect(invalidateSpy).toHaveBeenCalledWith({ queryKey: ['tasks'] })
|
||||
expect(invalidateSpy).toHaveBeenCalledTimes(3)
|
||||
})
|
||||
|
||||
it('does nothing on connected event', async () => {
|
||||
const invalidateSpy = vi.spyOn(queryClient, 'invalidateQueries')
|
||||
render()
|
||||
|
||||
const msg: SSEMessage = {
|
||||
type: 'connected',
|
||||
data: { clientCount: 1 },
|
||||
}
|
||||
mockSSEOnMessage!(msg)
|
||||
|
||||
expect(invalidateSpy).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('does nothing on unknown event types', async () => {
|
||||
const invalidateSpy = vi.spyOn(queryClient, 'invalidateQueries')
|
||||
render()
|
||||
|
||||
mockSSEOnMessage!({ type: 'unknown.event', data: {} })
|
||||
|
||||
expect(invalidateSpy).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('returns sseStatus from useSSE', () => {
|
||||
const { result } = render()
|
||||
expect(result.current.sseStatus).toBe('connected')
|
||||
})
|
||||
})
|
||||
64
frontend/src/hooks/useRealtimeSync.ts
Normal file
64
frontend/src/hooks/useRealtimeSync.ts
Normal file
@@ -0,0 +1,64 @@
|
||||
/**
|
||||
* useRealtimeSync — mounts the SSE connection once at the app level and
|
||||
* wires incoming events to React Query cache invalidation.
|
||||
*
|
||||
* Event → query key mapping:
|
||||
* agent.status → ['agents']
|
||||
* agent.task → ['tasks'], ['agents']
|
||||
* agent.progress → ['tasks'], ['agents']
|
||||
* fleet.update → ['agents'], ['sessions'], ['tasks']
|
||||
*/
|
||||
import { useQueryClient } from '@tanstack/react-query'
|
||||
import { useCallback } from 'react'
|
||||
import { useSSE, type SSEStatus } from './useSSE'
|
||||
import type { SSEMessage } from '../services/sse'
|
||||
|
||||
export function useRealtimeSync(): { sseStatus: SSEStatus } {
|
||||
const queryClient = useQueryClient()
|
||||
|
||||
const handleMessage = useCallback(
|
||||
(raw: { type: string; data: unknown }) => {
|
||||
// Cast to discriminated union — the backend contract guarantees these shapes
|
||||
const msg = raw as SSEMessage
|
||||
|
||||
switch (msg.type) {
|
||||
case 'agent.status':
|
||||
// msg.data: AgentStatusEvent { agentId, status, reason? }
|
||||
void msg.data.agentId // retained for type-narrowing — ensures payload matches contract
|
||||
queryClient.invalidateQueries({ queryKey: ['agents'] })
|
||||
break
|
||||
|
||||
case 'agent.task':
|
||||
// msg.data: AgentTaskEvent { agentId, taskId, title, action }
|
||||
void msg.data.agentId
|
||||
queryClient.invalidateQueries({ queryKey: ['tasks'] })
|
||||
queryClient.invalidateQueries({ queryKey: ['agents'] })
|
||||
break
|
||||
|
||||
case 'agent.progress':
|
||||
// msg.data: AgentProgressEvent { agentId, taskId, progress, message? }
|
||||
void msg.data.agentId
|
||||
queryClient.invalidateQueries({ queryKey: ['tasks'] })
|
||||
queryClient.invalidateQueries({ queryKey: ['agents'] })
|
||||
break
|
||||
|
||||
case 'fleet.update':
|
||||
// msg.data: FleetUpdateEvent { timestamp, agentCount }
|
||||
void msg.data.agentCount
|
||||
queryClient.invalidateQueries({ queryKey: ['agents'] })
|
||||
queryClient.invalidateQueries({ queryKey: ['sessions'] })
|
||||
queryClient.invalidateQueries({ queryKey: ['tasks'] })
|
||||
break
|
||||
|
||||
default:
|
||||
// 'connected' and unknown events — no action needed
|
||||
break
|
||||
}
|
||||
},
|
||||
[queryClient],
|
||||
)
|
||||
|
||||
const { status: sseStatus } = useSSE({ onMessage: handleMessage })
|
||||
|
||||
return { sseStatus }
|
||||
}
|
||||
267
frontend/src/hooks/useSSE.test.ts
Normal file
267
frontend/src/hooks/useSSE.test.ts
Normal file
@@ -0,0 +1,267 @@
|
||||
/**
|
||||
* Tests for useSSE — SSE connection lifecycle, back-off, event parsing, and cleanup.
|
||||
*
|
||||
* jsdom does not include EventSource, so we mock it completely.
|
||||
*/
|
||||
import { renderHook, act, waitFor } from '@testing-library/react'
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import { useSSE } from './useSSE'
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Mock EventSource — defined as a plain class so `new EventSource()` works
|
||||
// ---------------------------------------------------------------------------
|
||||
class MockEventSource {
|
||||
url: string
|
||||
onopen: (() => void) | null = null
|
||||
onerror: ((evt: Event) => void) | null = null
|
||||
onmessage: ((evt: MessageEvent) => void) | null = null
|
||||
private listeners: Map<string, Array<(evt: Event) => void>> = new Map()
|
||||
readyState: number = 0
|
||||
|
||||
constructor(url: string) {
|
||||
this.url = url
|
||||
}
|
||||
|
||||
addEventListener(type: string, handler: (evt: Event) => void) {
|
||||
if (!this.listeners.has(type)) this.listeners.set(type, [])
|
||||
this.listeners.get(type)!.push(handler)
|
||||
}
|
||||
|
||||
removeEventListener() { /* no-op for tests */ }
|
||||
|
||||
close() {
|
||||
this.readyState = 2
|
||||
this.onopen = null
|
||||
this.onerror = null
|
||||
this.onmessage = null
|
||||
this.listeners.clear()
|
||||
}
|
||||
|
||||
// Test helpers
|
||||
_simulateOpen() { this.onopen?.() }
|
||||
_simulateError() { this.onerror?.(new Event('error')) }
|
||||
_simulateNamedEvent(type: string, data: string) {
|
||||
const handlers = this.listeners.get(type)
|
||||
if (handlers) {
|
||||
const evt = new MessageEvent(type, { data }) as Event
|
||||
handlers.forEach((h) => h(evt))
|
||||
}
|
||||
}
|
||||
_simulateMessage(data: string) {
|
||||
this.onmessage?.(new MessageEvent('message', { data }) as MessageEvent)
|
||||
}
|
||||
|
||||
static readonly CONNECTING = 0
|
||||
static readonly OPEN = 1
|
||||
static readonly CLOSED = 2
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Tests
|
||||
// ---------------------------------------------------------------------------
|
||||
let esInstances: MockEventSource[]
|
||||
|
||||
describe('useSSE', () => {
|
||||
beforeEach(() => {
|
||||
esInstances = []
|
||||
// Replace global EventSource with our mock class
|
||||
Object.defineProperty(globalThis, 'EventSource', {
|
||||
// The mock must use a class for `new EventSource()` to work
|
||||
value: class extends MockEventSource {
|
||||
constructor(url: string) {
|
||||
super(url)
|
||||
esInstances.push(this)
|
||||
}
|
||||
},
|
||||
writable: true,
|
||||
configurable: true,
|
||||
})
|
||||
vi.useFakeTimers({ shouldAdvanceTime: true })
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks()
|
||||
vi.useRealTimers()
|
||||
})
|
||||
|
||||
// ── Initial connection ──────────────────────────────────────────────────
|
||||
it('starts in "connecting" state and creates an EventSource', () => {
|
||||
const { result } = renderHook(() => useSSE({ url: '/api/events' }))
|
||||
|
||||
expect(result.current.status).toBe('connecting')
|
||||
expect(esInstances.length).toBeGreaterThanOrEqual(1)
|
||||
expect(esInstances[0].url).toBe('/api/events')
|
||||
})
|
||||
|
||||
it('transitions to "connected" on open', async () => {
|
||||
const onOpen = vi.fn()
|
||||
const { result } = renderHook(() => useSSE({ url: '/api/events', onOpen }))
|
||||
|
||||
act(() => { esInstances[0]._simulateOpen() })
|
||||
|
||||
await waitFor(() => {
|
||||
expect(result.current.status).toBe('connected')
|
||||
})
|
||||
expect(onOpen).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
// ── Reconnection with exponential back-off ──────────────────────────────
|
||||
it('retries after error with exponential back-off', async () => {
|
||||
const { result } = renderHook(() =>
|
||||
useSSE({ url: '/api/events', reconnectBaseMs: 1000, reconnectMaxMs: 30000 }),
|
||||
)
|
||||
|
||||
// First error → reconnecting, retry at 1s
|
||||
act(() => { esInstances[0]._simulateError() })
|
||||
await waitFor(() => { expect(result.current.status).toBe('reconnecting') })
|
||||
expect(esInstances).toHaveLength(1)
|
||||
|
||||
// Advance 1000ms → second EventSource created
|
||||
act(() => { vi.advanceTimersByTime(1000) })
|
||||
expect(esInstances).toHaveLength(2)
|
||||
|
||||
// Second error → reconnecting, retry at 2s
|
||||
act(() => { esInstances[1]._simulateError() })
|
||||
await waitFor(() => { expect(result.current.status).toBe('reconnecting') })
|
||||
act(() => { vi.advanceTimersByTime(2000) })
|
||||
expect(esInstances).toHaveLength(3)
|
||||
|
||||
// Third error → reconnecting, retry at 4s
|
||||
act(() => { esInstances[2]._simulateError() })
|
||||
act(() => { vi.advanceTimersByTime(4000) })
|
||||
expect(esInstances).toHaveLength(4)
|
||||
})
|
||||
|
||||
it('caps reconnect delay at reconnectMaxMs', async () => {
|
||||
renderHook(() =>
|
||||
useSSE({ url: '/api/events', reconnectBaseMs: 1000, reconnectMaxMs: 10000 }),
|
||||
)
|
||||
|
||||
// Force 5 errors to push the exponent past the cap
|
||||
for (let i = 0; i < 5; i++) {
|
||||
act(() => { esInstances[i]._simulateError() })
|
||||
const expectedDelay = Math.min(1000 * 2 ** i, 10000)
|
||||
act(() => { vi.advanceTimersByTime(expectedDelay) })
|
||||
}
|
||||
|
||||
// 6 ES instances created (initial + 5 retries)
|
||||
expect(esInstances).toHaveLength(6)
|
||||
})
|
||||
|
||||
// ── Circuit-breaker (max retries) ───────────────────────────────────────
|
||||
it('transitions to "error" after reconnectLimit is exceeded', async () => {
|
||||
const { result } = renderHook(() =>
|
||||
useSSE({ url: '/api/events', reconnectBaseMs: 100, reconnectLimit: 2 }),
|
||||
)
|
||||
|
||||
// First error → reconnecting
|
||||
act(() => { esInstances[0]._simulateError() })
|
||||
await waitFor(() => { expect(result.current.status).toBe('reconnecting') })
|
||||
|
||||
// Advance → retry
|
||||
act(() => { vi.advanceTimersByTime(100) })
|
||||
|
||||
// Second error → reconnecting (attempt 2, still ≤ limit)
|
||||
act(() => { esInstances[1]._simulateError() })
|
||||
await waitFor(() => { expect(result.current.status).toBe('reconnecting') })
|
||||
act(() => { vi.advanceTimersByTime(200) })
|
||||
|
||||
// Third error → limit exceeded (3 > 2) → error
|
||||
act(() => { esInstances[2]._simulateError() })
|
||||
await waitFor(() => { expect(result.current.status).toBe('error') })
|
||||
})
|
||||
|
||||
it('resets reconnect counter on successful connection', async () => {
|
||||
const { result } = renderHook(() =>
|
||||
useSSE({ url: '/api/events', reconnectBaseMs: 100, reconnectLimit: 3 }),
|
||||
)
|
||||
|
||||
// Two errors then a successful connect
|
||||
act(() => { esInstances[0]._simulateError() })
|
||||
act(() => { vi.advanceTimersByTime(100) })
|
||||
|
||||
act(() => { esInstances[1]._simulateOpen() })
|
||||
await waitFor(() => { expect(result.current.status).toBe('connected') })
|
||||
|
||||
// Now error again — counter should be reset, so we get fresh attempts
|
||||
act(() => { esInstances[1]._simulateError() })
|
||||
await waitFor(() => { expect(result.current.status).toBe('reconnecting') })
|
||||
expect(result.current.status).toBe('reconnecting')
|
||||
})
|
||||
|
||||
// ── Cleanup on unmount ───────────────────────────────────────────────────
|
||||
it('closes EventSource on unmount', () => {
|
||||
const closeSpy = vi.spyOn(MockEventSource.prototype, 'close')
|
||||
const { unmount } = renderHook(() => useSSE({ url: '/api/events' }))
|
||||
|
||||
unmount()
|
||||
expect(closeSpy).toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('does not update state after unmount', async () => {
|
||||
const { result, unmount } = renderHook(() => useSSE({ url: '/api/events' }))
|
||||
|
||||
unmount()
|
||||
|
||||
// These should be no-ops after unmount (mountedRef guards)
|
||||
act(() => { esInstances[0]._simulateOpen() })
|
||||
act(() => { esInstances[0]._simulateError() })
|
||||
|
||||
// State should not have changed
|
||||
expect(result.current.status).toBe('connecting')
|
||||
})
|
||||
|
||||
// ── Event parsing ───────────────────────────────────────────────────────
|
||||
it('parses valid JSON data into objects', async () => {
|
||||
const onMessage = vi.fn()
|
||||
renderHook(() => useSSE({ url: '/api/events', onMessage }))
|
||||
|
||||
act(() => {
|
||||
esInstances[0]._simulateNamedEvent('agent.status', JSON.stringify({ agentId: 'a1', status: 'active' }))
|
||||
})
|
||||
|
||||
expect(onMessage).toHaveBeenCalledWith({
|
||||
type: 'agent.status',
|
||||
data: { agentId: 'a1', status: 'active' },
|
||||
})
|
||||
})
|
||||
|
||||
it('passes invalid JSON through as raw string', async () => {
|
||||
const onMessage = vi.fn()
|
||||
renderHook(() => useSSE({ url: '/api/events', onMessage }))
|
||||
|
||||
act(() => {
|
||||
esInstances[0]._simulateNamedEvent('agent.status', 'not valid json {{{')
|
||||
})
|
||||
|
||||
expect(onMessage).toHaveBeenCalledWith({
|
||||
type: 'agent.status',
|
||||
data: 'not valid json {{{',
|
||||
})
|
||||
})
|
||||
|
||||
// ── enabled=false skips connection ──────────────────────────────────────
|
||||
it('does not create EventSource when enabled=false', () => {
|
||||
const { result } = renderHook(() => useSSE({ url: '/api/events', enabled: false }))
|
||||
|
||||
expect(esInstances).toHaveLength(0)
|
||||
expect(result.current.status).toBe('connecting')
|
||||
})
|
||||
|
||||
// ── onError callback ────────────────────────────────────────────────────
|
||||
it('calls onError on connection failure', async () => {
|
||||
const onError = vi.fn()
|
||||
renderHook(() =>
|
||||
useSSE({ url: '/api/events', onError, reconnectBaseMs: 100 }),
|
||||
)
|
||||
|
||||
act(() => { esInstances[0]._simulateError() })
|
||||
expect(onError).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
// ── Default URL ─────────────────────────────────────────────────────────
|
||||
it('uses /api/events as default URL', () => {
|
||||
renderHook(() => useSSE())
|
||||
expect(esInstances[0].url).toBe('/api/events')
|
||||
})
|
||||
})
|
||||
180
frontend/src/hooks/useSSE.ts
Normal file
180
frontend/src/hooks/useSSE.ts
Normal file
@@ -0,0 +1,180 @@
|
||||
import { useEffect, useRef, useCallback, useState } from 'react'
|
||||
|
||||
/** SSE connection state reported to consumers. */
|
||||
export type SSEStatus = 'connecting' | 'connected' | 'reconnecting' | 'error'
|
||||
|
||||
/** Typed SSE event received from the backend. */
|
||||
export interface SSEMessage {
|
||||
/** event: field from the SSE frame */
|
||||
type: string
|
||||
/** parsed JSON from the data: field */
|
||||
data: unknown
|
||||
}
|
||||
|
||||
export interface UseSSEOptions {
|
||||
/** Endpoint URL — defaults to /api/events */
|
||||
url?: string
|
||||
/** Called for every SSE message (all event types) */
|
||||
onMessage?: (msg: SSEMessage) => void
|
||||
/** Called when connection opens or reconnects */
|
||||
onOpen?: () => void
|
||||
/** Called on every connection error (both transient and terminal) */
|
||||
onError?: (err: Event) => void
|
||||
/** Base delay in ms before the first reconnect attempt (default 1 000) */
|
||||
reconnectBaseMs?: number
|
||||
/** Maximum reconnect delay in ms (default 30 000) */
|
||||
reconnectMaxMs?: number
|
||||
/**
|
||||
* Maximum number of consecutive reconnect attempts before giving up.
|
||||
* When the limit is reached, status transitions to 'error'.
|
||||
* Default undefined (unlimited).
|
||||
*/
|
||||
reconnectLimit?: number
|
||||
/** Set false to disable auto-connect (useful in tests) */
|
||||
enabled?: boolean
|
||||
}
|
||||
|
||||
const SSE_EVENTS = ['agent.status', 'agent.task', 'agent.progress', 'fleet.update', 'connected'] as const
|
||||
|
||||
/**
|
||||
* useSSE — mounts a persistent SSE connection to the Control Center backend.
|
||||
*
|
||||
* Handles:
|
||||
* - Initial connection on mount
|
||||
* - Exponential back-off reconnection on drop (1s → 2s → 4s … capped at reconnectMaxMs)
|
||||
* - Circuit-breaker: after reconnectLimit consecutive failures, transitions to 'error'
|
||||
* - Cleanup on unmount
|
||||
* - All five event types: agent.status, agent.task, agent.progress, fleet.update, connected
|
||||
*
|
||||
* The 'connected' SSE event is an application-level handshake sent by the backend
|
||||
* after the transport opens. This is distinct from onOpen, which fires at the
|
||||
* transport level when the EventSource HTTP connection is established.
|
||||
*/
|
||||
export function useSSE({
|
||||
url = '/api/events',
|
||||
onMessage,
|
||||
onOpen,
|
||||
onError,
|
||||
reconnectBaseMs = 1_000,
|
||||
reconnectMaxMs = 30_000,
|
||||
reconnectLimit,
|
||||
enabled = true,
|
||||
}: UseSSEOptions = {}): { status: SSEStatus } {
|
||||
const [status, setStatus] = useState<SSEStatus>('connecting')
|
||||
|
||||
// Stable refs so the effect doesn't need to re-run when callbacks change
|
||||
const onMessageRef = useRef(onMessage)
|
||||
const onOpenRef = useRef(onOpen)
|
||||
const onErrorRef = useRef(onError)
|
||||
onMessageRef.current = onMessage
|
||||
onOpenRef.current = onOpen
|
||||
onErrorRef.current = onError
|
||||
|
||||
const reconnectAttemptRef = useRef(0)
|
||||
const reconnectTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null)
|
||||
const esRef = useRef<EventSource | null>(null)
|
||||
const mountedRef = useRef(true)
|
||||
|
||||
const clearReconnectTimer = useCallback(() => {
|
||||
if (reconnectTimerRef.current !== null) {
|
||||
clearTimeout(reconnectTimerRef.current)
|
||||
reconnectTimerRef.current = null
|
||||
}
|
||||
}, [])
|
||||
|
||||
const connect = useCallback(() => {
|
||||
if (!mountedRef.current || !enabled) return
|
||||
|
||||
// Clean up any existing connection
|
||||
if (esRef.current) {
|
||||
esRef.current.close()
|
||||
esRef.current = null
|
||||
}
|
||||
|
||||
setStatus(reconnectAttemptRef.current === 0 ? 'connecting' : 'reconnecting')
|
||||
|
||||
const es = new EventSource(url)
|
||||
esRef.current = es
|
||||
|
||||
es.onopen = () => {
|
||||
if (!mountedRef.current) return
|
||||
reconnectAttemptRef.current = 0
|
||||
setStatus('connected')
|
||||
onOpenRef.current?.()
|
||||
}
|
||||
|
||||
es.onerror = (evt) => {
|
||||
if (!mountedRef.current) return
|
||||
|
||||
// EventSource auto-retries but we manage our own to get back-off control
|
||||
es.close()
|
||||
esRef.current = null
|
||||
|
||||
onErrorRef.current?.(evt)
|
||||
|
||||
reconnectAttemptRef.current += 1
|
||||
|
||||
// Circuit-breaker: give up after reconnectLimit consecutive failures
|
||||
if (reconnectLimit !== undefined && reconnectAttemptRef.current > reconnectLimit) {
|
||||
setStatus('error')
|
||||
return
|
||||
}
|
||||
|
||||
// Exponential back-off: 1s, 2s, 4s … capped at reconnectMaxMs
|
||||
// Note: attempt is 1-based here (already incremented), so we use attempt-1 for the exponent
|
||||
const delay = Math.min(
|
||||
reconnectBaseMs * 2 ** (reconnectAttemptRef.current - 1),
|
||||
reconnectMaxMs,
|
||||
)
|
||||
setStatus('reconnecting')
|
||||
|
||||
clearReconnectTimer()
|
||||
reconnectTimerRef.current = setTimeout(() => {
|
||||
if (mountedRef.current) connect()
|
||||
}, delay)
|
||||
}
|
||||
|
||||
// Register listeners for all known event types
|
||||
for (const eventType of SSE_EVENTS) {
|
||||
es.addEventListener(eventType, (evt: MessageEvent) => {
|
||||
if (!mountedRef.current) return
|
||||
let data: unknown = evt.data
|
||||
try {
|
||||
data = JSON.parse(evt.data as string)
|
||||
} catch {
|
||||
// leave as raw string
|
||||
}
|
||||
onMessageRef.current?.({ type: eventType, data })
|
||||
})
|
||||
}
|
||||
|
||||
// Catch-all for unnamed events (type == 'message').
|
||||
// Won't fire for the named events registered via addEventListener above.
|
||||
es.onmessage = (evt: MessageEvent) => {
|
||||
if (!mountedRef.current) return
|
||||
let data: unknown = evt.data
|
||||
try {
|
||||
data = JSON.parse(evt.data as string)
|
||||
} catch {
|
||||
// leave as raw string
|
||||
}
|
||||
onMessageRef.current?.({ type: 'message', data })
|
||||
}
|
||||
}, [url, enabled, reconnectBaseMs, reconnectMaxMs, reconnectLimit, clearReconnectTimer])
|
||||
|
||||
useEffect(() => {
|
||||
mountedRef.current = true
|
||||
if (enabled) connect()
|
||||
|
||||
return () => {
|
||||
mountedRef.current = false
|
||||
clearReconnectTimer()
|
||||
if (esRef.current) {
|
||||
esRef.current.close()
|
||||
esRef.current = null
|
||||
}
|
||||
}
|
||||
}, [connect, enabled, clearReconnectTimer])
|
||||
|
||||
return { status }
|
||||
}
|
||||
@@ -4,13 +4,16 @@ import { QueryClient, QueryClientProvider } from '@tanstack/react-query'
|
||||
import { BrowserRouter } from 'react-router-dom'
|
||||
import ErrorBoundary from './components/ErrorBoundary'
|
||||
import { ThemeProvider } from './hooks/useTheme'
|
||||
import { SSEProvider } from './contexts/SSEContext'
|
||||
import './index.css'
|
||||
import App from './App'
|
||||
|
||||
const queryClient = new QueryClient({
|
||||
defaultOptions: {
|
||||
queries: {
|
||||
staleTime: 30_000,
|
||||
// No polling — real-time updates come through SSE.
|
||||
// staleTime is kept high; data is pushed, not pulled.
|
||||
staleTime: 60_000,
|
||||
refetchOnWindowFocus: false,
|
||||
retry: 1,
|
||||
},
|
||||
@@ -22,9 +25,13 @@ createRoot(document.getElementById('root')!).render(
|
||||
<ErrorBoundary>
|
||||
<ThemeProvider>
|
||||
<QueryClientProvider client={queryClient}>
|
||||
{/* SSEProvider must live inside QueryClientProvider so it can call
|
||||
useQueryClient() to invalidate caches on incoming events. */}
|
||||
<SSEProvider>
|
||||
<BrowserRouter>
|
||||
<App />
|
||||
</BrowserRouter>
|
||||
</SSEProvider>
|
||||
</QueryClientProvider>
|
||||
</ThemeProvider>
|
||||
</ErrorBoundary>
|
||||
|
||||
@@ -1,18 +1,36 @@
|
||||
import { useTheme } from '../hooks/useTheme'
|
||||
import { useLocalStorage } from '../hooks/useLocalStorage'
|
||||
import { Sun, Moon, Monitor, Zap, Clock } from 'lucide-react'
|
||||
import { useSSEContext } from '../contexts/SSEContext'
|
||||
import { Sun, Moon, Monitor, Zap, Radio } from 'lucide-react'
|
||||
|
||||
const REFRESH_PRESETS = [
|
||||
{ label: '5s', value: 5_000 },
|
||||
{ label: '10s', value: 10_000 },
|
||||
{ label: '30s', value: 30_000 },
|
||||
{ label: '60s', value: 60_000 },
|
||||
]
|
||||
const SSE_STATUS_COPY: Record<string, { label: string; description: string; color: string }> = {
|
||||
connected: {
|
||||
label: 'Connected',
|
||||
description: 'Real-time updates are active. Agent status, tasks, and progress stream live.',
|
||||
color: 'text-green-500',
|
||||
},
|
||||
connecting: {
|
||||
label: 'Connecting…',
|
||||
description: 'Establishing SSE connection to the backend.',
|
||||
color: 'text-yellow-500',
|
||||
},
|
||||
reconnecting: {
|
||||
label: 'Reconnecting…',
|
||||
description: 'Connection lost. Retrying with exponential back-off.',
|
||||
color: 'text-yellow-500',
|
||||
},
|
||||
error: {
|
||||
label: 'Disconnected',
|
||||
description: 'Could not connect to the SSE endpoint. Check that the backend is reachable.',
|
||||
color: 'text-red-500',
|
||||
},
|
||||
}
|
||||
|
||||
export default function SettingsPage() {
|
||||
const { isDark, toggleTheme } = useTheme()
|
||||
const [gatewayUrl, setGatewayUrl] = useLocalStorage('cc-gateway-url', '')
|
||||
const [refreshInterval, setRefreshInterval] = useLocalStorage('cc-refresh-interval', 30_000)
|
||||
const { sseStatus } = useSSEContext()
|
||||
const sseInfo = SSE_STATUS_COPY[sseStatus] ?? SSE_STATUS_COPY.error
|
||||
|
||||
return (
|
||||
<div className="space-y-8 max-w-2xl">
|
||||
@@ -80,45 +98,31 @@ export default function SettingsPage() {
|
||||
</div>
|
||||
</section>
|
||||
|
||||
{/* Refresh */}
|
||||
{/* Real-time connection status */}
|
||||
<section className="space-y-4">
|
||||
<h2 className="text-lg font-semibold flex items-center gap-2">
|
||||
<Clock size={20} className="text-primary" />
|
||||
Auto Refresh
|
||||
<Radio size={20} className="text-primary" />
|
||||
Real-time Updates
|
||||
</h2>
|
||||
|
||||
<div className="p-5 rounded-xl border border-surface-light bg-surface-dark space-y-3">
|
||||
<p className="text-sm text-on-surface-variant">Data refresh interval for agent status and logs</p>
|
||||
|
||||
<div className="flex flex-col gap-2">
|
||||
<input
|
||||
type="range"
|
||||
min="0"
|
||||
max="3"
|
||||
step="1"
|
||||
value={REFRESH_PRESETS.findIndex((p) => p.value === refreshInterval)}
|
||||
onChange={(e) => {
|
||||
const idx = parseInt(e.target.value)
|
||||
setRefreshInterval(REFRESH_PRESETS[idx].value)
|
||||
}}
|
||||
className="w-full accent-primary"
|
||||
/>
|
||||
<div className="flex justify-between text-xs text-on-surface-muted">
|
||||
{REFRESH_PRESETS.map((p) => (
|
||||
<button
|
||||
key={p.label}
|
||||
onClick={() => setRefreshInterval(p.value)}
|
||||
className={`px-3 py-1 rounded-lg transition-colors ${
|
||||
refreshInterval === p.value
|
||||
? 'bg-primary/10 text-primary'
|
||||
: 'hover:bg-surface-light'
|
||||
}`}
|
||||
>
|
||||
{p.label}
|
||||
</button>
|
||||
))}
|
||||
<div className="flex items-center justify-between">
|
||||
<div>
|
||||
<p className="font-medium">SSE Connection</p>
|
||||
<p className="text-sm text-on-surface-variant mt-0.5">{sseInfo.description}</p>
|
||||
</div>
|
||||
<span className={`text-sm font-semibold whitespace-nowrap ${sseInfo.color}`}>
|
||||
{sseInfo.label}
|
||||
</span>
|
||||
</div>
|
||||
<p className="text-xs text-on-surface-muted">
|
||||
Endpoint: <code className="bg-surface-light px-1.5 py-0.5 rounded text-on-surface-variant">/api/events</code>
|
||||
· Events: agent.status, agent.task, agent.progress, fleet.update
|
||||
</p>
|
||||
<p className="text-xs text-on-surface-muted">
|
||||
Polling is disabled. All status updates are pushed from the server over a persistent SSE connection.
|
||||
The client reconnects automatically with exponential back-off on drop.
|
||||
</p>
|
||||
</div>
|
||||
</section>
|
||||
</div>
|
||||
|
||||
72
frontend/src/services/sse.ts
Normal file
72
frontend/src/services/sse.ts
Normal file
@@ -0,0 +1,72 @@
|
||||
/**
|
||||
* SSE event payload types matching the Go backend (internal/handler/sse.go).
|
||||
*
|
||||
* Event format on the wire:
|
||||
* event: <eventType>
|
||||
* data: <json>
|
||||
*
|
||||
* The types below define the backend contract. The SSEPayloadMap maps
|
||||
* each event type string to its expected payload shape. SSEMessage is a
|
||||
* discriminated union on `type` — when you switch on msg.type, TypeScript
|
||||
* narrows msg.data to the correct payload interface automatically.
|
||||
*/
|
||||
|
||||
import type { AgentStatus } from '../types'
|
||||
|
||||
/** agent.status — agent came online, went offline, changed state */
|
||||
export interface AgentStatusEvent {
|
||||
agentId: string
|
||||
status: AgentStatus
|
||||
/** Optional human-readable reason (e.g. error message) */
|
||||
reason?: string
|
||||
}
|
||||
|
||||
/** agent.task — a task was assigned to or completed by an agent */
|
||||
export interface AgentTaskEvent {
|
||||
agentId: string
|
||||
taskId: string
|
||||
title: string
|
||||
action: 'assigned' | 'completed' | 'failed'
|
||||
}
|
||||
|
||||
/** agent.progress — incremental progress update for a running task */
|
||||
export interface AgentProgressEvent {
|
||||
agentId: string
|
||||
taskId: string
|
||||
progress: number
|
||||
/** Optional description of what is currently happening */
|
||||
message?: string
|
||||
}
|
||||
|
||||
/**
|
||||
* fleet.update — bulk refresh of all agents (e.g. after a deployment).
|
||||
* The backend may send partial or complete agent state.
|
||||
*/
|
||||
export interface FleetUpdateEvent {
|
||||
/** ISO timestamp of when the snapshot was taken */
|
||||
timestamp: string
|
||||
/** Number of agents in the fleet */
|
||||
agentCount: number
|
||||
}
|
||||
|
||||
/** Union of all SSE data payloads keyed by event type. */
|
||||
export type SSEPayloadMap = {
|
||||
'agent.status': AgentStatusEvent
|
||||
'agent.task': AgentTaskEvent
|
||||
'agent.progress': AgentProgressEvent
|
||||
'fleet.update': FleetUpdateEvent
|
||||
connected: { clientCount: number }
|
||||
message: unknown
|
||||
}
|
||||
|
||||
/**
|
||||
* Discriminated SSE message — the `type` field narrows `data` via SSEPayloadMap.
|
||||
*
|
||||
* Usage:
|
||||
* if (msg.type === 'agent.status') {
|
||||
* msg.data.agentId // ✅ TypeScript knows this is AgentStatusEvent
|
||||
* }
|
||||
*/
|
||||
export type SSEMessage = {
|
||||
[K in keyof SSEPayloadMap]: { type: K; data: SSEPayloadMap[K] }
|
||||
}[keyof SSEPayloadMap]
|
||||
1
frontend/src/test-setup.ts
Normal file
1
frontend/src/test-setup.ts
Normal file
@@ -0,0 +1 @@
|
||||
import '@testing-library/jest-dom'
|
||||
@@ -1,4 +1,4 @@
|
||||
export type AgentStatus = 'active' | 'idle' | 'thinking' | 'error'
|
||||
export type AgentStatus = 'active' | 'idle' | 'thinking' | 'error' | 'offline'
|
||||
|
||||
export interface Agent {
|
||||
id: string
|
||||
|
||||
@@ -4,7 +4,7 @@
|
||||
"target": "es2023",
|
||||
"lib": ["ES2023", "DOM"],
|
||||
"module": "esnext",
|
||||
"types": ["vite/client"],
|
||||
"types": ["vite/client", "vitest/globals"],
|
||||
"skipLibCheck": true,
|
||||
|
||||
/* Bundler mode */
|
||||
|
||||
@@ -20,5 +20,5 @@
|
||||
"erasableSyntaxOnly": true,
|
||||
"noFallthroughCasesInSwitch": true
|
||||
},
|
||||
"include": ["vite.config.ts"]
|
||||
"include": ["vite.config.ts", "vitest.config.ts"]
|
||||
}
|
||||
|
||||
11
frontend/vitest.config.ts
Normal file
11
frontend/vitest.config.ts
Normal file
@@ -0,0 +1,11 @@
|
||||
import { defineConfig } from 'vitest/config'
|
||||
import react from '@vitejs/plugin-react'
|
||||
|
||||
export default defineConfig({
|
||||
plugins: [react()],
|
||||
test: {
|
||||
environment: 'jsdom',
|
||||
globals: true,
|
||||
setupFiles: ['./src/test-setup.ts'],
|
||||
},
|
||||
})
|
||||
Reference in New Issue
Block a user