import { randomUUID } from 'node:crypto'; import WebSocket from 'ws'; import type { LogEntry, Logger } from '../../core/logger.js'; import { isBrowserClientEvent, type BrowserClientEvent, type ClientEvent, type ServerEvent, type SessionStartConfig } from '../../core/protocol.js'; import { formatToolError } from '../../core/tools.js'; import type { AnalyticsRecorder } from './analytics.js'; import { createLogger, safeSerialize, verifyDebugAccessKey } from './logger.js'; import type { ProviderSession, RealtimeProvider, ResolvedSessionConfig } from './provider.js'; import type { ToolRegistry } from './tools.js'; export type RealtimeSessionState = 'idle' | 'connecting' | 'active' | 'closing' | 'closed'; /** Server-owned policy for a realtime session. Never client-controlled. */ export interface SessionDefaults { model: string; instructions: string; /** Used only when the browser's `session.start` doesn't specify a voice. */ defaultVoice?: string; toolRegistry: ToolRegistry; } /** * Merges the browser's (deliberately tiny) client-visible config into the * server-owned defaults. `instructions`/`model`/`tools` always come from * `defaults` — this is the trust boundary that keeps a shared relay safe to * expose to untrusted browsers. */ export function resolveSessionConfig( defaults: SessionDefaults, clientConfig?: SessionStartConfig ): ResolvedSessionConfig { return { model: defaults.model, instructions: defaults.instructions, voice: clientConfig?.voice ?? defaults.defaultVoice, tools: defaults.toolRegistry.definitions }; } /** * Owns the lifecycle of one browser WS connection: turns its first * `session.start` into a `ProviderSession` via the injected `RealtimeProvider`, * pipes events both directions afterward, and guarantees that either side * closing/erroring tears down the other. */ export class RealtimeSession { private state: RealtimeSessionState = 'idle'; private providerSession: ProviderSession | null = null; /** Tool calls currently awaiting resolution in the current turn (server executor pending or client round-trip pending). */ private outstandingToolCalls = 0; /** Client-executed tool calls forwarded to the browser, awaiting `tool.result`. */ private readonly pendingClientToolCalls = new Map(); /** Set once `start()` reaches 'active'; null before then and stays that way if the session never gets that far. */ private sessionId: string | null = null; /** Populated at the top of handleToolCall (before any branching) so even the unknown-tool path is covered; consumed in resolveToolCall. */ private readonly toolCallMeta = new Map< string, { name: string; arguments: string; startedAt: number } >(); /** Assistant voice replies stream as many transcript.delta chunks then one text-less transcript.done — see docs/ANALYTICS.md. */ private assistantTranscriptBuffer = ''; /** True once this connection is allowed to receive `debug.log` events — see docs/LOGGING.md. Starts true in dev (frictionless), false in prod until a valid `debug.subscribe`. */ private debugSubscribed = false; /** Forwards to this session's own browser only when `debugSubscribed` — see forwardDebugLog. Also mirrors to console, except the `messages` category when `DEBUG_LOG_MESSAGES=false` — see createLogger. */ private readonly logger: Logger; constructor( private readonly browserSocket: WebSocket, private readonly provider: RealtimeProvider, private readonly defaults: SessionDefaults, private readonly recorder: AnalyticsRecorder, /** How long a client-executed tool call may go unanswered before we give up on it. */ private readonly clientToolTimeoutMs: number, /** Dev builds forward debug logs automatically; prod requires `debug.subscribe` with a valid key. See docs/LOGGING.md. */ private readonly isDev: boolean = false, /** Gates the `messages` category's console mirror only — browser forwarding is unaffected. See `DEBUG_LOG_MESSAGES` in config.ts and docs/LOGGING.md. */ logMessages: boolean = true ) { this.debugSubscribed = isDev; this.logger = createLogger((entry) => this.forwardDebugLog(entry), { logMessages }); this.browserSocket.on('message', (data, isBinary) => { void this.handleBrowserMessage(data, isBinary).catch((error) => { this.handleBrowserMessageFailure(error); }); }); this.browserSocket.on('close', () => this.close('client_disconnected')); this.browserSocket.on('error', () => this.close('client_disconnected')); } getState(): RealtimeSessionState { return this.state; } private async handleBrowserMessage(data: WebSocket.RawData, isBinary: boolean): Promise { if (isBinary) { this.logger.warn('ws', 'Rejected binary frame from browser.'); this.sendToBrowser({ type: 'error', code: 'invalid_client_event', message: 'Binary frames are not supported yet.' }); return; } let parsed: unknown; try { parsed = JSON.parse(data.toString()); } catch { this.logger.warn('ws', 'Rejected malformed JSON from browser.'); this.sendToBrowser({ type: 'error', code: 'invalid_client_event', message: 'Malformed JSON.' }); return; } if (!isBrowserClientEvent(parsed)) { this.logger.warn('ws', 'Rejected invalid client event.'); this.sendToBrowser({ type: 'error', code: 'invalid_client_event', message: 'Invalid client event.' }); return; } const event: BrowserClientEvent = parsed; if (event.type === 'session.start') { await this.start(event.config); return; } if (event.type === 'session.end') { this.close('client_requested'); return; } if (event.type === 'debug.subscribe') { this.handleDebugSubscribe(event.key); return; } if (this.state !== 'active' || !this.providerSession) { this.logger.warn('ws', `Rejected '${event.type}' before session.start.`); this.sendToBrowser({ type: 'error', code: 'invalid_client_event', message: `Cannot handle '${event.type}' before session.start.` }); return; } if (event.type === 'tool.result') { const timeout = this.pendingClientToolCalls.get(event.callId); if (!timeout) return; // stale/unknown callId (e.g. already timed out) — ignore clearTimeout(timeout); this.pendingClientToolCalls.delete(event.callId); this.resolveToolCall(event.callId, event.output); return; } if (event.type === 'input_text.send') { // Typed text is always already a complete string — unlike the voice // path, no buffering is needed. this.logger.info('messages', 'User message', { role: 'user', text: event.text }); this.recordAnalytics('message', () => { this.recorder.messageLogged({ sessionId: this.sessionId!, role: 'user', text: event.text, loggedAt: Date.now() }); }); } this.providerSession.send(event); } /** * Last-resort containment for unexpected failures in the async websocket * message path. A single connection must never create an unhandled * rejection in the process shared by every realtime session. */ private handleBrowserMessageFailure(error: unknown): void { let message = 'Unknown error'; try { message = error instanceof Error ? error.message : String(error); } catch { // Even a hostile thrown value with a broken toString stays contained. } try { this.logger.error('ws', 'Unexpected failure handling browser message.', { message }); } catch { // Error reporting itself must not escape this rejection boundary. } try { this.sendToBrowser({ type: 'error', code: 'internal_error', message: 'Failed to handle client event.' }); } catch { // A failing analytics/logging sink may be the original cause. } try { this.close('internal_error'); } catch { // Keep the failure scoped to this connection even if teardown throws. } } private async start(clientConfig?: SessionStartConfig): Promise { if (this.state !== 'idle') { this.sendToBrowser({ type: 'error', code: 'invalid_client_event', message: 'Session already started.' }); return; } this.state = 'connecting'; const config = resolveSessionConfig(this.defaults, clientConfig); let providerSession: ProviderSession; try { providerSession = await this.provider.createSession(config); } catch (err) { this.state = 'closed'; const message = err instanceof Error ? err.message : 'Failed to start session.'; this.logger.error('session', 'Failed to start session.', { message }); this.sendToBrowser({ type: 'error', code: 'upstream_unavailable', message }); this.browserSocket.close(); return; } // The browser may have disconnected (session.end / close) while we were // awaiting the provider handshake — don't leak the upstream connection. const stateAfterConnect: RealtimeSessionState = this.state; if (stateAfterConnect !== 'connecting') { providerSession.close('client_requested'); return; } this.providerSession = providerSession; const sessionId = randomUUID(); this.sessionId = sessionId; this.state = 'active'; this.logger.info('session', 'Session started.', { sessionId, model: config.model, voice: config.voice }); // Logged once here, alongside the user/assistant message logging below, // so the debug panel can show the full conversation context — the // system prompt is otherwise invisible while debugging a live session. this.logger.info('messages', 'System instructions', { role: 'system', text: config.instructions }); this.recordAnalytics('session start', () => { this.recorder.sessionStarted({ sessionId, model: config.model, voice: config.voice, startedAt: Date.now() }); }); // Recorded above before wiring onEvent, so there's no window where a // provider event could reach handleProviderEvent before sessionId is set. this.providerSession.onEvent((event) => this.handleProviderEvent(event)); this.sendToBrowser({ type: 'session.ready', sessionId }); } private handleProviderEvent(event: ServerEvent): void { if (event.type === 'tool.call') { void this.handleToolCall(event); return; } if (event.type === 'transcript.delta') { if (event.role === 'assistant') { this.assistantTranscriptBuffer += event.text; } else { // User voice transcripts arrive as a single already-complete chunk // (conversation.item.input_audio_transcription.completed in // providers/openai.ts) with no corresponding // transcript.done{role:'user'} ever emitted today — log // immediately rather than buffering. Rests on unverified live // behavior; see docs/ANALYTICS.md. this.logger.info('messages', 'User message', { role: 'user', text: event.text }); this.recordAnalytics('message', () => { this.recorder.messageLogged({ sessionId: this.sessionId!, role: 'user', text: event.text, loggedAt: Date.now() }); }); } this.sendToBrowser(event); return; } if (event.type === 'transcript.done' && event.role === 'assistant') { const text = this.assistantTranscriptBuffer; this.assistantTranscriptBuffer = ''; this.logger.info('messages', 'Assistant message', { role: 'assistant', text }); this.recordAnalytics('message', () => { this.recorder.messageLogged({ sessionId: this.sessionId!, role: 'assistant', text, loggedAt: Date.now() }); }); this.sendToBrowser(event); return; } if (event.type === 'session.closed') { // Forward first, then close: close() moves the browser socket out of // OPEN, and sendToBrowser no-ops once that happens, so closing first // would silently swallow this notification. this.sendToBrowser(event); this.close('upstream_disconnected'); return; } this.sendToBrowser(event); } /** * Routes a tool call by its registered `location`: a server tool is * executed here directly (the browser never sees it); a client tool is * forwarded to the browser and tracked with a timeout so an unresponsive * tab can't hang the model's turn forever. */ private async handleToolCall(event: Extract): Promise { this.outstandingToolCalls++; this.toolCallMeta.set(event.callId, { name: event.name, arguments: event.arguments, startedAt: Date.now() }); this.logger.debug('tools', `Tool call started: ${event.name}`, { callId: event.callId }); const definition = this.defaults.toolRegistry.definitions.find((d) => d.name === event.name); if (!definition) { this.logger.warn('tools', `Unknown tool "${event.name}".`, { callId: event.callId }); this.resolveToolCall(event.callId, formatToolError(`Unknown tool "${event.name}".`)); return; } if (definition.location === 'client') { this.sendToBrowser(event); const timeout = setTimeout(() => { this.pendingClientToolCalls.delete(event.callId); this.logger.warn('tools', `Tool call timed out: ${event.name}`, { callId: event.callId }); this.resolveToolCall(event.callId, formatToolError('Client tool call timed out.')); }, this.clientToolTimeoutMs); this.pendingClientToolCalls.set(event.callId, timeout); return; } let args: Record; try { args = JSON.parse(event.arguments); } catch { this.resolveToolCall(event.callId, formatToolError('Malformed tool arguments.')); return; } try { const executor = this.defaults.toolRegistry.executors.get(event.name); const result = await executor!(args); this.resolveToolCall(event.callId, JSON.stringify(result)); } catch (err) { this.resolveToolCall( event.callId, formatToolError(err instanceof Error ? err.message : 'Tool execution failed.') ); } } /** * Sends a tool's result to the provider and, once every tool call * outstanding in the current turn has resolved, explicitly triggers the * model to continue — see the `response.trigger` note in protocol.ts for * why this can't just be bundled with each individual result. */ private resolveToolCall(callId: string, output: string): void { this.sendToProvider({ type: 'tool.result', callId, output }); const meta = this.toolCallMeta.get(callId); if (meta) { // Should always be present given handleToolCall sets it for every // path (including the unknown-tool branch) before this can run — // guarded defensively rather than asserted, so a future refactor // that breaks the invariant fails soft here instead of crashing. this.toolCallMeta.delete(callId); this.logger.debug('tools', `Tool call completed: ${meta.name}`, { callId, durationMs: Date.now() - meta.startedAt }); this.recordAnalytics('tool call', () => { this.recorder.toolCallLogged({ sessionId: this.sessionId!, callId, name: meta.name, arguments: meta.arguments, output, durationMs: Date.now() - meta.startedAt, loggedAt: Date.now() }); }); } this.outstandingToolCalls--; if (this.outstandingToolCalls === 0) { this.sendToProvider({ type: 'response.trigger' }); } } private close(reason: string): void { if (this.state === 'closing' || this.state === 'closed') return; this.state = 'closing'; this.logger.info('session', `Session closed: ${reason}`); for (const timeout of this.pendingClientToolCalls.values()) clearTimeout(timeout); this.pendingClientToolCalls.clear(); this.providerSession?.close(reason); this.providerSession = null; if (this.browserSocket.readyState === WebSocket.OPEN) { this.browserSocket.close(); } this.state = 'closed'; if (this.sessionId) { // Guarded: sessionId is still null if the browser disconnected // mid-handshake, before start() ever reached 'active' — no sessions // row exists yet to update in that case. this.recordAnalytics('session end', () => { this.recorder.sessionEnded({ sessionId: this.sessionId!, endedAt: Date.now(), reason }); }); } } private sendToBrowser(event: ServerEvent): void { if (event.type === 'error' || event.type === 'session.closed') { // Logged before the delivery guard below: the event "happened" from // the session's point of view even if the browser socket is already // dead by the time we try to deliver it. this.recordAnalytics('event', () => { this.recorder.eventLogged({ sessionId: this.sessionId, type: event.type, detail: JSON.stringify(event), loggedAt: Date.now() }); }); } if (this.browserSocket.readyState !== WebSocket.OPEN) return; this.browserSocket.send(JSON.stringify(event)); } /** * Analytics is best-effort telemetry, never part of session correctness. * A full disk, locked database, or broken recorder must not interrupt * provider traffic, browser delivery, or teardown. */ private recordAnalytics(operation: string, write: () => void): void { try { write(); } catch (error) { let message = 'Unknown error'; try { message = error instanceof Error ? error.message : String(error); } catch { // Even a hostile thrown value with a broken toString stays contained. } try { this.logger.error('analytics', `Failed to record ${operation}.`, { message }); } catch { // Telemetry failure reporting is best-effort too. } } } /** * Verifies a `debug.subscribe` request and replies with the result. In * dev, any request succeeds (frictionless — see docs/LOGGING.md); in * prod, `key` must match `DEBUG_ACCESS_KEY`. */ private handleDebugSubscribe(key: string | undefined): void { this.debugSubscribed = this.isDev || verifyDebugAccessKey(key); this.sendToBrowser({ type: 'debug.subscribe.result', ok: this.debugSubscribed, reason: this.debugSubscribed ? undefined : 'Invalid or missing debug access key.' }); } /** * Forwards a log entry to this session's own browser only — never another * session's, since each `RealtimeSession` instance owns exactly one * `browserSocket`. No-ops until `debugSubscribed`. Never lets a bad log * entry (e.g. unserializable `data`) affect the session itself. */ private forwardDebugLog(entry: LogEntry): void { if (!this.debugSubscribed) return; try { this.sendToBrowser({ type: 'debug.log', ...entry, data: safeSerialize(entry.data) }); } catch { // Forwarding a debug entry must never break real session traffic. } } // Guarded the same way sendToBrowser is: tool resolution can now reach // the provider from an async callback (executor await, timeout) rather // than only from a browser message already known to be in the 'active' // state, so providerSession may have gone null by the time this runs. private sendToProvider(event: ClientEvent): void { this.providerSession?.send(event); } } export function createRealtimeSession( browserSocket: WebSocket, provider: RealtimeProvider, defaults: SessionDefaults, recorder: AnalyticsRecorder, clientToolTimeoutMs: number, isDev = false, logMessages = true ): RealtimeSession { return new RealtimeSession( browserSocket, provider, defaults, recorder, clientToolTimeoutMs, isDev, logMessages ); }