diff --git a/.codegraph/codegraph.db b/.codegraph/codegraph.db new file mode 100644 index 0000000000..f90d4d501a Binary files /dev/null and b/.codegraph/codegraph.db differ diff --git a/.env.example b/.env.example index ba5e0b8c08..ff1b67655c 100644 --- a/.env.example +++ b/.env.example @@ -77,6 +77,16 @@ PORT=20128 # API_HOST=0.0.0.0 # DASHBOARD_PORT=20128 +# Port for the real-time WebSocket live monitoring server. +# Used by: src/server/ws/liveServer.ts, src/app/api/v1/ws/route.ts +# Default: 20129 +# LIVE_WS_PORT=20129 + +# Disable the real-time WebSocket server. +# Used by: src/server/ws/liveServer.ts, scripts/start-ws-server.mjs +# Default: false | Set to 1 or true to disable. +# OMNIROUTE_DISABLE_LIVE_WS=false + # Use Turbopack in local dev. Next 16.2.4 can fail to compile next/font/google # through the custom dev runner without this on Windows. OMNIROUTE_USE_TURBOPACK=1 diff --git a/docs/reference/ENVIRONMENT.md b/docs/reference/ENVIRONMENT.md index 776e4165d3..7ac467e6cf 100644 --- a/docs/reference/ENVIRONMENT.md +++ b/docs/reference/ENVIRONMENT.md @@ -119,6 +119,8 @@ OmniRoute uses **SQLite** (via `better-sqlite3`) for all persistence. These vari | `PROD_DASHBOARD_PORT` | `20130` | `docker-compose.prod.yml` | Host-side published port for the Dashboard in Docker production mode. | | `PROD_API_PORT` | `20131` | `docker-compose.prod.yml` | Host-side published port for the API in Docker production mode. | | `OMNIROUTE_PORT` | _(unset)_ | `src/lib/runtime/ports.ts` | Takes precedence over `PORT` when running inside Electron or other wrappers. | +| `LIVE_WS_PORT` | `20129` | `src/server/ws/liveServer.ts` | Port for the real-time WebSocket live monitoring server. | +| `OMNIROUTE_DISABLE_LIVE_WS` | `false` | `src/server/ws/liveServer.ts` | Set to `1` or `true` to disable the real-time WebSocket server. | | `NODE_ENV` | `production` | Next.js core | Controls logging verbosity, caching, error detail exposure, and Next.js optimizations. | | `OMNIROUTE_USE_TURBOPACK` | `1` (default in `.env.example`) | `package.json` / Next.js 16 | Toggles the Next.js 16 Turbopack bundler in `npm run dev` and `npm run build`. Set to `0` on Windows or when running into native binding incompatibilities. | | `OMNIROUTE_SKIP_DB_HEALTHCHECK` | _(unset)_ | `src/lib/db/core.ts` / `src/lib/db/healthCheck.ts` | Set to `1` to skip the SQLite integrity health check on startup. Useful for faster boot on large databases. | diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index d7fe60ba71..2131cf532a 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -65,6 +65,7 @@ import { getChatLogMaxObjectKeys, } from "@/lib/logEnv"; import { logAuditEvent } from "@/lib/compliance"; +import { emit } from "@/lib/events/eventBus"; import { extractProviderWarnings } from "@/lib/compliance/providerAudit"; import { adaptBodyForCompression } from "../services/compression/bodyAdapter.ts"; import { handleBypassRequest } from "../utils/bypassHandler.ts"; @@ -1366,6 +1367,17 @@ export async function handleChatCore({ // Per-request trace id + checkpoint helper. Lets us see exactly which await // a hung request was sitting on in `[STAGE_TRACE]` log lines. const traceId = Math.random().toString(36).slice(2, 8); + + // Emit request.started event for real-time dashboard + setImmediate(() => { + emit("request.started", { + id: traceId, + model: model || "unknown", + provider: provider || "unknown", + timestamp: startTime, + comboName: comboName || undefined, + }); + }); const trace = (label: string, extra?: Record) => { const elapsed = Date.now() - startTime; const suffix = extra ? ` ${JSON.stringify(extra)}` : ""; diff --git a/open-sse/services/combo.ts b/open-sse/services/combo.ts index 1969acbdeb..a30999006e 100644 --- a/open-sse/services/combo.ts +++ b/open-sse/services/combo.ts @@ -37,6 +37,8 @@ import { getCircuitBreaker } from "../../src/shared/utils/circuitBreaker"; import { fisherYatesShuffle, getNextFromDeck } from "../../src/shared/utils/shuffleDeck"; import { parseModel } from "./model.ts"; import { applyComboAgentMiddleware, injectModelTag } from "./comboAgentMiddleware.ts"; +import { checkCredentialGate, logCredentialSkip } from "./credentialGate.ts"; +import { emit } from "../../src/lib/events/eventBus"; import { classifyWithConfig, DEFAULT_INTENT_CONFIG } from "./intentClassifier.ts"; import { selectProvider as selectAutoProvider } from "./autoCombo/engine.ts"; import { selectWithStrategy } from "./autoCombo/routerStrategy.ts"; @@ -2074,6 +2076,17 @@ export async function handleComboChat({ } } + // Credential gate: skip targets with known-bad credentials (fail-fast) + const connectionId = target.connectionId as string | undefined; + if (connectionId) { + const gateResult = checkCredentialGate(connectionId, provider, modelStr); + if (gateResult.allowed === false) { + logCredentialSkip(log, modelStr, gateResult.reason || "Credential gate blocked"); + if (i > 0) fallbackCount++; + continue; + } + } + // Retry loop for transient errors for (let retry = 0; retry <= maxRetries; retry++) { // Fix #1681: Bail out immediately if the client has disconnected @@ -2115,6 +2128,14 @@ export async function handleComboChat({ "COMBO", `Trying model ${i + 1}/${orderedTargets.length}: ${modelStr}${retry > 0 ? ` (retry ${retry})` : ""}` ); + emit("combo.target.attempt", { + comboName: combo.name, + targetIndex: i, + provider, + model: modelStr, + timestamp: Date.now(), + strategy, + }); // Universal handoff: inject existing handoff if model changed if ( @@ -2161,8 +2182,23 @@ export async function handleComboChat({ if (!lastStatus) lastStatus = 502; if (i > 0) fallbackCount++; break; // move to next model + emit("combo.target.failed", { + comboName: combo.name, + targetIndex: i, + provider, + model: modelStr, + error: `Quality: ${quality.reason}`, + latencyMs: Date.now() - startTime, + }); } const latencyMs = Date.now() - startTime; + emit("combo.target.succeeded", { + comboName: combo.name, + targetIndex: i, + provider, + model: modelStr, + latencyMs, + }); log.info( "COMBO", `Model ${modelStr} succeeded (${latencyMs}ms, ${fallbackCount} fallbacks)` diff --git a/package.json b/package.json index fddc1ae399..f67d93a416 100644 --- a/package.json +++ b/package.json @@ -192,7 +192,8 @@ "xxhash-wasm": "^1.1.0", "yazl": "^3.3.1", "zod": "^4.4.3", - "zustand": "^5.0.13" + "zustand": "^5.0.13", + "ws": "^8.18.0" }, "optionalDependencies": { "better-sqlite3": "^12.10.0", @@ -211,6 +212,7 @@ "@types/node": "^25.9.1", "@types/react": "^19.2.15", "@types/react-dom": "^19.2.3", + "@types/ws": "^8.18.0", "@vitejs/plugin-react": "^6.0.2", "c8": "^11.0.0", "concurrently": "^9.2.1", diff --git a/scripts/start-ws-server.mjs b/scripts/start-ws-server.mjs new file mode 100644 index 0000000000..81c28d3ded --- /dev/null +++ b/scripts/start-ws-server.mjs @@ -0,0 +1,42 @@ +/** + * Live Dashboard WebSocket Server — Startup Script + * + * This script starts the live dashboard WebSocket server as a separate + * process alongside the Next.js app. Run it with: + * + * node scripts/start-ws-server.js + * + * Or use the built-in auto-start in src/server/ws/liveServer.ts. + * + * Environment variables: + * LIVE_WS_PORT — WebSocket server port (default: 20129) + * OMNIROUTE_DISABLE_LIVE_WS — Set to "1" or "true" to disable + */ + +if (process.env.OMNIROUTE_DISABLE_LIVE_WS === "1" || process.env.OMNIROUTE_DISABLE_LIVE_WS === "true") { + console.log("[LiveWS] Disabled via OMNIROUTE_DISABLE_LIVE_WS"); + process.exit(0); +} + +// Register tsx to handle TypeScript imports +import { register } from "node:module"; +import { pathToFileURL } from "node:url"; + +register("tsx", pathToFileURL("./")); + +const { startLiveDashboardServer } = await import("../src/server/ws/liveServer"); + +const port = parseInt(process.env.LIVE_WS_PORT || "20129", 10); + +console.log(`[LiveWS] Starting dashboard WebSocket server on port ${port}...`); + +startLiveDashboardServer(port) + .then((server) => { + console.log(`[LiveWS] Dashboard WebSocket server listening on ws://0.0.0.0:${port}`); + console.log(`[LiveWS] Connect via: ws://localhost:${port}?token=`); + console.log(`[LiveWS] Channels: requests, combo, credentials`); + }) + .catch((err) => { + console.error("[LiveWS] Failed to start:", err); + process.exit(1); + }); diff --git a/src/app/api/v1/ws/route.ts b/src/app/api/v1/ws/route.ts index 6233c3ecc1..85876a77ac 100644 --- a/src/app/api/v1/ws/route.ts +++ b/src/app/api/v1/ws/route.ts @@ -13,6 +13,14 @@ const WS_PROTOCOL = { payload: { model: "openai/gpt-4.1-mini", messages: [] }, }, cancel: { type: "cancel", id: "req-1" }, + live: { + port: parseInt(process.env.LIVE_WS_PORT || "20129", 10), + path: "/live", + protocol: "json", + channels: ["requests", "combo", "credentials"], + auth: "api-key", + heartbeatMs: 15000, + }, }; export async function OPTIONS() { @@ -59,6 +67,14 @@ export async function GET(request: Request) { authenticated: auth.authenticated, authType: auth.authType, protocol: WS_PROTOCOL, + live: { + port: parseInt(process.env.LIVE_WS_PORT || "20129", 10), + path: "/live", + protocol: "json", + channels: ["requests", "combo", "credentials"], + auth: "api-key", + description: "Real-time dashboard events via WebSocket", + }, }, { headers: WS_HANDSHAKE_HEADERS, diff --git a/src/hooks/useLiveDashboard.ts b/src/hooks/useLiveDashboard.ts new file mode 100644 index 0000000000..eef033c530 --- /dev/null +++ b/src/hooks/useLiveDashboard.ts @@ -0,0 +1,417 @@ +/** + * useLiveDashboard — React hooks for real-time dashboard WebSocket + * + * Provides hooks for connecting to the live dashboard WebSocket server + * and subscribing to event channels. + * + * Usage: + * const { requests, isConnected } = useLiveRequests(); + * const { comboEvents, lastComboEvent } = useLiveComboStatus(); + */ + +"use client"; + +import { useEffect, useRef, useState, useCallback } from "react"; +import type { DashboardChannel, DashboardEventName } from "@/lib/events/types"; + +// ── Config ──────────────────────────────────────────────────────────────── + +const WS_RECONNECT_DELAYS = [1000, 2000, 4000, 8000, 16000, 30000]; +const DEFAULT_WS_URL = `ws://${typeof window !== "undefined" ? window.location.hostname : "localhost"}:20129`; + +// ── Types ───────────────────────────────────────────────────────────────── + +export interface WsEventPayload { + event: string; + channel: DashboardChannel; + data: unknown; + timestamp: number; +} + +export interface DashboardConnectionState { + isConnected: boolean; + isConnecting: boolean; + error: string | null; + reconnectAttempt: number; +} + +// ── Core Hook ───────────────────────────────────────────────────────────── + +export interface UseLiveDashboardOptions { + /** WebSocket URL (default: ws://hostname:20129) */ + wsUrl?: string; + /** API key for authentication */ + apiKey?: string; + /** Channels to subscribe to */ + channels?: DashboardChannel[]; + /** Auto-reconnect on disconnect (default: true) */ + autoReconnect?: boolean; +} + +/** + * Core WebSocket connection hook. + * Manages connection lifecycle, reconnection, and event streaming. + */ +export function useLiveDashboard({ + wsUrl = DEFAULT_WS_URL, + apiKey, + channels = ["requests", "combo", "credentials"], + autoReconnect = true, +}: UseLiveDashboardOptions = {}) { + const [connection, setConnection] = useState({ + isConnected: false, + isConnecting: false, + error: null, + reconnectAttempt: 0, + }); + + const [events, setEvents] = useState([]); + const wsRef = useRef(null); + const reconnectTimeoutRef = useRef | null>(null); + const mountedRef = useRef(true); + const maxEvents = 500; + + const connect = useCallback(() => { + if (!mountedRef.current) return; + if (wsRef.current?.readyState === WebSocket.OPEN) return; + + setConnection((prev) => ({ + ...prev, + isConnecting: true, + error: null, + })); + + try { + const wsUrlWithAuth = apiKey + ? `${wsUrl}?token=${encodeURIComponent(apiKey)}` + : wsUrl; + + const ws = new WebSocket(wsUrlWithAuth); + wsRef.current = ws; + + ws.onopen = () => { + if (!mountedRef.current) return; + setConnection({ + isConnected: true, + isConnecting: false, + error: null, + reconnectAttempt: 0, + }); + + // Subscribe to channels + ws.send(JSON.stringify({ type: "subscribe", channels })); + }; + + ws.onmessage = (event) => { + if (!mountedRef.current) return; + try { + const msg = JSON.parse(event.data); + + if (msg.type === "event") { + const payload: WsEventPayload = { + event: msg.event, + channel: msg.channel, + data: msg.data, + timestamp: msg.timestamp || Date.now(), + }; + setEvents((prev) => { + const next = [...prev, payload]; + return next.length > maxEvents ? next.slice(-maxEvents) : next; + }); + } else if (msg.type === "pong") { + // Heartbeat response + } else if (msg.type === "welcome") { + // Send backlog + if (Array.isArray(msg.data)) { + setEvents((prev) => { + const next = [...prev, ...msg.data]; + return next.length > maxEvents ? next.slice(-maxEvents) : next; + }); + } + } else if (msg.type === "error") { + console.error("[LiveWS] Server error:", msg.code, msg.message); + } + } catch { + // Ignore parse errors + } + }; + + ws.onclose = () => { + if (!mountedRef.current) return; + wsRef.current = null; + setConnection((prev) => ({ + ...prev, + isConnected: false, + isConnecting: false, + })); + + if (autoReconnect) { + const attempt = connection.reconnectAttempt; + const delay = WS_RECONNECT_DELAYS[Math.min(attempt, WS_RECONNECT_DELAYS.length - 1)]; + reconnectTimeoutRef.current = setTimeout(() => { + setConnection((prev) => ({ + ...prev, + reconnectAttempt: prev.reconnectAttempt + 1, + })); + }, delay); + } + }; + + ws.onerror = () => { + if (!mountedRef.current) return; + setConnection((prev) => ({ + ...prev, + isConnecting: false, + error: "Connection failed", + })); + }; + } catch (err) { + setConnection((prev) => ({ + ...prev, + isConnecting: false, + error: err instanceof Error ? err.message : "Connection failed", + })); + } + }, [wsUrl, apiKey, channels.join(","), autoReconnect, connection.reconnectAttempt]); + + // Connect on mount and on reconnect trigger + useEffect(() => { + connect(); + return () => { + mountedRef.current = false; + if (reconnectTimeoutRef.current) { + clearTimeout(reconnectTimeoutRef.current); + } + wsRef.current?.close(); + }; + }, [connect]); + + // Connect (for manual retry) + const reconnect = useCallback(() => { + wsRef.current?.close(); + setConnection((prev) => ({ + ...prev, + reconnectAttempt: prev.reconnectAttempt + 1, + })); + }, []); + + return { + connection, + events, + reconnect, + /** Filter events by channel */ + getEventsByChannel: useCallback( + (channel: DashboardChannel) => events.filter((e) => e.channel === channel), + [events], + ), + /** Filter events by name */ + getEventsByName: useCallback( + (eventName: string) => events.filter((e) => e.event === eventName), + [events], + ), + /** Clear event history */ + clearEvents: useCallback(() => setEvents([]), []), + }; +} + +// ── Request Monitoring Hook ─────────────────────────────────────────────── + +export interface LiveRequest { + id: string; + model: string; + provider: string; + timestamp: number; + status: "pending" | "running" | "success" | "error"; + tokensInput?: number; + tokensOutput?: number; + latencyMs?: number; + error?: string; + comboName?: string; +} + +/** + * Hook for monitoring live requests. + */ +export function useLiveRequests(options?: UseLiveDashboardOptions) { + const { events, connection, reconnect } = useLiveDashboard({ + channels: ["requests"], + ...options, + }); + + const [activeRequests, setActiveRequests] = useState>(new Map()); + const [completedRequests, setCompletedRequests] = useState([]); + const maxCompleted = 100; + + useEffect(() => { + setActiveRequests((prev) => { + const next = new Map(prev); + + for (const event of events) { + if (event.event === "request.started") { + const data = event.data as any; + next.set(data.id, { + id: data.id, + model: data.model, + provider: data.provider, + timestamp: data.timestamp, + status: "pending", + comboName: data.comboName, + }); + } else if (event.event === "request.streaming") { + const data = event.data as any; + const existing = next.get(data.id); + if (existing) { + next.set(data.id, { ...existing, status: "running" }); + } + } else if (event.event === "request.completed") { + const data = event.data as any; + const existing = next.get(data.id); + if (existing) { + next.delete(data.id); + setCompletedRequests((prev) => { + const done: LiveRequest = { + ...existing, + status: data.status === "success" ? "success" : "error", + tokensInput: data.tokensInput, + tokensOutput: data.tokensOutput, + latencyMs: data.latencyMs, + error: data.error, + }; + return [done, ...prev].slice(0, maxCompleted); + }); + } + } else if (event.event === "request.failed") { + const data = event.data as any; + const existing = next.get(data.id); + if (existing) { + next.delete(data.id); + setCompletedRequests((prev) => { + const done: LiveRequest = { + ...existing, + status: "error", + error: data.error, + latencyMs: data.latencyMs, + }; + return [done, ...prev].slice(0, maxCompleted); + }); + } + } + } + + return next; + }); + + // Reset events after processing + }, [events]); + + return { + activeRequests: Array.from(activeRequests.values()), + completedRequests, + activeCount: activeRequests.size, + isConnected: connection.isConnected, + reconnect, + }; +} + +// ── Combo Status Hook ───────────────────────────────────────────────────── + +export interface LiveComboEvent { + comboName: string; + targetIndex: number; + provider: string; + model: string; + type: "attempt" | "succeeded" | "failed"; + latencyMs?: number; + error?: string; + timestamp: number; +} + +/** + * Hook for monitoring live combo cascade status. + */ +export function useLiveComboStatus(options?: UseLiveDashboardOptions) { + const { events, connection, reconnect } = useLiveDashboard({ + channels: ["combo"], + ...options, + }); + + const [comboEvents, setComboEvents] = useState([]); + const maxComboEvents = 200; + + useEffect(() => { + for (const event of events) { + const data = event.data as any; + let comboEvent: LiveComboEvent | null = null; + + if (event.event === "combo.target.attempt") { + comboEvent = { + comboName: data.comboName, + targetIndex: data.targetIndex, + provider: data.provider, + model: data.model, + type: "attempt", + timestamp: event.timestamp, + }; + } else if (event.event === "combo.target.succeeded") { + comboEvent = { + comboName: data.comboName, + targetIndex: data.targetIndex, + provider: data.provider, + model: data.model, + type: "succeeded", + latencyMs: data.latencyMs, + timestamp: event.timestamp, + }; + } else if (event.event === "combo.target.failed") { + comboEvent = { + comboName: data.comboName, + targetIndex: data.targetIndex, + provider: data.provider, + model: data.model, + type: "failed", + error: data.error, + latencyMs: data.latencyMs, + timestamp: event.timestamp, + }; + } + + if (comboEvent) { + setComboEvents((prev) => [comboEvent!, ...prev].slice(0, maxComboEvents)); + } + } + }, [events]); + + /** Get events for a specific combo */ + const getComboHistory = useCallback( + (comboName: string) => comboEvents.filter((e) => e.comboName === comboName), + [comboEvents], + ); + + /** Get the last event for a specific combo */ + const getLastComboEvent = useCallback( + (comboName: string) => comboEvents.find((e) => e.comboName === comboName), + [comboEvents], + ); + + return { + comboEvents, + activeCombos: new Set(comboEvents.map((e) => e.comboName)), + isConnected: connection.isConnected, + getComboHistory, + getLastComboEvent, + reconnect, + }; +} + +// ── Connection Status Hook ──────────────────────────────────────────────── + +/** + * Hook for checking connection status only (no events). + */ +export function useLiveConnectionStatus(options?: UseLiveDashboardOptions) { + const { connection, reconnect } = useLiveDashboard({ + channels: [], + ...options, + }); + return { ...connection, reconnect }; +} diff --git a/src/lib/db/contextHandoffs.ts b/src/lib/db/contextHandoffs.ts index eb4ff1e455..6ef426c964 100644 --- a/src/lib/db/contextHandoffs.ts +++ b/src/lib/db/contextHandoffs.ts @@ -202,10 +202,7 @@ export function recordSessionModelUsage( * @param comboName - The combo name. * @returns The model string, or null if no record exists. */ -export function getLastSessionModel( - sessionId: string, - comboName: string, -): string | null { +export function getLastSessionModel(sessionId: string, comboName: string): string | null { const db = getDbInstance() as unknown as DbLike; const row = db .prepare( diff --git a/src/lib/events/eventBus.ts b/src/lib/events/eventBus.ts new file mode 100644 index 0000000000..776ecc80ad --- /dev/null +++ b/src/lib/events/eventBus.ts @@ -0,0 +1,201 @@ +/** + * Event Bus + * + * Singleton typed EventEmitter for dashboard real-time events. + * Uses globalThis pattern to survive HMR. + * + * Events are emitted by various parts of the system: + * - chatCore.ts: request.started + * - stream.ts: request.completed, request.failed + * - combo.ts: combo.target.* + * - credentialGate.ts: credential.health.changed + * + * Consumers (WebSocket server, dashboard hooks, etc.) subscribe + * to specific event types via on/off. + */ + +import { + type DashboardEventName, + type DashboardEventMap, + type DashboardEventListener, +} from "./types"; + +// ── State (globalThis singleton) ────────────────────────────────────────── + +declare global { + var __omnirouteEventBus: + | { + initialized: boolean; + listeners: Map>; + wildcardListeners: Set; + history: Array<{ event: DashboardEventName; payload: unknown; timestamp: number }>; + maxHistory: number; + emitCount: number; + } + | undefined; +} + +function getBusState() { + if (!globalThis.__omnirouteEventBus) { + globalThis.__omnirouteEventBus = { + initialized: false, + listeners: new Map(), + wildcardListeners: new Set(), + history: [], + maxHistory: 100, + emitCount: 0, + }; + } + return globalThis.__omnirouteEventBus; +} + +// ── Event History ───────────────────────────────────────────────────────── + +export interface HistoryEntry { + event: DashboardEventName; + payload: unknown; + timestamp: number; +} + +/** + * Get recent event history (for clients that connect late). + */ +export function getEventHistory( + sinceTimestamp?: number, + maxEntries = 50, +): HistoryEntry[] { + const state = getBusState(); + let history = state.history; + if (sinceTimestamp) { + history = history.filter((h) => h.timestamp > sinceTimestamp); + } + return history.slice(-maxEntries); +} + +// ── Public API ──────────────────────────────────────────────────────────── + +/** + * Emit an event to all subscribers. + */ +export function emit( + event: E, + payload: DashboardEventMap[E], +): void { + const state = getBusState(); + state.emitCount++; + + const entry = { event, payload, timestamp: Date.now() }; + state.history.push(entry); + if (state.history.length > state.maxHistory) { + state.history.shift(); + } + + // Notify specific listeners + const eventListeners = state.listeners.get(event); + if (eventListeners) { + for (const listener of eventListeners) { + try { + listener(payload); + } catch (err) { + console.error(`[EventBus] Error in listener for ${event}:`, err); + } + } + } + + // Notify wildcard listeners (\*) + for (const listener of state.wildcardListeners) { + try { + listener(event, payload); + } catch (err) { + console.error(`[EventBus] Error in wildcard listener for ${event}:`, err); + } + } +} + +/** + * Subscribe to a specific event. + * Returns an unsubscribe function. + */ +export function on( + event: E, + listener: DashboardEventListener, +): () => void { + const state = getBusState(); + if (!state.listeners.has(event)) { + state.listeners.set(event, new Set()); + } + state.listeners.get(event)!.add(listener as Function); + + return () => { + state.listeners.get(event)?.delete(listener as Function); + }; +} + +/** + * Subscribe to ALL events (wildcard). + * Returns an unsubscribe function. + */ +export function onAny( + listener: (event: DashboardEventName, payload: unknown) => void, +): () => void { + const state = getBusState(); + state.wildcardListeners.add(listener); + return () => { + state.wildcardListeners.delete(listener); + }; +} + +/** + * Remove a specific listener. + */ +export function off( + event: E, + listener: DashboardEventListener, +): void { + getBusState().listeners.get(event)?.delete(listener as Function); +} + +/** + * Remove all listeners for an event. + */ +export function removeAllListeners(event?: DashboardEventName): void { + const state = getBusState(); + if (event) { + state.listeners.delete(event); + } else { + state.listeners.clear(); + state.wildcardListeners.clear(); + } +} + +/** + * Get bus stats for monitoring. + */ +export function getBusStats(): { + totalListeners: number; + eventsWithListeners: number; + totalEmitted: number; + historySize: number; +} { + const state = getBusState(); + let totalListeners = state.wildcardListeners.size; + for (const set of state.listeners.values()) { + totalListeners += set.size; + } + return { + totalListeners, + eventsWithListeners: state.listeners.size, + totalEmitted: state.emitCount, + historySize: state.history.length, + }; +} + +/** + * Initialize event bus (idempotent). + */ +export function initEventBus(): void { + getBusState().initialized = true; +} + +// Auto-initialize on import +initEventBus(); diff --git a/src/lib/events/types.ts b/src/lib/events/types.ts new file mode 100644 index 0000000000..fe1f1b43bc --- /dev/null +++ b/src/lib/events/types.ts @@ -0,0 +1,130 @@ +/** + * Event Bus Types + * + * Typed event definitions for the real-time event bus. + * Each event has a name and a typed payload interface. + */ + +// ── Event Names ─────────────────────────────────────────────────────────── + +export type DashboardEventName = + | "request.started" + | "request.streaming" + | "request.completed" + | "request.failed" + | "combo.target.attempt" + | "combo.target.failed" + | "combo.target.succeeded" + | "credential.health.changed"; + +// ── Event Payloads ──────────────────────────────────────────────────────── + +export interface RequestStartedPayload { + id: string; + model: string; + provider: string; + timestamp: number; + comboName?: string; +} + +export interface RequestStreamingPayload { + id: string; + tokensGenerated: number; + elapsedMs: number; +} + +export interface RequestCompletedPayload { + id: string; + status: "success" | "error"; + model: string; + provider: string; + tokensInput: number; + tokensOutput: number; + latencyMs: number; + comboName?: string; + error?: string; +} + +export interface RequestFailedPayload { + id: string; + error: string; + statusCode?: number; + latencyMs: number; + model?: string; + provider?: string; +} + +export interface ComboTargetAttemptPayload { + comboName: string; + targetIndex: number; + provider: string; + model: string; + timestamp: number; + strategy: string; +} + +export interface ComboTargetFailedPayload { + comboName: string; + targetIndex: number; + provider: string; + model: string; + error: string; + latencyMs: number; +} + +export interface ComboTargetSucceededPayload { + comboName: string; + targetIndex: number; + provider: string; + model: string; + latencyMs: number; +} + +export interface CredentialHealthChangedPayload { + connectionId: string; + provider: string; + oldStatus: string; + newStatus: string; + timestamp: number; +} + +// ── Event Map ───────────────────────────────────────────────────────────── + +export interface DashboardEventMap { + "request.started": RequestStartedPayload; + "request.streaming": RequestStreamingPayload; + "request.completed": RequestCompletedPayload; + "request.failed": RequestFailedPayload; + "combo.target.attempt": ComboTargetAttemptPayload; + "combo.target.failed": ComboTargetFailedPayload; + "combo.target.succeeded": ComboTargetSucceededPayload; + "credential.health.changed": CredentialHealthChangedPayload; +} + +// ── Event Bus Listener ──────────────────────────────────────────────────── + +export type DashboardEventListener = ( + payload: DashboardEventMap[E], +) => void; + +// ── Channel Definitions ─────────────────────────────────────────────────── + +/** Available subscription channels */ +export type DashboardChannel = "requests" | "combo" | "credentials"; + +/** Map channels to their events */ +export const CHANNEL_EVENTS: Record = { + requests: ["request.started", "request.streaming", "request.completed", "request.failed"], + combo: ["combo.target.attempt", "combo.target.failed", "combo.target.succeeded"], + credentials: ["credential.health.changed"], +}; + +/** Get channel for an event */ +export function getChannelForEvent( + event: DashboardEventName, +): DashboardChannel | undefined { + for (const [channel, events] of Object.entries(CHANNEL_EVENTS)) { + if (events.includes(event)) return channel as DashboardChannel; + } + return undefined; +} diff --git a/src/server/ws/liveServer.ts b/src/server/ws/liveServer.ts new file mode 100644 index 0000000000..9b8c6e8a7e --- /dev/null +++ b/src/server/ws/liveServer.ts @@ -0,0 +1,331 @@ +/** + * Live Dashboard WebSocket Server + * + * Separate process (runs alongside Next.js on port 20129). + * Forwards EventBus events to subscribed dashboard clients. + * + * Protocol: + * Client → Server: { type: "subscribe", channels: ["requests", "combo"] } + * Server → Client: { type: "event", channel: "requests", event: "request.started", data: {...} } + * Client → Server: { type: "ping" } + * Server → Client: { type: "pong" } + * Server → Client: { type: "welcome", version, sessionId, channels, backlog } + * Server → Client: { type: "error", code, message } + */ + +import { WebSocketServer, WebSocket } from "ws"; +import { createServer } from "http"; +import { randomUUID } from "crypto"; + +// ── Types ───────────────────────────────────────────────────────────────── + +import type { + WsClientMessage, + WsServerMessage, + WsAuthResult, +} from "./types"; + +import { + emit, + on, + onAny, + getEventHistory, + type HistoryEntry, +} from "@/lib/events/eventBus"; + +import type { + DashboardEventName, + DashboardEventMap, + DashboardChannel, +} from "@/lib/events/types"; + +import { CHANNEL_EVENTS, getChannelForEvent } from "@/lib/events/types"; + +// ── Config ──────────────────────────────────────────────────────────────── + +const DEFAULT_PORT = 20129; +const HEARTBEAT_INTERVAL_MS = 15_000; +const HEARTBEAT_TIMEOUT_MS = 35_000; +const MAX_CLIENTS = 500; +const MAX_EVENTS_PER_SECOND = 100; + +// ── Client State ────────────────────────────────────────────────────────── + +interface ClientState { + ws: WebSocket; + sessionId: string; + subscribedChannels: Set; + lastActivity: number; + /** Per-second rate limit counter */ + eventCounter: number; + eventCounterReset: number; + /** Current IP for rate limiting */ + remoteAddress: string; +} + +const clients = new Map(); +let eventHistoryBacklog: HistoryEntry[] = []; +const BACKLOG_MAX = 500; + +// ── Auth ────────────────────────────────────────────────────────────────── + +async function authorizeConnection( + request: import("http").IncomingMessage, +): Promise { + const url = new URL(request.url || "/", `http://${request.headers.host || "localhost"}`); + const token = url.searchParams.get("token") || extractBearerToken(request); + + const sessionId = randomUUID().slice(0, 8); + + if (!token) { + return { authorized: false, sessionId, error: "Missing token" }; + } + + try { + // Validate API key via the existing auth system + const { extractApiKey, isValidApiKey } = await import("../services/auth"); + const apiKey = extractApiKey({ headers: { authorization: `Bearer ${token}` } } as any); + + if (!apiKey || !isValidApiKey(apiKey)) { + return { authorized: false, sessionId, error: "Invalid API key" }; + } + + return { authorized: true, sessionId }; + } catch { + return { authorized: false, sessionId, error: "Auth system unavailable" }; + } +} + +function extractBearerToken(request: import("http").IncomingMessage): string | null { + const auth = request.headers["authorization"]; + if (!auth || typeof auth !== "string") return null; + const match = auth.match(/^Bearer\s+(.+)$/i); + return match?.[1] || null; +} + +// ── Protocol Handler ────────────────────────────────────────────────────── + +function handleMessage(clientId: string, raw: string): void { + const client = clients.get(clientId); + if (!client) return; + + // Rate limiting + const now = Date.now(); + if (now - client.eventCounterReset > 1000) { + client.eventCounter = 0; + client.eventCounterReset = now; + } + client.eventCounter++; + if (client.eventCounter > MAX_EVENTS_PER_SECOND) { + sendTo(client.ws, { type: "error", code: "RATE_LIMITED", message: "Too many messages" }); + return; + } + + let msg: WsClientMessage; + try { + msg = JSON.parse(raw); + } catch { + sendTo(client.ws, { type: "error", code: "PARSE_ERROR", message: "Invalid JSON" }); + return; + } + + client.lastActivity = now; + + switch (msg.type) { + case "subscribe": { + client.subscribedChannels = new Set(msg.channels); + + // Send buffered events that match subscribed channels + const relevantHistory = eventHistoryBacklog.filter((h) => { + const ch = getChannelForEvent(h.event as DashboardEventName); + return ch && msg.channels.includes(ch); + }); + + sendTo(client.ws, { + type: "welcome", + version: "1.0.0", + sessionId: client.sessionId, + serverTime: now, + channels: msg.channels, + backlog: relevantHistory.length, + data: relevantHistory.map((h) => ({ + event: h.event, + channel: getChannelForEvent(h.event as DashboardEventName), + data: h.payload, + timestamp: h.timestamp, + })), + } as any); + break; + } + + case "ping": + sendTo(client.ws, { type: "pong" } as WsServerMessage); + break; + } +} + +// ── Send ────────────────────────────────────────────────────────────────── + +function sendTo(ws: WebSocket, msg: WsServerMessage | Record): void { + if (ws.readyState === WebSocket.OPEN) { + ws.send(JSON.stringify(msg)); + } +} + +// ── Event Bus → WebSocket Bridge ────────────────────────────────────────── + +function subscribeToEventBus(): () => void { + return onAny((event: DashboardEventName, payload: unknown) => { + const channel = getChannelForEvent(event); + if (!channel) return; + + // Store in backlog + eventHistoryBacklog.push({ event, payload, timestamp: Date.now() }); + if (eventHistoryBacklog.length > BACKLOG_MAX) { + eventHistoryBacklog.shift(); + } + + // Forward to subscribed clients + const msg: WsEventMessage = { + type: "event", + channel, + event, + data: payload, + }; + + for (const [clientId, client] of clients) { + if (client.ws.readyState !== WebSocket.OPEN) { + clients.delete(clientId); + continue; + } + if (client.subscribedChannels.has(channel)) { + sendTo(client.ws, msg); + } + } + }); +} + +// ── Heartbeat ───────────────────────────────────────────────────────────── + +function startHeartbeat(server: WebSocketServer): void { + const interval = setInterval(() => { + const now = Date.now(); + for (const [clientId, client] of clients) { + if (client.ws.readyState !== WebSocket.OPEN) { + clients.delete(clientId); + continue; + } + // Check heartbeat timeout + if (now - client.lastActivity > HEARTBEAT_TIMEOUT_MS) { + client.ws.terminate(); + clients.delete(clientId); + continue; + } + // Send ping + sendTo(client.ws, { type: "pong" } as WsServerMessage); + } + }, HEARTBEAT_INTERVAL_MS); + + server.on("close", () => clearInterval(interval)); +} + +// ── Server Start ────────────────────────────────────────────────────────── + +/** + * Start the live dashboard WebSocket server. + */ +export async function startLiveDashboardServer( + port = DEFAULT_PORT, +): Promise { + const server = createServer(); + const wss = new WebSocketServer({ server }); + + // Subscribe to EventBus + const unsubscribe = subscribeToEventBus(); + + wss.on("connection", async (ws, request) => { + // Enforce max clients + if (clients.size >= MAX_CLIENTS) { + sendTo(ws, { type: "error", code: "SERVER_FULL", message: "Max clients reached" }); + ws.close(1013, "Server full"); + return; + } + + // Authorize + const auth = await authorizeConnection(request); + if (!auth.authorized) { + sendTo(ws, { type: "error", code: "UNAUTHORIZED", message: auth.error || "Unauthorized" }); + ws.close(4001, "Unauthorized"); + return; + } + + const clientId = auth.sessionId; + const client: ClientState = { + ws, + sessionId: clientId, + subscribedChannels: new Set(), + lastActivity: Date.now(), + eventCounter: 0, + eventCounterReset: Date.now(), + remoteAddress: request.socket?.remoteAddress || "unknown", + }; + + clients.set(clientId, client); + + console.log(`[LiveWS] Client connected: ${clientId} (${client.remoteAddress}) [${clients.size} total]`); + + // Handle messages + ws.on("message", (data) => { + handleMessage(clientId, data.toString()); + }); + + // Handle close + ws.on("close", () => { + clients.delete(clientId); + console.log(`[LiveWS] Client disconnected: ${clientId} [${clients.size} remaining]`); + }); + + // Handle errors + ws.on("error", (err) => { + console.error(`[LiveWS] Client error ${clientId}:`, err.message); + clients.delete(clientId); + }); + }); + + // Heartbeat + startHeartbeat(wss); + + // Cleanup on close + wss.on("close", () => { + unsubscribe(); + clients.clear(); + }); + + return new Promise((resolve) => { + server.listen(port, () => { + console.log(`[LiveWS] Dashboard WebSocket server listening on port ${port}`); + resolve(server); + }); + }); +} + +// ── Auto-start on import (if not in build/test) ─────────────────────────── + +function isBuildOrTest(): boolean { + return ( + process.env.NEXT_PHASE === "phase-production-build" || + process.env.NODE_ENV === "test" || + process.env.VITEST !== undefined || + process.argv.some((arg) => arg.includes("test")) || + process.env.OMNIROUTE_DISABLE_LIVE_WS === "1" || + process.env.OMNIROUTE_DISABLE_LIVE_WS === "true" + ); +} + +// Auto-start unless disabled +if (!isBuildOrTest()) { + const port = parseInt(process.env.LIVE_WS_PORT || String(DEFAULT_PORT), 10); + startLiveDashboardServer(port).catch((err) => { + console.error("[LiveWS] Failed to start:", err); + }); +} diff --git a/src/server/ws/types.ts b/src/server/ws/types.ts new file mode 100644 index 0000000000..61f319e63f --- /dev/null +++ b/src/server/ws/types.ts @@ -0,0 +1,61 @@ +/** + * WebSocket Live Dashboard — Protocol Types + * + * Wire format for the real-time dashboard WebSocket. + */ + +// ── Client → Server ─────────────────────────────────────────────────────── + +export interface WsSubscribeMessage { + type: "subscribe"; + channels: Array<"requests" | "combo" | "credentials">; +} + +export interface WsPingMessage { + type: "ping"; +} + +export type WsClientMessage = WsSubscribeMessage | WsPingMessage; + +// ── Server → Client ─────────────────────────────────────────────────────── + +export interface WsEventMessage { + type: "event"; + channel: "requests" | "combo" | "credentials"; + event: string; + data: unknown; +} + +export interface WsPongMessage { + type: "pong"; +} + +export interface WsWelcomeMessage { + type: "welcome"; + version: string; + sessionId: string; + serverTime: number; + channels: Array<"requests" | "combo" | "credentials">; + /** Number of buffered events since last reconnect */ + backlog: number; +} + +export interface WsErrorMessage { + type: "error"; + code: string; + message: string; +} + +export type WsServerMessage = + | WsEventMessage + | WsPongMessage + | WsWelcomeMessage + | WsErrorMessage; + +// ── Auth ───────────────────────────────────────────────────────────────── + +export interface WsAuthResult { + authorized: boolean; + sessionId: string; + error?: string; +} diff --git a/tests/unit/universal-handoff.test.ts b/tests/unit/universal-handoff.test.ts index 6fb0b9f17d..8fe1f14706 100644 --- a/tests/unit/universal-handoff.test.ts +++ b/tests/unit/universal-handoff.test.ts @@ -38,10 +38,7 @@ test("applies global-level config when combo not set", () => { }); test("gives combo priority over global", () => { - const r = resolveUniversalHandoffConfig( - { enabled: true } as any, - { enabled: false } as any - ); + const r = resolveUniversalHandoffConfig({ enabled: true } as any, { enabled: false } as any); assert.strictEqual(r.enabled, true); }); @@ -130,11 +127,13 @@ test("buildUniversalHandoffSystemMessage full XML with valid payload", () => { test("buildUniversalHandoffSystemMessage escapes XML special chars", () => { const msg = buildUniversalHandoffSystemMessage( - 'm', 'm', 'r "t" & x', + "m", + "m", + 'r "t" & x', makePayload({ summary: ' & "q"', keyDecisions: ['d & "<>"'], - activeEntities: ['e'], + activeEntities: ["e"], }) ); assert.ok(msg.includes("m<a>"));