diff --git a/ui/src/components/streaming/StreamingDashboard.tsx b/ui/src/components/streaming/StreamingDashboard.tsx new file mode 100644 index 0000000..9d85e0f --- /dev/null +++ b/ui/src/components/streaming/StreamingDashboard.tsx @@ -0,0 +1,347 @@ +import React, { useState } from 'react'; +import { useStreaming } from '../../hooks/useStreaming'; + +// ============================================================================ +// Types - SSE Envelope and Payloads +// ============================================================================ + +export interface SseEnvelope { + type: 'connected' | 'disconnected' | 'hands' | 'mcp' | 'task' | 'stats' | 'heartbeat' | 'error'; + timestamp: number; + trace_id?: string; + payload: T; +} + +export interface ConnectionPayload { + task_id: string; + connection_id: string; + message: string; +} + +export interface HeartbeatPayload { + server_time: number; + active_connections: number; +} + +export interface HandsPayload { + thought?: string; + action?: string; + target?: string; +} + +export interface McpEventPayload { + type: 'tooldiscovery' | 'toolstart' | 'toolprogress' | 'toolresult' | 'toolerror'; + tool?: string; + id?: string; + input?: Record; + progress?: Record; + result?: Record; + error?: string; +} + +export interface TaskPayload { + status: 'pending' | 'running' | 'completed' | 'failed'; + step?: number; + max_steps?: number; + output?: string; +} + +export interface StatsPayload { + active_connections: number; + total_connections: number; + events: { + thought: number; + tool_call: number; + tool_progress: number; + tool_result: number; + approval_needed: number; + error: number; + complete: number; + total: number; + }; + errors: { + auth: number; + replay: number; + internal: number; + total: number; + }; +} + +// ============================================================================ +// Stats Panel Component +// ============================================================================ + +interface StatsPanelProps { + endpoint: string; +} + +const StatsPanel: React.FC = ({ endpoint }) => { + const { events, connected, error } = useStreaming(endpoint); + + const stats: StatsPayload | null = events + .filter((e) => e.type === 'stats') + .map((e) => e.payload as StatsPayload) + .pop() || null; + + return ( +
+

Streaming Stats

+
+ + {connected ? 'Connected' : error ? 'Error' : 'Disconnected'} +
+ {stats ? ( +
+
+ Active Connections: + {stats.active_connections} +
+
+ Total Connections: + {stats.total_connections} +
+
+ Total Events: + {stats.events.total} +
+
+ Total Errors: + {stats.errors.total} +
+
+ ) : ( +

Waiting for stats...

+ )} +
+ ); +}; + +// ============================================================================ +// Hands Panel Component +// ============================================================================ + +interface HandsPanelProps { + taskId: string; +} + +const HandsPanel: React.FC = ({ taskId }) => { + const endpoint = `/stream/hands/${taskId}`; + const { events, connected, error } = useStreaming(endpoint); + + const handsEvents = events.filter((e) => e.type === 'hands'); + + return ( +
+

Hands Agent

+
+ + {connected ? 'Connected' : error ? 'Error' : 'Disconnected'} +
+
+ {handsEvents.length === 0 ? ( +

Waiting for hands events...

+ ) : ( + handsEvents.map((event, idx) => { + const payload = event.payload as HandsPayload; + return ( +
+ {payload.thought &&

{payload.thought}

} + {payload.action && ( +

+ Action: {payload.action} + {payload.target && ` -> ${payload.target}`} +

+ )} +
+ ); + }) + )} +
+
+ ); +}; + +// ============================================================================ +// MCP Panel Component +// ============================================================================ + +interface McpPanelProps { + taskId: string; +} + +const McpPanel: React.FC = ({ taskId }) => { + const endpoint = `/stream/mcp/${taskId}`; + const { events, connected, error } = useStreaming(endpoint); + + const mcpEvents = events.filter((e) => e.type === 'mcp'); + + return ( +
+

MCP Tools

+
+ + {connected ? 'Connected' : error ? 'Error' : 'Disconnected'} +
+
+ {mcpEvents.length === 0 ? ( +

Waiting for MCP events...

+ ) : ( + mcpEvents.map((event, idx) => { + const payload = event.payload as McpEventPayload; + return ( +
+ {payload.type} + {payload.tool &&

Tool: {payload.tool}

} + {payload.result && ( + 
+                    {JSON.stringify(payload.result, null, 2)}
+
+ )} + {payload.error &&

Error: {payload.error}

} +
+ ); + }) + )} +
+
+ ); +}; + +// ============================================================================ +// Task Panel Component +// ============================================================================ + +interface TaskPanelProps { + taskId: string; +} + +const TaskPanel: React.FC = ({ taskId }) => { + const endpoint = `/stream/task/${taskId}`; + const { events, connected, error } = useStreaming(endpoint); + + const taskEvents = events.filter((e) => e.type === 'task' || e.type === 'heartbeat'); + + return ( +
+

Task Execution

+
+ + {connected ? 'Connected' : error ? 'Error' : 'Disconnected'} +
+
+ {taskEvents.length === 0 ? ( +

Waiting for task events...

+ ) : ( + taskEvents.map((event, idx) => { + const payload = event.payload as TaskPayload | HeartbeatPayload; + if (event.type === 'heartbeat') { + const hb = payload as HeartbeatPayload; + return ( +
+ Heartbeat +

+ Server: {new Date(hb.server_time).toISOString()} | Active: {hb.active_connections} +

+
+ ); + } + const tp = payload as TaskPayload; + return ( +
+ + {tp.status} + + {tp.step !== undefined && tp.max_steps !== undefined && ( +

+ Step {tp.step} / {tp.max_steps} +

+ )} + {tp.output &&

{tp.output}

} +
+ ); + }) + )} +
+
+ ); +}; + +// ============================================================================ +// Main StreamingDashboard Component +// ============================================================================ + +interface StreamingDashboardProps { + taskId?: string; +} + +export const StreamingDashboard: React.FC = ({ taskId = 'default' }) => { + const [activeTab, setActiveTab] = useState<'stats' | 'hands' | 'mcp' | 'task'>('stats'); + + return ( +
+ {/* Header */} +
+

Streaming Dashboard

+

Real-time agent execution monitoring

+
+ + {/* Tabs */} +
+ {(['stats', 'hands', 'mcp', 'task'] as const).map((tab) => ( + + ))} +
+ + {/* Content */} +
+ {activeTab === 'stats' && } + {activeTab === 'hands' && } + {activeTab === 'mcp' && } + {activeTab === 'task' && } +
+ + {/* Footer */} +
+ APEX Streaming v1.7.0 | SSE Events | Heartbeat: 30s +
+
+ ); +}; + +export default StreamingDashboard; diff --git a/ui/src/hooks/useStreaming.ts b/ui/src/hooks/useStreaming.ts new file mode 100644 index 0000000..00fd028 --- /dev/null +++ b/ui/src/hooks/useStreaming.ts @@ -0,0 +1,180 @@ +import { useState, useEffect, useCallback, useRef } from 'react'; + +// ============================================================================ +// Types - SSE Envelope and Payloads (mirrored from StreamingDashboard) +// ============================================================================ + +export interface SseEnvelope { + type: 'connected' | 'disconnected' | 'hands' | 'mcp' | 'task' | 'stats' | 'heartbeat' | 'error'; + timestamp: number; + trace_id?: string; + payload: T; +} + +interface UseStreamingOptions { + /** Auto-connect on mount */ + autoConnect?: boolean; + /** Reconnection delay in ms */ + reconnectDelay?: number; + /** Maximum reconnection attempts */ + maxReconnectAttempts?: number; + /** Callback on connect */ + onConnect?: () => void; + /** Callback on disconnect */ + onDisconnect?: () => void; + /** Callback on error */ + onError?: (error: Event) => void; +} + +interface UseStreamingReturn { + /** All collected events */ + events: SseEnvelope[]; + /** Whether currently connected */ + connected: boolean; + /** Last error */ + error: Event | null; + /** Connect manually */ + connect: () => void; + /** Disconnect manually */ + disconnect: () => void; + /** Clear all events */ + clear: () => void; +} + +/** + * Hook to consume SSE streams from APEX streaming endpoints. + * Supports browser-friendly authentication via query params (?sig=...&ts=...) + * + * @param endpoint - The SSE endpoint path (e.g., '/stream/stats', '/stream/hands/:taskId') + * @param options - Configuration options + * @returns Streaming state and control functions + */ +export function useStreaming( + endpoint: string, + options: UseStreamingOptions = {} +): UseStreamingReturn { + const { + autoConnect = true, + reconnectDelay = 3000, + maxReconnectAttempts = 5, + onConnect, + onDisconnect, + onError, + } = options; + + const [events, setEvents] = useState[]>([]); + const [connected, setConnected] = useState(false); + const [error, setError] = useState(null); + + const eventSourceRef = useRef(null); + const reconnectAttemptsRef = useRef(0); + const reconnectTimeoutRef = useRef(null); + + // Parse the base URL from the current window location + const getBaseUrl = useCallback(() => { + if (typeof window === 'undefined') return 'http://localhost:3000'; + return `${window.location.protocol}//${window.location.host}`; + }, []); + + // Generate signed URL for SSE (browser-friendly auth) + // Note: In production, this should call a backend endpoint to get a signed URL + const getSignedUrl = useCallback(async (path: string): Promise => { + const baseUrl = getBaseUrl(); + + // For development, we'll try without auth first + // In production, this should call: await fetch('/api/v1/streams/sign?path=' + path) + // And use the returned signed URL + return `${baseUrl}${path}`; + }, [getBaseUrl]); + + // Connect to SSE endpoint + const connect = useCallback(async () => { + // Clean up existing connection + if (eventSourceRef.current) { + eventSourceRef.current.close(); + } + + try { + const url = await getSignedUrl(endpoint); + const eventSource = new EventSource(url); + + eventSource.onopen = () => { + setConnected(true); + setError(null); + reconnectAttemptsRef.current = 0; + onConnect?.(); + }; + + eventSource.onmessage = (event) => { + try { + const data = JSON.parse(event.data) as SseEnvelope; + setEvents((prev) => [...prev, data]); + } catch (parseError) { + console.error('Failed to parse SSE message:', parseError); + } + }; + + eventSource.onerror = (err) => { + setConnected(false); + setError(err); + onError?.(err); + + // Attempt reconnection + if (reconnectAttemptsRef.current < maxReconnectAttempts) { + reconnectAttemptsRef.current += 1; + reconnectTimeoutRef.current = setTimeout(() => { + connect(); + }, reconnectDelay); + } + }; + + eventSourceRef.current = eventSource; + } catch (err) { + setError(err as Event); + setConnected(false); + } + }, [endpoint, getSignedUrl, reconnectDelay, maxReconnectAttempts, onConnect, onError]); + + // Disconnect from SSE endpoint + const disconnect = useCallback(() => { + if (reconnectTimeoutRef.current) { + clearTimeout(reconnectTimeoutRef.current); + reconnectTimeoutRef.current = null; + } + + if (eventSourceRef.current) { + eventSourceRef.current.close(); + eventSourceRef.current = null; + } + + setConnected(false); + onDisconnect?.(); + }, [onDisconnect]); + + // Clear all events + const clear = useCallback(() => { + setEvents([]); + }, []); + + // Auto-connect on mount + useEffect(() => { + if (autoConnect) { + connect(); + } + + return () => { + disconnect(); + }; + }, [autoConnect, connect, disconnect]); + + return { + events, + connected, + error, + connect, + disconnect, + clear, + }; +} + +export default useStreaming;