Validate realtime browser events

docs/REALTIME_SESSION.md

@@ -104,6 +104,8 @@ See [`protocol.ts`](../src/lib/core/protocol.ts) for the authoritative, current
 - **Client → server**: `session.start`, `input_audio.chunk`, `input_audio.commit`, `input_text.send`, `response.cancel`, `session.end`, `debug.subscribe`
 - **Server → client**: `session.ready`, `output_audio.chunk`, `output_audio.done`, `transcript.delta`, `transcript.done`, `response.done`, `error`, `session.closed`, `debug.subscribe.result`, `debug.log`
 
+Parsed browser JSON is runtime-validated against the client-event contract before it reaches analytics or a provider adapter; the TypeScript union alone is not a wire-level guarantee. Missing or wrongly typed required fields are rejected with `invalid_client_event`, while an unexpected throw in the async message path closes only that connection with `internal_error`. This containment is process-critical: one Node process serves every realtime session.
+
 `debug.subscribe`/`debug.subscribe.result`/`debug.log` are a separate concern from the rest of this vocabulary — ephemeral developer debug-log forwarding, not session content. See [LOGGING.md](LOGGING.md).
 
 ## OpenAI Realtime API gotcha (verified live, Aug 2026)

src/lib/core/protocol.ts

@@ -40,6 +40,45 @@ export type ClientEvent =
 	// DEBUG_ACCESS_KEY) and ignored in dev, where forwarding is automatic.
 	| { type: 'debug.subscribe'; key?: string };
 
+/** Events accepted from the browser websocket (excludes server-internal control events). */
+export type BrowserClientEvent = Exclude<ClientEvent, { type: 'response.trigger' }>;
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+	return typeof value === 'object' && value !== null && !Array.isArray(value);
+}
+
+/**
+ * Runtime validation for the browser websocket trust boundary. TypeScript's
+ * `ClientEvent` union disappears at runtime, so parsed JSON must be checked
+ * before it reaches analytics storage or a provider adapter.
+ */
+export function isBrowserClientEvent(value: unknown): value is BrowserClientEvent {
+	if (!isRecord(value)) return false;
+
+	switch (value.type) {
+		case 'session.start':
+			return (
+				value.config === undefined ||
+				(isRecord(value.config) &&
+					(value.config.voice === undefined || typeof value.config.voice === 'string'))
+			);
+		case 'input_audio.chunk':
+			return typeof value.audio === 'string';
+		case 'input_text.send':
+			return typeof value.text === 'string';
+		case 'tool.result':
+			return typeof value.callId === 'string' && typeof value.output === 'string';
+		case 'debug.subscribe':
+			return value.key === undefined || typeof value.key === 'string';
+		case 'input_audio.commit':
+		case 'response.cancel':
+		case 'session.end':
+			return true;
+		default:
+			return false;
+	}
+}
+
 export type RealtimeErrorCode =
 	| 'auth_expired'
 	| 'invalid_client_event'

src/lib/server/core/session.spec.ts

@@ -3,7 +3,7 @@ import { afterEach, describe, expect, it, vi } from 'vitest';
 import type WebSocket from 'ws';
 import type { ClientEvent, ServerEvent } from '../../core/protocol.js';
 import type { ToolDefinition } from '../../core/tools.js';
-import type { AnalyticsRecorder } from './analytics.js';
+import { SqliteAnalyticsRecorder, type AnalyticsRecorder } from './analytics.js';
 import type { ProviderSession, RealtimeProvider } from './provider.js';
 import { createRealtimeSession, resolveSessionConfig } from './session.js';
 import { createToolRegistry } from './tools.js';
@@ -36,7 +36,7 @@ class FakeBrowserSocket extends EventEmitter {
 		return this.sentRaw.map((raw) => JSON.parse(raw));
 	}
 
-	receive(event: ClientEvent) {
+	receive(event: unknown) {
 		this.emit('message', Buffer.from(JSON.stringify(event)), false);
 	}
 }
@@ -210,6 +210,83 @@ describe('RealtimeSession', () => {
 		);
 	});
 
+	it('rejects input_text.send without text before it can violate SQLite constraints', async () => {
+		const provider = new FakeProvider();
+		const recorder = new SqliteAnalyticsRecorder(':memory:', {
+			messages: true,
+			toolCalls: true,
+			events: true
+		});
+		const { socket, asWs } = makeSocket();
+		createRealtimeSession(asWs, provider, DEFAULTS, recorder, TEST_TOOL_TIMEOUT_MS);
+
+		socket.receive({ type: 'session.start' });
+		await vi.waitFor(() => expect(provider.sessions).toHaveLength(1));
+
+		socket.receive({ type: 'input_text.send' });
+		await vi.waitFor(() =>
+			expect(socket.sent).toContainEqual({
+				type: 'error',
+				code: 'invalid_client_event',
+				message: 'Invalid client event.'
+			})
+		);
+		expect(provider.sessions[0].sent).not.toContainEqual(
+			expect.objectContaining({ type: 'input_text.send' })
+		);
+		expect(recorder.rawDb.prepare('SELECT * FROM messages').all()).toEqual([]);
+
+		// The malformed frame is isolated: this same session remains usable.
+		socket.receive({ type: 'input_text.send', text: 'still alive' });
+		await vi.waitFor(() =>
+			expect(provider.sessions[0].sent).toContainEqual({
+				type: 'input_text.send',
+				text: 'still alive'
+			})
+		);
+		expect(recorder.rawDb.prepare('SELECT text FROM messages').all()).toEqual([
+			{ text: 'still alive' }
+		]);
+
+		socket.emit('close');
+		recorder.close();
+	});
+
+	it('contains unexpected async message failures and leaves other sessions usable', async () => {
+		const provider = new FakeProvider();
+		const recorder = new FakeAnalyticsRecorder();
+		recorder.messageLogged = () => {
+			throw new Error('analytics unavailable');
+		};
+		const affected = makeSocket();
+		const other = makeSocket();
+		createRealtimeSession(affected.asWs, provider, DEFAULTS, recorder, TEST_TOOL_TIMEOUT_MS);
+		createRealtimeSession(other.asWs, provider, DEFAULTS, recorder, TEST_TOOL_TIMEOUT_MS);
+
+		affected.socket.receive({ type: 'session.start' });
+		await vi.waitFor(() => expect(provider.sessions).toHaveLength(1));
+		other.socket.receive({ type: 'session.start' });
+		await vi.waitFor(() => expect(provider.sessions).toHaveLength(2));
+		affected.socket.receive({ type: 'input_text.send', text: 'hello' });
+
+		await vi.waitFor(() =>
+			expect(affected.socket.sent).toContainEqual({
+				type: 'error',
+				code: 'internal_error',
+				message: 'Failed to handle client event.'
+			})
+		);
+		expect(provider.sessions[0].closed).toBe(true);
+		expect(provider.sessions[0].closeReason).toBe('internal_error');
+		expect(affected.socket.closeCalls).toBe(1);
+		expect(provider.sessions[1].closed).toBe(false);
+
+		other.socket.receive({ type: 'response.cancel' });
+		await vi.waitFor(() =>
+			expect(provider.sessions[1].sent).toContainEqual({ type: 'response.cancel' })
+		);
+	});
+
 	it('relays provider server events back to the browser socket', async () => {
 		const provider = new FakeProvider();
 		const recorder = new FakeAnalyticsRecorder();

src/lib/server/core/session.ts

@@ -1,7 +1,13 @@
 import { randomUUID } from 'node:crypto';
 import WebSocket from 'ws';
 import type { LogEntry, Logger } from '../../core/logger.js';
-import type { ClientEvent, ServerEvent, SessionStartConfig } from '../../core/protocol.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';
@@ -79,7 +85,9 @@ export class RealtimeSession {
 		this.debugSubscribed = isDev;
 		this.logger = createLogger((entry) => this.forwardDebugLog(entry), { logMessages });
 		this.browserSocket.on('message', (data, isBinary) => {
-			void this.handleBrowserMessage(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'));
@@ -100,9 +108,9 @@ export class RealtimeSession {
 			return;
 		}
 
-		let event: ClientEvent;
+		let parsed: unknown;
 		try {
-			event = JSON.parse(data.toString());
+			parsed = JSON.parse(data.toString());
 		} catch {
 			this.logger.warn('ws', 'Rejected malformed JSON from browser.');
 			this.sendToBrowser({
@@ -112,6 +120,16 @@ export class RealtimeSession {
 			});
 			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);
@@ -156,6 +174,39 @@ export class RealtimeSession {
 		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<void> {
 		if (this.state !== 'idle') {
 			this.sendToBrowser({