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

import { EventEmitter } from 'node:events';
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 { ProviderSession, RealtimeProvider } from './provider.js';
import { createRealtimeSession, resolveSessionConfig } from './session.js';
import { createToolRegistry } from './tools.js';

const DEFAULTS = {
	model: 'test-model',
	instructions: 'be helpful',
	toolRegistry: createToolRegistry([], {})
};

/** Minimal stand-in for a `ws` WebSocket — only what RealtimeSession touches. */
class FakeBrowserSocket extends EventEmitter {
	readyState = 1; // WebSocket.OPEN
	sentRaw: string[] = [];
	closeCalls = 0;

	send(data: string) {
		this.sentRaw.push(data);
	}

	close() {
		this.closeCalls++;
		this.readyState = 3; // WebSocket.CLOSED
	}

	get sent(): ServerEvent[] {
		return this.sentRaw.map((raw) => JSON.parse(raw));
	}

	receive(event: ClientEvent) {
		this.emit('message', Buffer.from(JSON.stringify(event)), false);
	}
}

class FakeProviderSession implements ProviderSession {
	sent: ClientEvent[] = [];
	closed = false;
	closeReason: string | undefined;
	private handler: ((event: ServerEvent) => void) | null = null;

	send(event: ClientEvent): void {
		this.sent.push(event);
	}

	onEvent(handler: (event: ServerEvent) => void): void {
		this.handler = handler;
	}

	close(reason?: string): void {
		this.closed = true;
		this.closeReason = reason;
	}

	emitToOrchestrator(event: ServerEvent): void {
		this.handler?.(event);
	}
}

class FakeProvider implements RealtimeProvider {
	sessions: FakeProviderSession[] = [];
	private readonly gate: Promise<void>;
	private release!: () => void;

	constructor(opts: { deferred?: boolean } = {}) {
		this.gate = opts.deferred
			? new Promise((resolve) => (this.release = resolve))
			: Promise.resolve();
	}

	async createSession(): Promise<ProviderSession> {
		await this.gate;
		const session = new FakeProviderSession();
		this.sessions.push(session);
		return session;
	}

	releaseDeferred(): void {
		this.release();
	}
}

function makeSocket() {
	const socket = new FakeBrowserSocket();
	return { socket, asWs: socket as unknown as WebSocket };
}

describe('resolveSessionConfig', () => {
	it('always uses server-owned model/instructions/tools, merging only the client-visible voice', () => {
		expect(resolveSessionConfig(DEFAULTS, { voice: 'coral' })).toEqual({
			model: 'test-model',
			instructions: 'be helpful',
			voice: 'coral',
			tools: []
		});
	});

	it('is safe to call with no client config at all', () => {
		expect(resolveSessionConfig(DEFAULTS)).toEqual({
			model: 'test-model',
			instructions: 'be helpful',
			voice: undefined,
			tools: []
		});
	});
});

describe('RealtimeSession', () => {
	it('starts a provider session on session.start and replies with session.ready', async () => {
		const provider = new FakeProvider();
		const { socket, asWs } = makeSocket();
		createRealtimeSession(asWs, provider, DEFAULTS);

		socket.receive({ type: 'session.start' });
		await vi.waitFor(() => expect(provider.sessions).toHaveLength(1));
		await vi.waitFor(() =>
			expect(socket.sent).toContainEqual(expect.objectContaining({ type: 'session.ready' }))
		);
	});

	it('rejects content events sent before session.start', async () => {
		const provider = new FakeProvider();
		const { socket, asWs } = makeSocket();
		createRealtimeSession(asWs, provider, DEFAULTS);

		socket.receive({ type: 'input_text.send', text: 'too early' });
		await vi.waitFor(() =>
			expect(socket.sent).toContainEqual(
				expect.objectContaining({ type: 'error', code: 'invalid_client_event' })
			)
		);
		expect(provider.sessions).toHaveLength(0);
	});

	it('forwards events to the provider session once active', async () => {
		const provider = new FakeProvider();
		const { socket, asWs } = makeSocket();
		createRealtimeSession(asWs, provider, DEFAULTS);

		socket.receive({ type: 'session.start' });
		await vi.waitFor(() => expect(provider.sessions).toHaveLength(1));

		socket.receive({ type: 'input_text.send', text: 'hello' });
		await vi.waitFor(() =>
			expect(provider.sessions[0].sent).toContainEqual({ type: 'input_text.send', text: 'hello' })
		);
	});

	it('relays provider server events back to the browser socket', async () => {
		const provider = new FakeProvider();
		const { socket, asWs } = makeSocket();
		createRealtimeSession(asWs, provider, DEFAULTS);

		socket.receive({ type: 'session.start' });
		await vi.waitFor(() => expect(provider.sessions).toHaveLength(1));

		provider.sessions[0].emitToOrchestrator({
			type: 'transcript.delta',
			role: 'assistant',
			text: 'hi'
		});
		await vi.waitFor(() =>
			expect(socket.sent).toContainEqual({
				type: 'transcript.delta',
				role: 'assistant',
				text: 'hi'
			})
		);
	});

	it('closes the provider session when the browser disconnects', async () => {
		const provider = new FakeProvider();
		const { socket, asWs } = makeSocket();
		createRealtimeSession(asWs, provider, DEFAULTS);

		socket.receive({ type: 'session.start' });
		await vi.waitFor(() => expect(provider.sessions).toHaveLength(1));

		socket.emit('close');
		await vi.waitFor(() => expect(provider.sessions[0].closed).toBe(true));
		expect(provider.sessions[0].closeReason).toBe('client_disconnected');
	});

	it('closes the provider session on session.end and closes the browser socket', async () => {
		const provider = new FakeProvider();
		const { socket, asWs } = makeSocket();
		createRealtimeSession(asWs, provider, DEFAULTS);

		socket.receive({ type: 'session.start' });
		await vi.waitFor(() => expect(provider.sessions).toHaveLength(1));

		socket.receive({ type: 'session.end' });
		await vi.waitFor(() => expect(provider.sessions[0].closed).toBe(true));
		expect(socket.closeCalls).toBe(1);
	});

	it('does not leak an upstream session if the browser disconnects mid-handshake', async () => {
		const provider = new FakeProvider({ deferred: true });
		const { socket, asWs } = makeSocket();
		createRealtimeSession(asWs, provider, DEFAULTS);

		socket.receive({ type: 'session.start' });
		socket.emit('close'); // disconnect while the provider handshake is still pending
		provider.releaseDeferred();

		await vi.waitFor(() => expect(provider.sessions).toHaveLength(1));
		await vi.waitFor(() => expect(provider.sessions[0].closed).toBe(true));
	});
});

const FAST_TOOL: ToolDefinition = {
	name: 'fast_tool',
	description: 'Resolves immediately.',
	parameters: { type: 'object', properties: {}, required: [] },
	location: 'server'
};
const SLOW_TOOL: ToolDefinition = {
	name: 'slow_tool',
	description: 'Resolves once released.',
	parameters: { type: 'object', properties: {}, required: [] },
	location: 'server'
};
const CLIENT_TOOL: ToolDefinition = {
	name: 'client_tool',
	description: 'Executed in the browser.',
	parameters: { type: 'object', properties: {}, required: [] },
	location: 'client'
};

async function startActiveSession(defaults: typeof DEFAULTS) {
	const provider = new FakeProvider();
	const { socket, asWs } = makeSocket();
	createRealtimeSession(asWs, provider, defaults);
	socket.receive({ type: 'session.start' });
	await vi.waitFor(() => expect(provider.sessions).toHaveLength(1));
	return { provider, session: provider.sessions[0], socket };
}

describe('RealtimeSession tool calls', () => {
	afterEach(() => {
		vi.useRealTimers();
	});

	it('executes a server tool and sends its result plus response.trigger to the provider', async () => {
		const toolRegistry = createToolRegistry([FAST_TOOL], {
			fast_tool: async () => ({ done: true })
		});
		const { session } = await startActiveSession({ ...DEFAULTS, toolRegistry });

		session.emitToOrchestrator({
			type: 'tool.call',
			callId: 'call-1',
			name: 'fast_tool',
			arguments: '{}'
		});

		await vi.waitFor(() =>
			expect(session.sent).toContainEqual({
				type: 'tool.result',
				callId: 'call-1',
				output: JSON.stringify({ done: true })
			})
		);
		await vi.waitFor(() => expect(session.sent).toContainEqual({ type: 'response.trigger' }));
	});

	it('only triggers response.trigger once every parallel tool call has resolved', async () => {
		let releaseSlow!: () => void;
		const slowGate = new Promise<void>((resolve) => (releaseSlow = resolve));
		const toolRegistry = createToolRegistry([FAST_TOOL, SLOW_TOOL], {
			fast_tool: async () => ({ done: 'fast' }),
			slow_tool: async () => {
				await slowGate;
				return { done: 'slow' };
			}
		});
		const { session } = await startActiveSession({ ...DEFAULTS, toolRegistry });

		session.emitToOrchestrator({
			type: 'tool.call',
			callId: 'call-fast',
			name: 'fast_tool',
			arguments: '{}'
		});
		session.emitToOrchestrator({
			type: 'tool.call',
			callId: 'call-slow',
			name: 'slow_tool',
			arguments: '{}'
		});

		await vi.waitFor(() =>
			expect(session.sent).toContainEqual({
				type: 'tool.result',
				callId: 'call-fast',
				output: JSON.stringify({ done: 'fast' })
			})
		);
		expect(session.sent).not.toContainEqual({ type: 'response.trigger' });

		releaseSlow();
		await vi.waitFor(() =>
			expect(session.sent).toContainEqual({
				type: 'tool.result',
				callId: 'call-slow',
				output: JSON.stringify({ done: 'slow' })
			})
		);
		expect(session.sent).toContainEqual({ type: 'response.trigger' });
	});

	it('forwards a client tool call to the browser and relays its result to the provider', async () => {
		const toolRegistry = createToolRegistry([CLIENT_TOOL], {});
		const { session, socket } = await startActiveSession({ ...DEFAULTS, toolRegistry });

		session.emitToOrchestrator({
			type: 'tool.call',
			callId: 'call-1',
			name: 'client_tool',
			arguments: '{}'
		});

		await vi.waitFor(() =>
			expect(socket.sent).toContainEqual({
				type: 'tool.call',
				callId: 'call-1',
				name: 'client_tool',
				arguments: '{}'
			})
		);

		socket.receive({ type: 'tool.result', callId: 'call-1', output: '{"ok":true}' });

		await vi.waitFor(() =>
			expect(session.sent).toContainEqual({
				type: 'tool.result',
				callId: 'call-1',
				output: '{"ok":true}'
			})
		);
		expect(session.sent).toContainEqual({ type: 'response.trigger' });
	});

	it('resolves an unanswered client tool call with an error after the timeout, without hanging', async () => {
		const toolRegistry = createToolRegistry([CLIENT_TOOL], {});
		const { session, socket } = await startActiveSession({ ...DEFAULTS, toolRegistry });

		vi.useFakeTimers();
		session.emitToOrchestrator({
			type: 'tool.call',
			callId: 'call-1',
			name: 'client_tool',
			arguments: '{}'
		});
		expect(socket.sent).toContainEqual({
			type: 'tool.call',
			callId: 'call-1',
			name: 'client_tool',
			arguments: '{}'
		});

		await vi.advanceTimersByTimeAsync(10_000);

		expect(session.sent).toContainEqual({
			type: 'tool.result',
			callId: 'call-1',
			output: JSON.stringify({ error: 'Client tool call timed out.' })
		});
		expect(session.sent).toContainEqual({ type: 'response.trigger' });
	});

	it('resolves immediately with an error for an unknown tool name', async () => {
		const { session } = await startActiveSession(DEFAULTS);

		session.emitToOrchestrator({
			type: 'tool.call',
			callId: 'call-1',
			name: 'nonexistent_tool',
			arguments: '{}'
		});

		await vi.waitFor(() =>
			expect(session.sent).toContainEqual({
				type: 'tool.result',
				callId: 'call-1',
				output: JSON.stringify({ error: 'Unknown tool "nonexistent_tool".' })
			})
		);
		expect(session.sent).toContainEqual({ type: 'response.trigger' });
	});

	it('clears a pending client tool timer on close, without throwing when it would have fired', async () => {
		const toolRegistry = createToolRegistry([CLIENT_TOOL], {});
		const { provider, session, socket } = await startActiveSession({ ...DEFAULTS, toolRegistry });

		vi.useFakeTimers();
		session.emitToOrchestrator({
			type: 'tool.call',
			callId: 'call-1',
			name: 'client_tool',
			arguments: '{}'
		});

		socket.emit('close'); // synchronous: RealtimeSession.close() tears down the provider session inline
		expect(provider.sessions[0].closed).toBe(true);

		// If the timer weren't cleared, this would try to send through a
		// now-null providerSession — sendToProvider's optional-chaining guard
		// (rather than a throw) is what's being verified here: awaiting this
		// without the test failing means the timer callback didn't throw.
		await vi.advanceTimersByTimeAsync(10_000);
	});
});