diff --git a/app-config.ts b/app-config.ts index cc0a5db09..79d6b54c8 100644 --- a/app-config.ts +++ b/app-config.ts @@ -220,9 +220,9 @@ export const APP_CONFIG_DEFAULTS: AppConfig = { remoteVideoHeight: 480, remoteVideoFps: 25, - logo: '/lk-logo.png', + logo: 'lk-logo.png', accent: '#002cf2', - logoDark: '/lk-logo-dark.png', + logoDark: 'lk-logo-dark.png', accentDark: '#1fd5f9', startButtonText: 'Start call', diff --git a/app/api/session/agent-worker-readiness.ts b/app/api/session/agent-worker-readiness.ts new file mode 100644 index 000000000..e8456cc87 --- /dev/null +++ b/app/api/session/agent-worker-readiness.ts @@ -0,0 +1,129 @@ +import { readFile } from 'node:fs/promises'; +import path from 'node:path'; + +const DEFAULT_READY_TIMEOUT_MS = 30_000; +const DEFAULT_READY_POLL_MS = 200; + +type AgentWorkerReadyMarker = { + version: number; + agentName: string; + workerId: string; + registeredAt: string; +}; + +export type WaitForAgentWorkerReadyOptions = { + readyFile?: string; + timeoutMs?: number; + maxWaitMs?: number; + pollMs?: number; + readFile?: (filePath: string) => Promise; + sleep?: (ms: number) => Promise; +}; + +export type AgentWorkerReadiness = + | { + state: 'ready'; + agentName: string; + workerId: string; + registeredAt: string; + waitedMs: number; + } + | { + state: 'skipped'; + agentName: string; + reason: 'not_sandbox'; + waitedMs: 0; + }; + +export function resolveAgentWorkerReadyFile(env: NodeJS.ProcessEnv = process.env): string { + const configured = (env.LIVEAVATAR_AGENT_WORKER_READY_FILE || '').trim(); + if (configured) { + return configured; + } + if ((env.LIVEAVATAR_RUNTIME_MODE || '').trim().toLowerCase() !== 'sandbox') { + return ''; + } + const workspaceDataDir = (env.LIVEAVATAR_SANDBOX_WORKSPACE_DATA_DIR || '/workspace/data').trim(); + return workspaceDataDir + ? path.join(workspaceDataDir, 'logs', 'sandbox', 'agent-worker-ready.json') + : ''; +} + +export function resolveAgentWorkerReadyTimeoutMs(env: NodeJS.ProcessEnv = process.env): number { + return readPositiveInt(env.LIVEAVATAR_AGENT_WORKER_READY_TIMEOUT_MS, DEFAULT_READY_TIMEOUT_MS); +} + +export async function waitForAgentWorkerReady( + agentName: string, + options: WaitForAgentWorkerReadyOptions = {} +): Promise { + const readyFile = options.readyFile ?? resolveAgentWorkerReadyFile(); + if (!readyFile) { + return { state: 'skipped', agentName, reason: 'not_sandbox', waitedMs: 0 }; + } + + const configuredTimeoutMs = options.timeoutMs ?? resolveAgentWorkerReadyTimeoutMs(); + const timeoutMs = + options.maxWaitMs === undefined + ? configuredTimeoutMs + : Math.max(0, Math.min(configuredTimeoutMs, options.maxWaitMs)); + if (timeoutMs <= 0) { + throw new Error(`agent worker did not register before prewarm timeout: ${agentName}`); + } + const pollMs = + options.pollMs ?? + readPositiveInt(process.env.LIVEAVATAR_AGENT_WORKER_READY_POLL_MS, DEFAULT_READY_POLL_MS); + const readMarkerFile = options.readFile || ((filePath: string) => readFile(filePath, 'utf8')); + const sleepFn = options.sleep || sleep; + const startedAt = Date.now(); + const deadline = startedAt + timeoutMs; + + do { + const marker = await readReadyMarker(readyFile, readMarkerFile); + if (marker?.agentName === agentName && marker.workerId) { + return { + state: 'ready', + agentName, + workerId: marker.workerId, + registeredAt: marker.registeredAt, + waitedMs: Date.now() - startedAt, + }; + } + + const waitMs = Math.min(pollMs, deadline - Date.now()); + if (waitMs > 0) { + await sleepFn(waitMs); + } + } while (Date.now() < deadline); + + throw new Error(`agent worker did not register before prewarm timeout: ${agentName}`); +} + +async function readReadyMarker( + readyFile: string, + readMarkerFile: (filePath: string) => Promise +): Promise { + try { + const payload = JSON.parse(await readMarkerFile(readyFile)) as Partial; + if ( + payload.version !== 1 || + typeof payload.agentName !== 'string' || + typeof payload.workerId !== 'string' || + typeof payload.registeredAt !== 'string' + ) { + return null; + } + return payload as AgentWorkerReadyMarker; + } catch { + return null; + } +} + +function readPositiveInt(value: string | undefined, fallback: number): number { + const parsed = Number.parseInt(String(value || ''), 10); + return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback; +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} diff --git a/app/api/session/dispatch/route.ts b/app/api/session/dispatch/route.ts index 431c59c71..982817d87 100644 --- a/app/api/session/dispatch/route.ts +++ b/app/api/session/dispatch/route.ts @@ -1,43 +1,17 @@ import { NextResponse } from 'next/server'; -import { AgentDispatchClient, RoomServiceClient } from 'livekit-server-sdk'; -import { type ParticipantInfo } from '@livekit/protocol'; +import { + RoomSessionCancelledError, + dispatchRoomSession, +} from '@/app/api/session/session-dispatch-service'; import { deriveLiveKitRoomName, deriveSessionIdFromLiveKitRoomName, isValidConnectionRoomId, } from '@/lib/connection-room-id'; -import { - type AgentParticipantMatchOptions, - type ReusableAgentParticipantOptions, - findAgentParticipantInList, - findReusableAgentParticipant as findReusableAgentParticipantInList, -} from '@/lib/session-dispatch-readiness'; -import { resolveLiveKitHttpUrl } from '@/lib/session-stop'; -import { - type RoomSessionToken, - beginRoomSessionDispatch, - finishRoomSessionDispatch, - isRoomSessionCancelled, - markRoomSessionRunning, - registerRoomSessionDispatchId, -} from '../session-registry'; - -const AGENT_DISPATCH_TIMEOUT_MS = readPositiveIntEnv('AGENT_DISPATCH_TIMEOUT_MS', 8_000); -const AGENT_DISPATCH_RETRY_MS = readPositiveIntEnv('AGENT_DISPATCH_RETRY_MS', 500); -const AGENT_DISPATCH_POLL_MS = readPositiveIntEnv('AGENT_DISPATCH_POLL_MS', 200); export const runtime = 'nodejs'; export const revalidate = 0; -class RoomSessionCancelledError extends Error { - constructor(session: RoomSessionToken) { - super( - `room session was cancelled for sessionId=${session.sessionId} roomName=${session.roomName}` - ); - this.name = 'RoomSessionCancelledError'; - } -} - export async function POST(req: Request) { let body: { roomName?: string; @@ -76,56 +50,25 @@ export async function POST(req: Request) { if (!sessionId) { return NextResponse.json({ status: 'error', error: 'sessionId is required' }, { status: 400 }); } - const requireRoomVideoInputReady = - body.requireRoomVideoInputReady === true || body.require_room_video_input_ready === true; - - const liveKitHttpUrl = resolveLiveKitHttpUrl(process.env.LIVEKIT_URL); - const apiKey = process.env.LIVEKIT_API_KEY; - const apiSecret = process.env.LIVEKIT_API_SECRET; - if (!liveKitHttpUrl || !apiKey || !apiSecret) { - return NextResponse.json( - { status: 'error', error: 'LiveKit API configuration is required' }, - { status: 500 } - ); - } try { - const dispatchClient = new AgentDispatchClient(liveKitHttpUrl, apiKey, apiSecret); - const roomClient = new RoomServiceClient(liveKitHttpUrl, apiKey, apiSecret); - const session = beginRoomSessionDispatch(roomName, sessionId, agentName); - try { - const dispatch = await createAgentDispatchWithRetry( - dispatchClient, - roomClient, - roomName, - agentName, - session, - { requireRoomVideoInputReady } - ); - console.info('agent session dispatch completed', { - roomName, - sessionId, - agentName, - dispatch, - }); - return NextResponse.json({ status: 'dispatched', roomName, agentName, sessionId, dispatch }); - } finally { - finishRoomSessionDispatch(session); - } + const dispatch = await dispatchRoomSession({ + roomName, + sessionId, + agentName, + readiness: { + requireRoomVideoInputReady: + body.requireRoomVideoInputReady === true || body.require_room_video_input_ready === true, + }, + }); + return NextResponse.json({ status: 'dispatched', roomName, agentName, sessionId, dispatch }); } catch (error) { if (error instanceof RoomSessionCancelledError) { return NextResponse.json( - { - status: 'cancelled', - roomName, - agentName, - sessionId, - error: error.message, - }, + { status: 'cancelled', roomName, agentName, sessionId, error: error.message }, { status: 409 } ); } - return NextResponse.json( { status: 'error', @@ -138,207 +81,3 @@ export async function POST(req: Request) { ); } } - -function readPositiveIntEnv(name: string, fallback: number) { - const raw = process.env[name]; - if (!raw) { - return fallback; - } - - const value = Number.parseInt(raw, 10); - return Number.isFinite(value) && value > 0 ? value : fallback; -} - -async function createAgentDispatchWithRetry( - dispatchClient: AgentDispatchClient, - roomClient: RoomServiceClient, - roomName: string, - agentName: string, - session: RoomSessionToken, - reusableAgentOptions: ReusableAgentParticipantOptions = {} -) { - const startedAt = Date.now(); - let lastError: unknown; - let attempts = 0; - - do { - try { - throwIfSessionCancelled(session); - - const alreadyJoined = await findReusableAgentParticipant( - roomClient, - roomName, - agentName, - reusableAgentOptions - ); - throwIfSessionCancelled(session); - if (alreadyJoined) { - markRoomSessionRunning(session); - return { - attempts, - alreadyJoined: true, - agentParticipant: summarizeAgentParticipant(alreadyJoined), - }; - } - - const dispatch = await dispatchClient.createDispatch(roomName, agentName); - attempts += 1; - registerRoomSessionDispatchId(session, dispatch.id); - - if (isRoomSessionCancelled(session)) { - await deleteDispatchQuietly(dispatchClient, dispatch.id, roomName); - await deleteLiveKitRoomQuietly(roomClient, roomName); - throw new RoomSessionCancelledError(session); - } - - // A successful LiveKit dispatch often needs multiple seconds before the - // agent worker joins the room. Wait for the full remaining session-start - // budget here; the retry loop is for API/listParticipants failures, not - // for repeatedly recreating a healthy dispatch every retry interval. - const agentParticipant = await waitForAgentParticipant( - roomClient, - roomName, - agentName, - remainingDispatchTime(startedAt), - session - ); - if (agentParticipant) { - if (isRoomSessionCancelled(session)) { - await deleteLiveKitRoomQuietly(roomClient, roomName); - throw new RoomSessionCancelledError(session); - } - markRoomSessionRunning(session); - return { - attempts, - dispatchId: dispatch.id, - agentParticipant: summarizeAgentParticipant(agentParticipant), - }; - } - - lastError = new Error('agent participant did not join before retry'); - await deleteDispatchQuietly(dispatchClient, dispatch.id, roomName); - } catch (error) { - if (error instanceof RoomSessionCancelledError) { - throw error; - } - lastError = error; - await sleep( - Math.min(calculateDispatchRetryDelay(attempts), remainingDispatchTime(startedAt)) - ); - } - - if (Date.now() - startedAt >= AGENT_DISPATCH_TIMEOUT_MS) { - break; - } - } while (true); - - throw new Error( - `agent dispatch failed for ${agentName} after ${attempts} attempt(s): ${ - lastError instanceof Error ? lastError.message : String(lastError) - }` - ); -} - -function remainingDispatchTime(startedAt: number) { - return Math.max(0, AGENT_DISPATCH_TIMEOUT_MS - (Date.now() - startedAt)); -} - -function calculateDispatchRetryDelay(attempts: number) { - const multiplier = 2 ** Math.max(0, attempts - 1); - return Math.min(AGENT_DISPATCH_RETRY_MS * multiplier, AGENT_DISPATCH_TIMEOUT_MS); -} - -async function waitForAgentParticipant( - roomClient: RoomServiceClient, - roomName: string, - agentName: string, - maxWaitMs: number, - session: RoomSessionToken -) { - const deadline = Date.now() + maxWaitMs; - - do { - throwIfSessionCancelled(session); - const participant = await findAgentParticipant(roomClient, roomName, agentName, { - allowAnonymousLiveKitAgentFallback: true, - }); - if (participant) { - throwIfSessionCancelled(session); - return participant; - } - - const waitMs = Math.min(AGENT_DISPATCH_POLL_MS, deadline - Date.now()); - if (waitMs > 0) { - await sleep(waitMs); - } - } while (Date.now() < deadline); - - throwIfSessionCancelled(session); - return findAgentParticipant(roomClient, roomName, agentName, { - allowAnonymousLiveKitAgentFallback: true, - }); -} - -async function findAgentParticipant( - roomClient: RoomServiceClient, - roomName: string, - agentName: string, - options: AgentParticipantMatchOptions = {} -) { - const participants = await roomClient.listParticipants(roomName); - return findAgentParticipantInList(participants, agentName, options); -} - -async function findReusableAgentParticipant( - roomClient: RoomServiceClient, - roomName: string, - agentName: string, - options: ReusableAgentParticipantOptions = {} -) { - const participants = await roomClient.listParticipants(roomName); - return findReusableAgentParticipantInList(participants, agentName, options); -} - -function summarizeAgentParticipant(participant: ParticipantInfo | null) { - if (!participant) { - return null; - } - - return { - identity: participant.identity, - }; -} - -async function deleteDispatchQuietly( - dispatchClient: AgentDispatchClient, - dispatchId: string, - roomName: string -) { - if (!dispatchId) { - return; - } - - try { - await dispatchClient.deleteDispatch(dispatchId, roomName); - } catch { - // The dispatch may already have been consumed or cleaned up by LiveKit. - } -} - -async function deleteLiveKitRoomQuietly(roomClient: RoomServiceClient, roomName: string) { - try { - await roomClient.deleteRoom(roomName); - } catch { - // The room may already be gone because /stop won the race. - } -} - -function throwIfSessionCancelled(session: RoomSessionToken): void { - if (isRoomSessionCancelled(session)) { - throw new RoomSessionCancelledError(session); - } -} - -function sleep(ms: number) { - return new Promise((resolve) => setTimeout(resolve, ms)); -} diff --git a/app/api/session/prewarm/prewarm-use-guard.ts b/app/api/session/prewarm/prewarm-use-guard.ts new file mode 100644 index 000000000..ffa08e64c --- /dev/null +++ b/app/api/session/prewarm/prewarm-use-guard.ts @@ -0,0 +1,59 @@ +type PrewarmUseState = 'started' | 'in_progress' | 'completed'; +type PrewarmUseStates = Map>; + +const globalForPrewarmUse = globalThis as typeof globalThis & { + __liveavatarPrewarmUseStates?: PrewarmUseStates; +}; + +// Process-local like session-registry: deploy one Next.js process or use sticky routing. +// globalThis also keeps one state if that process loads this module in multiple chunks. +// Completed identities remain consumed: Gateway uses a fresh UUID per sandbox session and +// releases the whole sandbox after any ambiguous/failed prewarm response instead of replaying it. +const prewarmUseStates = + globalForPrewarmUse.__liveavatarPrewarmUseStates ?? + (globalForPrewarmUse.__liveavatarPrewarmUseStates = new Map()); + +export function buildPrewarmUseKey(sessionId: string, roomName: string, agentName: string) { + return `${sessionId}\u0000${roomName}\u0000${agentName}`; +} + +export function beginPrewarmUse(key: string): PrewarmUseState { + const state = prewarmUseStates.get(key); + if (state) { + return state; + } + + prewarmUseStates.set(key, 'in_progress'); + return 'started'; +} + +export function completePrewarmUse(key: string) { + if (prewarmUseStates.get(key) === 'in_progress') { + prewarmUseStates.set(key, 'completed'); + } +} + +export function failPrewarmUse(key: string) { + if (prewarmUseStates.get(key) === 'in_progress') { + prewarmUseStates.delete(key); + } +} + +export function releasePrewarmUseAfterFailure(key: string, error: unknown): void { + const retryReady = + error && + typeof error === 'object' && + 'retryReady' in error && + error.retryReady && + typeof (error.retryReady as PromiseLike).then === 'function' + ? (error.retryReady as PromiseLike) + : undefined; + if (!retryReady) { + failPrewarmUse(key); + return; + } + void retryReady.then( + () => failPrewarmUse(key), + () => failPrewarmUse(key) + ); +} diff --git a/app/api/session/prewarm/route.ts b/app/api/session/prewarm/route.ts new file mode 100644 index 000000000..fabb944cd --- /dev/null +++ b/app/api/session/prewarm/route.ts @@ -0,0 +1,124 @@ +import { NextResponse } from 'next/server'; +import { timingSafeEqual } from 'node:crypto'; +import { + beginPrewarmUse, + buildPrewarmUseKey, + completePrewarmUse, + releasePrewarmUseAfterFailure, +} from '@/app/api/session/prewarm/prewarm-use-guard'; +import { + PrewarmRoomSessionError, + prewarmRoomSession, +} from '@/app/api/session/session-dispatch-service'; +import { deriveLiveKitRoomName, isValidConnectionRoomId } from '@/lib/connection-room-id'; + +export const runtime = 'nodejs'; +export const revalidate = 0; + +function secretsMatch(actual: string, expected: string): boolean { + const actualBuffer = Buffer.from(actual); + const expectedBuffer = Buffer.from(expected); + return ( + actualBuffer.length === expectedBuffer.length && timingSafeEqual(actualBuffer, expectedBuffer) + ); +} + +export async function POST(request: Request) { + const expectedSecret = (process.env.LIVEAVATAR_PREWARM_SECRET || '').trim(); + const actualSecret = (request.headers.get('x-liveavatar-prewarm-secret') || '').trim(); + if (!expectedSecret || !actualSecret || !secretsMatch(actualSecret, expectedSecret)) { + return NextResponse.json({ status: 'error', error: 'unauthorized' }, { status: 401 }); + } + + const sessionId = (process.env.LIVEAVATAR_VOICE_SESSION_ID || '').trim(); + const agentName = (process.env.AGENT_NAME || '').trim(); + const configuredRoomName = (process.env.LIVEAVATAR_LIVEKIT_ROOM_NAME || '').trim(); + const roomName = isValidConnectionRoomId(sessionId) ? deriveLiveKitRoomName(sessionId) : ''; + + if (!sessionId || !roomName || configuredRoomName !== roomName || !agentName) { + return NextResponse.json( + { status: 'error', error: 'server-owned prewarm identity is not configured' }, + { status: 503 } + ); + } + + const prewarmUseKey = buildPrewarmUseKey(sessionId, roomName, agentName); + const prewarmUseState = beginPrewarmUse(prewarmUseKey); + if (prewarmUseState !== 'started') { + return NextResponse.json( + { + status: 'error', + error: + prewarmUseState === 'in_progress' + ? 'prewarm already in progress' + : 'prewarm authorization already consumed', + }, + { status: 409, headers: { 'Cache-Control': 'no-store' } } + ); + } + + try { + const result = await prewarmRoomSession({ roomName, sessionId, agentName }); + completePrewarmUse(prewarmUseKey); + return NextResponse.json( + { + status: 'prewarmed', + roomName, + sessionId, + agentName, + readiness: result.readiness, + timings: result.timings, + }, + { headers: { 'Cache-Control': 'no-store' } } + ); + } catch (error) { + releasePrewarmUseAfterFailure(prewarmUseKey, error); + const phase = error instanceof PrewarmRoomSessionError ? error.phase : 'dispatch_readiness'; + const timings = + error instanceof PrewarmRoomSessionError ? error.timings : { totalPrewarmMs: 0 }; + const cause = error instanceof PrewarmRoomSessionError ? error.cause : error; + console.error('session prewarm failed', { + phase, + roomName, + sessionId, + agentName, + ...safeErrorDiagnostics(cause), + }); + return NextResponse.json( + { + status: 'error', + roomName, + sessionId, + agentName, + error: `prewarm failed during ${phase}`, + phase, + timings, + }, + { status: 502, headers: { 'Cache-Control': 'no-store' } } + ); + } +} + +function safeErrorDiagnostics(error: unknown): { + causeName: string; + causeCode?: string; + causeStack?: string; +} { + const causeName = error instanceof Error ? error.name : typeof error; + const rawCode = + error && typeof error === 'object' && 'code' in error ? String(error.code || '') : ''; + const causeCode = /^[A-Za-z0-9_.:-]{1,64}$/.test(rawCode) ? rawCode : undefined; + const stackLines = + error instanceof Error + ? String(error.stack || '') + .split('\n') + .slice(1, 9) + .map((line) => line.trim()) + .filter((line) => line.startsWith('at ')) + : []; + return { + causeName, + ...(causeCode ? { causeCode } : {}), + ...(stackLines.length > 0 ? { causeStack: stackLines.join('\n') } : {}), + }; +} diff --git a/app/api/session/session-dispatch-service.ts b/app/api/session/session-dispatch-service.ts new file mode 100644 index 000000000..3ef3d6ce8 --- /dev/null +++ b/app/api/session/session-dispatch-service.ts @@ -0,0 +1,739 @@ +import { AgentDispatchClient, RoomServiceClient } from 'livekit-server-sdk'; +import { type ParticipantInfo } from '@livekit/protocol'; +import { + type ReusableAgentParticipantOptions, + findReusableAgentParticipant as findReusableAgentParticipantInList, + summarizeRoomInputReadiness, +} from '@/lib/session-dispatch-readiness'; +import { resolveLiveKitHttpUrl } from '@/lib/session-stop'; +import { + type AgentWorkerReadiness, + type WaitForAgentWorkerReadyOptions, + resolveAgentWorkerReadyTimeoutMs, + waitForAgentWorkerReady, +} from './agent-worker-readiness'; +import { + type RoomSessionToken, + beginRoomSessionDispatch, + finishRoomSessionDispatch, + isRoomSessionCancelled, + markRoomSessionRunning, + registerRoomSessionDispatchId, +} from './session-registry'; + +type DispatchClient = Pick; +type RoomClient = Pick< + RoomServiceClient, + 'createRoom' | 'deleteRoom' | 'listParticipants' | 'listRooms' +>; + +type DispatchDependencies = { + dispatchClient?: DispatchClient; + roomClient?: RoomClient; + dispatchTimeoutMs?: number; + dispatchDeadlineMs?: number; + dispatchRetryMs?: number; + dispatchPollMs?: number; + sleep?: (ms: number) => Promise; + waitForAgentWorkerReady?: ( + agentName: string, + options?: Pick + ) => Promise; +}; + +export type DispatchRoomSessionRequest = { + roomName: string; + sessionId: string; + agentName: string; + readiness?: ReusableAgentParticipantOptions; +}; + +export class RoomSessionCancelledError extends Error { + constructor(session: RoomSessionToken) { + super( + `room session was cancelled for sessionId=${session.sessionId} roomName=${session.roomName}` + ); + this.name = 'RoomSessionCancelledError'; + } +} + +type InFlightDispatch = { + operation: Promise>; + session: RoomSessionToken; + callers: number; + deadline: { value: number }; +}; + +type InFlightDispatches = Map; + +const globalForInFlightDispatches = globalThis as typeof globalThis & { + __liveavatarInFlightDispatches?: InFlightDispatches; +}; + +// Process-local like session-registry; globalThis keeps dispatch de-duplication and +// deadline extensions shared when Next.js loads this service in multiple chunks. +const inFlightDispatches = + globalForInFlightDispatches.__liveavatarInFlightDispatches ?? + (globalForInFlightDispatches.__liveavatarInFlightDispatches = new Map()); +const DEFAULT_AGENT_DISPATCH_TIMEOUT_MS = 8_000; +const DEFAULT_PREWARM_TOTAL_TIMEOUT_MS = 45_000; + +export type PrewarmPhase = 'room' | 'worker_readiness' | 'dispatch_readiness'; + +export type PrewarmTimings = { + totalPrewarmMs: number; + roomEnsureMs: number; + workerReadyWaitMs: number; + dispatchReadinessMs: number; +}; + +export type PrewarmFailureTimings = Partial & + Pick; + +export class PrewarmRoomSessionError extends Error { + readonly phase: PrewarmPhase; + readonly timings: PrewarmFailureTimings; + readonly retryReady?: Promise; + + constructor( + phase: PrewarmPhase, + timings: PrewarmFailureTimings, + cause: unknown, + retryReady?: Promise + ) { + const detail = cause instanceof Error ? cause.message : String(cause); + super(`prewarm failed during ${phase}: ${detail}`, { cause }); + this.name = 'PrewarmRoomSessionError'; + this.phase = phase; + this.timings = { ...timings }; + this.retryReady = retryReady; + } +} + +class PrewarmDeadlineError extends Error { + readonly settled: Promise; + + constructor(message: string, settled: Promise) { + super(message); + this.name = 'PrewarmDeadlineError'; + this.settled = settled; + } +} + +export async function dispatchRoomSession( + request: DispatchRoomSessionRequest, + dependencies: DispatchDependencies = {} +) { + const startedAt = Date.now(); + const callerDeadline = resolveDispatchDeadline(dependencies, startedAt); + if (callerDeadline <= startedAt) { + throw new Error('agent dispatch deadline expired before dispatch'); + } + const key = `${request.sessionId}\u0000${request.roomName}\u0000${request.agentName}`; + let inFlight = inFlightDispatches.get(key); + if (!inFlight) { + const clients = resolveClients(dependencies); + const deadline = { value: callerDeadline }; + // Dispatch creation is shared by identity. Each caller waits for its own + // readiness contract below, while the shared operation keeps the longest + // timeout budget of all concurrent callers. + const session = beginRoomSessionDispatch( + request.roomName, + request.sessionId, + request.agentName + ); + inFlight = { + operation: runRoomSessionDispatch( + { ...request, readiness: {} }, + { ...dependencies, ...clients }, + session, + () => deadline.value + ), + session, + callers: 0, + deadline, + }; + inFlightDispatches.set(key, inFlight); + } else { + inFlight.deadline.value = Math.max(inFlight.deadline.value, callerDeadline); + } + + inFlight.callers += 1; + try { + const dispatch = await inFlight.operation; + return await waitForRequestedRoomSessionReadiness( + request, + dependencies, + dispatch, + inFlight.session, + startedAt + ); + } finally { + inFlight.callers -= 1; + if (inFlight.callers === 0 && inFlightDispatches.get(key) === inFlight) { + finishRoomSessionDispatch(inFlight.session); + inFlightDispatches.delete(key); + } + } +} + +async function waitForRequestedRoomSessionReadiness( + request: DispatchRoomSessionRequest, + dependencies: DispatchDependencies, + dispatch: Record, + session: RoomSessionToken, + startedAt: number +) { + const readiness = request.readiness ?? {}; + if ( + readiness.requireRoomInputParticipantsReady !== true && + readiness.requireRoomVideoInputReady !== true + ) { + return dispatch; + } + + const clients = resolveClients(dependencies); + const deadline = resolveDispatchDeadline(dependencies, startedAt); + const participant = await waitForReusableAgentParticipant( + clients.roomClient, + request.roomName, + request.agentName, + readiness, + () => deadline, + dependencies.dispatchPollMs || readPositiveIntEnv('AGENT_DISPATCH_POLL_MS', 200), + session, + dependencies.sleep || sleep + ); + if (!participant) { + throw new Error('agent and required room inputs did not become ready'); + } + throwIfSessionCancelled(session); + markRoomSessionRunning(session); + return { + ...dispatch, + agentParticipant: summarizeAgentParticipant(participant), + }; +} + +export async function prewarmRoomSession( + request: Omit, + dependencies: DispatchDependencies = {} +) { + const prewarmStartedAt = Date.now(); + const prewarmTimeoutMs = + dependencies.dispatchTimeoutMs || + readPositiveIntEnv('LIVEAVATAR_PREWARM_TOTAL_TIMEOUT_MS', DEFAULT_PREWARM_TOTAL_TIMEOUT_MS); + const prewarmDeadline = prewarmStartedAt + prewarmTimeoutMs; + const prewarmAbortController = new AbortController(); + const completedTimings: Partial = {}; + + const roomPhaseStartedAt = Date.now(); + let roomAndClients: { + room: Awaited>; + clients: ReturnType; + }; + try { + roomAndClients = await runWithinPrewarmDeadline( + prewarmDeadline, + 'room', + async () => { + const clients = resolveClients(dependencies); + const room = await ensureLiveKitRoom( + clients.roomClient, + request.roomName, + prewarmDeadline, + prewarmAbortController.signal + ); + return { clients, room }; + }, + prewarmAbortController + ); + completedTimings.roomEnsureMs = elapsedSince(roomPhaseStartedAt); + } catch (error) { + throw buildPrewarmPhaseError( + 'room', + prewarmStartedAt, + roomPhaseStartedAt, + completedTimings, + error + ); + } + + const workerPhaseStartedAt = Date.now(); + let workerReadiness: AgentWorkerReadiness; + try { + const workerMaxWaitMs = Math.min( + resolveAgentWorkerReadyTimeoutMs(), + remainingDispatchTime(prewarmDeadline) + ); + workerReadiness = await runWithinPrewarmDeadline( + prewarmDeadline, + 'worker_readiness', + () => + (dependencies.waitForAgentWorkerReady || waitForAgentWorkerReady)(request.agentName, { + maxWaitMs: workerMaxWaitMs, + }), + prewarmAbortController + ); + completedTimings.workerReadyWaitMs = elapsedSince(workerPhaseStartedAt); + } catch (error) { + throw buildPrewarmPhaseError( + 'worker_readiness', + prewarmStartedAt, + workerPhaseStartedAt, + completedTimings, + error + ); + } + + const dispatchPhaseStartedAt = Date.now(); + let dispatchAndParticipants: { + dispatch: Awaited>; + participants: ParticipantInfo[]; + }; + try { + dispatchAndParticipants = await runWithinPrewarmDeadline( + prewarmDeadline, + 'dispatch_readiness', + async () => { + const dispatch = await dispatchRoomSession( + { + ...request, + readiness: { requireRoomInputParticipantsReady: true }, + }, + { + ...dependencies, + ...roomAndClients.clients, + dispatchDeadlineMs: prewarmDeadline, + } + ); + throwIfPrewarmUnavailable( + prewarmDeadline, + prewarmAbortController.signal, + 'final readiness check' + ); + const participants = await roomAndClients.clients.roomClient.listParticipants( + request.roomName + ); + throwIfPrewarmUnavailable( + prewarmDeadline, + prewarmAbortController.signal, + 'final readiness check' + ); + return { dispatch, participants }; + }, + prewarmAbortController + ); + completedTimings.dispatchReadinessMs = elapsedSince(dispatchPhaseStartedAt); + } catch (error) { + throw buildPrewarmPhaseError( + 'dispatch_readiness', + prewarmStartedAt, + dispatchPhaseStartedAt, + completedTimings, + error + ); + } + + const timings: PrewarmTimings = { + totalPrewarmMs: elapsedSince(prewarmStartedAt), + roomEnsureMs: completedTimings.roomEnsureMs, + workerReadyWaitMs: completedTimings.workerReadyWaitMs, + dispatchReadinessMs: completedTimings.dispatchReadinessMs, + }; + return { + room: { name: roomAndClients.room.name }, + workerReadiness, + dispatch: dispatchAndParticipants.dispatch, + readiness: summarizeRoomInputReadiness(dispatchAndParticipants.participants), + timings, + }; +} + +async function runRoomSessionDispatch( + request: DispatchRoomSessionRequest, + dependencies: DispatchDependencies, + session: RoomSessionToken, + getDeadline: () => number +) { + const { roomName, sessionId, agentName, readiness = {} } = request; + const clients = resolveClients(dependencies); + const dispatch = await createAgentDispatchWithRetry( + clients.dispatchClient, + clients.roomClient, + roomName, + agentName, + session, + readiness, + { + timeoutMs: dependencies.dispatchTimeoutMs, + getDeadline, + retryMs: dependencies.dispatchRetryMs, + pollMs: dependencies.dispatchPollMs, + sleep: dependencies.sleep, + } + ); + console.info('agent session dispatch completed', { + roomName, + sessionId, + agentName, + dispatch, + }); + return dispatch; +} + +function resolveClients(dependencies: DispatchDependencies): { + dispatchClient: DispatchClient; + roomClient: RoomClient; +} { + if (dependencies.dispatchClient && dependencies.roomClient) { + return { + dispatchClient: dependencies.dispatchClient, + roomClient: dependencies.roomClient, + }; + } + + const liveKitHttpUrl = resolveLiveKitHttpUrl(process.env.LIVEKIT_URL); + const apiKey = process.env.LIVEKIT_API_KEY; + const apiSecret = process.env.LIVEKIT_API_SECRET; + if (!liveKitHttpUrl || !apiKey || !apiSecret) { + throw new Error('LiveKit API configuration is required'); + } + + return { + dispatchClient: + dependencies.dispatchClient || new AgentDispatchClient(liveKitHttpUrl, apiKey, apiSecret), + roomClient: dependencies.roomClient || new RoomServiceClient(liveKitHttpUrl, apiKey, apiSecret), + }; +} + +async function ensureLiveKitRoom( + roomClient: RoomClient, + roomName: string, + deadline: number, + abortSignal: AbortSignal +) { + const existing = await roomClient.listRooms([roomName]); + throwIfPrewarmUnavailable(deadline, abortSignal, 'room ensure'); + if (existing.length > 0) { + return existing[0]; + } + + try { + const created = await roomClient.createRoom({ name: roomName }); + if (abortSignal.aborted || remainingDispatchTime(deadline) <= 0) { + await deleteLiveKitRoomQuietly(roomClient, roomName); + throw new Error('room ensure deadline expired after room creation'); + } + return created; + } catch (error) { + throwIfPrewarmUnavailable(deadline, abortSignal, 'room ensure'); + const raced = await roomClient.listRooms([roomName]); + throwIfPrewarmUnavailable(deadline, abortSignal, 'room ensure'); + if (raced.length > 0) { + return raced[0]; + } + throw error; + } +} + +async function createAgentDispatchWithRetry( + dispatchClient: DispatchClient, + roomClient: RoomClient, + roomName: string, + agentName: string, + session: RoomSessionToken, + reusableAgentOptions: ReusableAgentParticipantOptions, + options: { + timeoutMs?: number; + getDeadline?: () => number; + retryMs?: number; + pollMs?: number; + sleep?: (ms: number) => Promise; + } +) { + const timeoutMs = + options.timeoutMs || + readPositiveIntEnv('AGENT_DISPATCH_TIMEOUT_MS', DEFAULT_AGENT_DISPATCH_TIMEOUT_MS); + const fixedDeadline = Date.now() + timeoutMs; + const getDeadline = options.getDeadline || (() => fixedDeadline); + const retryMs = options.retryMs || readPositiveIntEnv('AGENT_DISPATCH_RETRY_MS', 500); + const pollMs = options.pollMs || readPositiveIntEnv('AGENT_DISPATCH_POLL_MS', 200); + const sleepFn = options.sleep || sleep; + let lastError: unknown; + let attempts = 0; + let dispatchId = ''; + + do { + try { + throwIfSessionCancelled(session); + throwIfDeadlineExpired(getDeadline(), 'agent dispatch'); + const alreadyJoined = await findReusableAgentParticipant( + roomClient, + roomName, + agentName, + reusableAgentOptions + ); + throwIfSessionCancelled(session); + throwIfDeadlineExpired(getDeadline(), 'agent dispatch'); + if (alreadyJoined) { + markRoomSessionRunning(session); + return { + attempts, + alreadyJoined: true, + agentParticipant: summarizeAgentParticipant(alreadyJoined), + }; + } + + if (!dispatchId) { + throwIfDeadlineExpired(getDeadline(), 'agent dispatch creation'); + const dispatch = await dispatchClient.createDispatch(roomName, agentName); + attempts += 1; + dispatchId = dispatch.id; + registerRoomSessionDispatchId(session, dispatchId); + throwIfDeadlineExpired(getDeadline(), 'agent dispatch creation'); + } + + if (isRoomSessionCancelled(session)) { + await deleteDispatchQuietly(dispatchClient, dispatchId, roomName); + await deleteLiveKitRoomQuietly(roomClient, roomName); + throw new RoomSessionCancelledError(session); + } + + const agentParticipant = await waitForReusableAgentParticipant( + roomClient, + roomName, + agentName, + reusableAgentOptions, + getDeadline, + pollMs, + session, + sleepFn + ); + if (agentParticipant) { + throwIfSessionCancelled(session); + markRoomSessionRunning(session); + return { + attempts, + dispatchId, + agentParticipant: summarizeAgentParticipant(agentParticipant), + }; + } + + lastError = new Error('agent and required room inputs did not become ready'); + } catch (error) { + if (error instanceof RoomSessionCancelledError) { + throw error; + } + lastError = error; + const waitMs = Math.min( + calculateDispatchRetryDelay(attempts, retryMs), + remainingDispatchTime(getDeadline()) + ); + if (waitMs > 0) { + await sleepFn(waitMs); + } + } + } while (Date.now() < getDeadline()); + + await deleteDispatchQuietly(dispatchClient, dispatchId, roomName); + throw new Error( + `agent dispatch failed for ${agentName} after ${attempts} attempt(s): ${ + lastError instanceof Error ? lastError.message : String(lastError) + }` + ); +} + +function resolveDispatchDeadline(dependencies: DispatchDependencies, startedAt: number) { + if ( + dependencies.dispatchDeadlineMs !== undefined && + Number.isFinite(dependencies.dispatchDeadlineMs) + ) { + return dependencies.dispatchDeadlineMs; + } + const timeoutMs = + dependencies.dispatchTimeoutMs || + readPositiveIntEnv('AGENT_DISPATCH_TIMEOUT_MS', DEFAULT_AGENT_DISPATCH_TIMEOUT_MS); + return startedAt + timeoutMs; +} + +function remainingDispatchTime(deadline: number) { + return Math.max(0, deadline - Date.now()); +} + +function elapsedSince(startedAt: number) { + return Math.max(0, Date.now() - startedAt); +} + +function buildPrewarmPhaseError( + phase: PrewarmPhase, + prewarmStartedAt: number, + phaseStartedAt: number, + completedTimings: Partial, + cause: unknown +) { + const phaseTimingKey: Record< + PrewarmPhase, + 'roomEnsureMs' | 'workerReadyWaitMs' | 'dispatchReadinessMs' + > = { + room: 'roomEnsureMs', + worker_readiness: 'workerReadyWaitMs', + dispatch_readiness: 'dispatchReadinessMs', + }; + return new PrewarmRoomSessionError( + phase, + { + ...completedTimings, + [phaseTimingKey[phase]]: elapsedSince(phaseStartedAt), + totalPrewarmMs: elapsedSince(prewarmStartedAt), + }, + cause, + cause instanceof PrewarmDeadlineError ? cause.settled : undefined + ); +} + +async function runWithinPrewarmDeadline( + deadline: number, + phase: PrewarmPhase, + operation: () => Promise, + abortController: AbortController +): Promise { + const remainingMs = remainingDispatchTime(deadline); + if (remainingMs <= 0) { + abortController.abort(); + throw new Error(`prewarm deadline expired before ${phase}`); + } + + const operationPromise = Promise.resolve().then(operation); + const settled = operationPromise.then( + () => undefined, + () => undefined + ); + let timeoutId: ReturnType | undefined; + const timeout = new Promise((_, reject) => { + timeoutId = setTimeout(() => { + abortController.abort(); + reject(new PrewarmDeadlineError(`prewarm deadline expired during ${phase}`, settled)); + }, remainingMs); + }); + try { + const result = await Promise.race([operationPromise, timeout]); + if (Date.now() >= deadline) { + abortController.abort(); + throw new PrewarmDeadlineError(`prewarm deadline expired during ${phase}`, settled); + } + return result; + } finally { + if (timeoutId !== undefined) { + clearTimeout(timeoutId); + } + } +} + +function calculateDispatchRetryDelay(attempts: number, retryMs: number) { + const multiplier = 2 ** Math.max(0, attempts - 1); + return retryMs * multiplier; +} + +async function waitForReusableAgentParticipant( + roomClient: RoomClient, + roomName: string, + agentName: string, + readiness: ReusableAgentParticipantOptions, + getDeadline: () => number, + pollMs: number, + session: RoomSessionToken, + sleepFn: (ms: number) => Promise +) { + while (true) { + throwIfSessionCancelled(session); + if (remainingDispatchTime(getDeadline()) <= 0) { + return null; + } + const participant = await findReusableAgentParticipant(roomClient, roomName, agentName, { + allowAnonymousLiveKitAgentFallback: true, + ...readiness, + }); + if (remainingDispatchTime(getDeadline()) <= 0) { + return null; + } + if (participant) { + throwIfSessionCancelled(session); + return participant; + } + const waitMs = Math.min(pollMs, remainingDispatchTime(getDeadline())); + if (waitMs <= 0) { + return null; + } + await sleepFn(waitMs); + } +} + +function throwIfDeadlineExpired(deadline: number, phase: string): void { + if (remainingDispatchTime(deadline) <= 0) { + throw new Error(`${phase} deadline expired`); + } +} + +function throwIfPrewarmUnavailable( + deadline: number, + abortSignal: AbortSignal, + phase: string +): void { + if (abortSignal.aborted) { + throw new Error(`${phase} cancelled after prewarm timeout`); + } + throwIfDeadlineExpired(deadline, phase); +} + +async function findReusableAgentParticipant( + roomClient: RoomClient, + roomName: string, + agentName: string, + options: ReusableAgentParticipantOptions = {} +) { + const participants = await roomClient.listParticipants(roomName); + return findReusableAgentParticipantInList(participants, agentName, options); +} + +function summarizeAgentParticipant(participant: ParticipantInfo | null) { + return participant ? { identity: participant.identity } : null; +} + +async function deleteDispatchQuietly( + dispatchClient: DispatchClient, + dispatchId: string, + roomName: string +) { + if (!dispatchId) { + return; + } + try { + await dispatchClient.deleteDispatch(dispatchId, roomName); + } catch { + // The dispatch may already have been consumed or cleaned up by LiveKit. + } +} + +async function deleteLiveKitRoomQuietly(roomClient: RoomClient, roomName: string) { + try { + await roomClient.deleteRoom(roomName); + } catch { + // The room may already be gone because /stop won the race. + } +} + +function throwIfSessionCancelled(session: RoomSessionToken): void { + if (isRoomSessionCancelled(session)) { + throw new RoomSessionCancelledError(session); + } +} + +function readPositiveIntEnv(name: string, fallback: number) { + const value = Number.parseInt(process.env[name] || '', 10); + return Number.isFinite(value) && value > 0 ? value : fallback; +} + +function sleep(ms: number) { + return new Promise((resolve) => setTimeout(resolve, ms)); +} diff --git a/app/api/session/session-registry.ts b/app/api/session/session-registry.ts index ef378df5a..07414c4d0 100644 --- a/app/api/session/session-registry.ts +++ b/app/api/session/session-registry.ts @@ -28,9 +28,24 @@ type RoomSessionRecord = { dispatchWaiters: Set<() => void>; }; +type RoomSessionRegistryState = { + sessions: Map; + nextGeneration: number; +}; + +const globalForRoomSessionRegistry = globalThis as typeof globalThis & { + __liveavatarRoomSessionRegistry?: RoomSessionRegistryState; +}; + // This coordination is process-local; deploy /api/session/* on one instance or use sticky routing. -const sessions = new Map(); -let nextGeneration = 1; +// globalThis also keeps cancellation and generation state shared across Next.js server chunks. +const roomSessionRegistry = + globalForRoomSessionRegistry.__liveavatarRoomSessionRegistry ?? + (globalForRoomSessionRegistry.__liveavatarRoomSessionRegistry = { + sessions: new Map(), + nextGeneration: 1, + }); +const sessions = roomSessionRegistry.sessions; function normalize(value: string | null | undefined): string { return String(value ?? '').trim(); @@ -114,7 +129,7 @@ export function beginRoomSessionDispatch( roomName: normalizedRoomName, sessionId: normalizedSessionId, agentName: normalizedAgentName, - generation: nextGeneration++, + generation: roomSessionRegistry.nextGeneration++, state: 'starting', cancelled: false, dispatchIds: new Set(), @@ -174,7 +189,7 @@ export function markRoomSessionStopping( roomName: normalizedRoomName, sessionId: normalizedSessionId, agentName: '', - generation: nextGeneration++, + generation: roomSessionRegistry.nextGeneration++, state: 'stopping', cancelled: true, dispatchIds: new Set(), diff --git a/app/layout.tsx b/app/layout.tsx index 7098d56d9..6c1935ae5 100644 --- a/app/layout.tsx +++ b/app/layout.tsx @@ -1,5 +1,4 @@ import type { Metadata } from 'next'; -import localFont from 'next/font/local'; import { headers } from 'next/headers'; import { ApplyThemeScript, ThemeToggle } from '@/components/app/theme-toggle'; import { cn, getAppConfig, getStyles } from '@/lib/utils'; @@ -13,33 +12,6 @@ export const metadata: Metadata = { metadataBase: new URL(metadataBaseUrl), }; -const commitMono = localFont({ - display: 'swap', - variable: '--font-commit-mono', - src: [ - { - path: '../fonts/CommitMono-400-Regular.otf', - weight: '400', - style: 'normal', - }, - { - path: '../fonts/CommitMono-700-Regular.otf', - weight: '700', - style: 'normal', - }, - { - path: '../fonts/CommitMono-400-Italic.otf', - weight: '400', - style: 'italic', - }, - { - path: '../fonts/CommitMono-700-Italic.otf', - weight: '700', - style: 'italic', - }, - ], -}); - interface RootLayoutProps { children: React.ReactNode; } @@ -51,11 +23,7 @@ export default async function RootLayout({ children }: RootLayoutProps) { const styles = getStyles(appConfig); return ( - + {styles && } {pageTitle} diff --git a/hooks/useBrowserSourceClient.ts b/hooks/useBrowserSourceClient.ts index 778715168..bdd042dbc 100644 --- a/hooks/useBrowserSourceClient.ts +++ b/hooks/useBrowserSourceClient.ts @@ -11,11 +11,11 @@ import { createLocalVideoTrack, } from 'livekit-client'; import type { AppConfig } from '@/app-config'; -import { startMediaTrackVadObserver } from '@/lib/frontend-vad-observer'; +import { resolveVadAssetBasePaths, startMediaTrackVadObserver } from '@/lib/frontend-vad-observer'; import { FRONTEND_EVENTS, OBSERVABILITY_ATTRS, - publishFrontendObservabilityEvent, + recordFrontendObservabilityEvent, } from '@/lib/observability'; const BROWSER_AUDIO_TRACK_NAME = 'browser_audio_track'; @@ -102,7 +102,7 @@ export function useBrowserSourceClient( attributes?: Record, options?: { wallTimeUnixMs?: number } ) => { - void publishFrontendObservabilityEvent({ + void recordFrontendObservabilityEvent({ enabled: !!appConfig.observabilityEnabled, room, name, @@ -128,18 +128,22 @@ export function useBrowserSourceClient( [OBSERVABILITY_ATTRS.TRACK_SID]: null, [OBSERVABILITY_ATTRS.TRACK_STREAM_NAME]: browserMediaStreamName, }; + recordFrontendObservability(FRONTEND_EVENTS.BROWSER_AUDIO_CAPTURE_STARTED); const audioTrack = await createLocalAudioTrack( buildAudioCaptureOptions(audioDeviceIdRef.current) ); + recordFrontendObservability(FRONTEND_EVENTS.BROWSER_AUDIO_CAPTURE_FINISHED); const captureTrack = audioTrack.mediaStreamTrack; audioTrack.mediaStreamTrack.enabled = runtime.audioEnabled; try { + recordFrontendObservability(FRONTEND_EVENTS.BROWSER_AUDIO_PUBLISH_STARTED); const publication = await room.localParticipant.publishTrack(audioTrack, { name: BROWSER_AUDIO_TRACK_NAME, source: Track.Source.Microphone, stream: browserMediaStreamName, }); + recordFrontendObservability(FRONTEND_EVENTS.BROWSER_AUDIO_PUBLISH_FINISHED); runtime.audioTrack = audioTrack; runtime.audioPublication = publication; runtime.audioObserverStop = null; @@ -150,8 +154,12 @@ export function useBrowserSourceClient( [OBSERVABILITY_ATTRS.TRACK_STREAM_NAME]: browserMediaStreamName, }); if (appConfig.observabilityEnabled) { + const vadAssetBasePaths = appConfig.sandboxId + ? resolveVadAssetBasePaths(window.location.pathname) + : undefined; void startMediaTrackVadObserver({ mediaStreamTrack: captureTrack, + ...vadAssetBasePaths, onSpeechStart: (event) => { recordFrontendObservability( FRONTEND_EVENTS.BROWSER_AUDIO_VAD_SPEECH_STARTED, @@ -202,6 +210,7 @@ export function useBrowserSourceClient( } }, [ appConfig.observabilityEnabled, + appConfig.sandboxId, audioConfigured, browserMediaStreamName, recordFrontendObservability, @@ -214,6 +223,7 @@ export function useBrowserSourceClient( return; } + recordFrontendObservability(FRONTEND_EVENTS.BROWSER_VIDEO_CAPTURE_STARTED); const videoTrack = await createLocalVideoTrack({ facingMode: 'user', frameRate: { ideal: browserVideoFrameRate, max: browserVideoFrameRate }, @@ -223,9 +233,11 @@ export function useBrowserSourceClient( frameRate: browserVideoFrameRate, }, }); + recordFrontendObservability(FRONTEND_EVENTS.BROWSER_VIDEO_CAPTURE_FINISHED); videoTrack.mediaStreamTrack.enabled = runtime.videoEnabled; try { + recordFrontendObservability(FRONTEND_EVENTS.BROWSER_VIDEO_PUBLISH_STARTED); const publication = await room.localParticipant.publishTrack(videoTrack, { name: BROWSER_VIDEO_TRACK_NAME, source: Track.Source.Camera, @@ -237,9 +249,15 @@ export function useBrowserSourceClient( maxFramerate: browserVideoFrameRate, }, }); + recordFrontendObservability(FRONTEND_EVENTS.BROWSER_VIDEO_PUBLISH_FINISHED); runtime.videoTrack = videoTrack; runtime.videoPublication = publication; setVideoTrackState(videoTrack); + recordFrontendObservability(FRONTEND_EVENTS.BROWSER_VIDEO_TRACK_PUBLISHED, { + [OBSERVABILITY_ATTRS.TRACK_NAME]: BROWSER_VIDEO_TRACK_NAME, + [OBSERVABILITY_ATTRS.TRACK_SID]: publication.trackSid || null, + [OBSERVABILITY_ATTRS.TRACK_STREAM_NAME]: browserMediaStreamName, + }); if (browserVideoStatsEnabled) { startBrowserVideoStatsLogging(runtime, videoTrack, publication, room); } @@ -254,6 +272,7 @@ export function useBrowserSourceClient( browserVideoMaxBitrate, browserVideoStatsEnabled, browserVideoWidth, + recordFrontendObservability, room, videoConfigured, ]); @@ -325,27 +344,28 @@ export function useBrowserSourceClient( audioObserverStop: null, }; - try { - if (audioEnabledRef.current) { - await ensureAudioPublished(); + // Sandbox sessions target Chromium and keep independent capture concurrent + // so camera permission does not delay the microphone or agent dispatch. + const audioStart = audioEnabledRef.current ? ensureAudioPublished() : Promise.resolve(); + const videoStart = videoEnabledRef.current ? ensureVideoPublished() : Promise.resolve(); + const [audioResult, videoResult] = await Promise.allSettled([audioStart, videoStart]); + + if (audioResult.status === 'rejected') { + if (videoResult.status === 'rejected') { + onVideoError?.(videoResult.reason as Error); } - } catch (error) { await stop(); - throw error; + throw audioResult.reason; } - if (videoEnabledRef.current) { - try { - await ensureVideoPublished(); - } catch (error) { - videoEnabledRef.current = false; - setVideoEnabledState(false); - const runtime = runtimeRef.current; - if (runtime) { - runtime.videoEnabled = false; - } - onVideoError?.(error as Error); + if (videoResult.status === 'rejected') { + videoEnabledRef.current = false; + setVideoEnabledState(false); + const runtime = runtimeRef.current; + if (runtime) { + runtime.videoEnabled = false; } + onVideoError?.(videoResult.reason as Error); } }, [enabled, ensureAudioPublished, ensureVideoPublished, onVideoError, stop]); diff --git a/hooks/useRoom.ts b/hooks/useRoom.ts index 662ada2a0..4073401a8 100644 --- a/hooks/useRoom.ts +++ b/hooks/useRoom.ts @@ -7,7 +7,13 @@ import { getVoiceSessionId, resetVoiceSessionId } from '@/lib/browser-room-sessi import { readConnectionDetailsResponse } from '@/lib/connection-details-response'; import { isValidConnectionRoomId } from '@/lib/connection-room-id'; import { usesServerRoomInputDevice } from '@/lib/input-device-config'; -import { FRONTEND_EVENTS, publishFrontendObservabilityEvent } from '@/lib/observability'; +import { + FRONTEND_EVENTS, + beginFrontendObservabilitySession, + endFrontendObservabilitySession, + flushFrontendObservabilityEvents, + recordFrontendObservabilityEvent, +} from '@/lib/observability'; import { waitForRoomDisconnected } from '@/lib/room-disconnect'; import { AgentSessionDispatchCancelledError, @@ -15,6 +21,7 @@ import { } from '@/lib/session-dispatch-client'; import { beginAgentSessionStart, + cancelAgentSessionStart, registerAgentSessionDispatch, requestAgentSessionStop, waitForAgentSessionStop, @@ -55,7 +62,7 @@ export function useRoom(appConfig: AppConfig) { }, [appConfig.voiceSessionId]); const recordFrontendObservability = useCallback( (name: string, attributes?: Record) => { - void publishFrontendObservabilityEvent({ + void recordFrontendObservabilityEvent({ enabled: !!appConfig.observabilityEnabled, room, name, @@ -94,6 +101,7 @@ export function useRoom(appConfig: AppConfig) { void requestAgentSessionStop(sessionIdRef.current, { waitForRemote: false, }); + endFrontendObservabilitySession(room); room.disconnect(); }; }, [room]); @@ -102,11 +110,12 @@ export function useRoom(appConfig: AppConfig) { () => TokenSource.custom(async () => { const url = new URL( - process.env.NEXT_PUBLIC_CONN_DETAILS_ENDPOINT ?? '/api/connection-details', - window.location.origin + process.env.NEXT_PUBLIC_CONN_DETAILS_ENDPOINT ?? 'api/connection-details', + window.location.href ); try { + recordFrontendObservability(FRONTEND_EVENTS.CONNECTION_DETAILS_STARTED); const sessionId = sessionIdRef.current ?? resolveVoiceSessionId(); sessionIdRef.current = sessionId; @@ -120,7 +129,9 @@ export function useRoom(appConfig: AppConfig) { sessionId, }), }); - return await readConnectionDetailsResponse(res, { sessionId }); + const connectionDetails = await readConnectionDetailsResponse(res, { sessionId }); + recordFrontendObservability(FRONTEND_EVENTS.CONNECTION_DETAILS_FINISHED); + return connectionDetails; } catch (error) { console.error('Error fetching connection details:', error); if (error instanceof Error) { @@ -129,7 +140,7 @@ export function useRoom(appConfig: AppConfig) { throw new Error('Error fetching connection details!'); } }), - [appConfig, resolveVoiceSessionId] + [appConfig, recordFrontendObservability, resolveVoiceSessionId] ); const startSession = useCallback(async () => { @@ -149,6 +160,19 @@ export function useRoom(appConfig: AppConfig) { const recoverFromStartError = async (error: unknown) => { const startError = error instanceof Error ? error : new Error(String(error)); + if (connectedRoomName) { + try { + await flushFrontendObservabilityEvents({ + enabled: !!appConfig.observabilityEnabled, + room, + }); + } catch (observabilityError) { + console.warn( + '[frontend-observability] failed to flush startup failure events', + observabilityError + ); + } + } try { await browserSourceClient.stop(); } catch (stopError) { @@ -166,6 +190,7 @@ export function useRoom(appConfig: AppConfig) { } } resetVoiceSessionId(); + endFrontendObservabilitySession(room); sessionIdRef.current = null; setIsSessionActive(false); toastAlert({ @@ -188,8 +213,10 @@ export function useRoom(appConfig: AppConfig) { }; setIsSessionActive(true); + beginFrontendObservabilitySession(room); const dispatchAgentSession = async () => { + recordFrontendObservability(FRONTEND_EVENTS.DISPATCH_STARTED); dispatchSessionId = sessionId; const signal = beginAgentSessionStart(room.name, sessionId); const dispatchPromise = requestAgentSessionDispatch(appConfig.agentName, sessionId, { @@ -198,6 +225,11 @@ export function useRoom(appConfig: AppConfig) { }); registerAgentSessionDispatch(room.name, sessionId, dispatchPromise); await dispatchPromise; + recordFrontendObservability(FRONTEND_EVENTS.DISPATCH_FINISHED); + await flushFrontendObservabilityEvents({ + enabled: !!appConfig.observabilityEnabled, + room, + }); }; const startDefaultMicrophone = async () => { @@ -220,28 +252,58 @@ export function useRoom(appConfig: AppConfig) { await startDefaultMicrophone(); }; + const startLocalInputOrCancelDispatch = async () => { + try { + await startLocalInput(); + } catch (error) { + cancelAgentSessionStart(sessionId); + throw error; + } + }; + try { await waitForAgentSessionStop(); await waitForRoomDisconnected(room); if (browserSourceClient.enabled || appConfig.usesServerRoomInput) { const connectionDetails = await tokenSource.fetch({ agentName: appConfig.agentName }); + recordFrontendObservability(FRONTEND_EVENTS.ROOM_CONNECT_STARTED); await room.connect(connectionDetails.serverUrl, connectionDetails.participantToken); + recordFrontendObservability(FRONTEND_EVENTS.ROOM_CONNECT_FINISHED); recordFrontendObservability(FRONTEND_EVENTS.ROOM_CONNECTED); connectedRoomName = room.name; - await startLocalInput(); + const [localInputResult, dispatchResult] = await Promise.allSettled([ + startLocalInputOrCancelDispatch(), + dispatchAgentSession(), + ]); + if (localInputResult.status === 'rejected') { + if (dispatchResult.status === 'rejected') { + console.warn( + 'Agent dispatch also failed while local input was starting', + dispatchResult.reason + ); + } + throw localInputResult.reason; + } + if (dispatchResult.status === 'rejected') { + throw dispatchResult.reason; + } } else { await Promise.all([ startDefaultMicrophone(), tokenSource.fetch({ agentName: appConfig.agentName }).then(async (connectionDetails) => { + recordFrontendObservability(FRONTEND_EVENTS.ROOM_CONNECT_STARTED); await room.connect(connectionDetails.serverUrl, connectionDetails.participantToken); + recordFrontendObservability(FRONTEND_EVENTS.ROOM_CONNECT_FINISHED); recordFrontendObservability(FRONTEND_EVENTS.ROOM_CONNECTED); connectedRoomName = room.name; }), ]); } - await dispatchAgentSession(); + if (!(browserSourceClient.enabled || appConfig.usesServerRoomInput)) { + await dispatchAgentSession(); + } } catch (error) { await handleStartError(error); } @@ -262,6 +324,7 @@ export function useRoom(appConfig: AppConfig) { } finally { room.disconnect(); resetVoiceSessionId(); + endFrontendObservabilitySession(room); sessionIdRef.current = null; setIsSessionActive(false); } diff --git a/lib/frontend-vad-observer.ts b/lib/frontend-vad-observer.ts index 657c115c6..72d114da0 100644 --- a/lib/frontend-vad-observer.ts +++ b/lib/frontend-vad-observer.ts @@ -45,6 +45,23 @@ const VAD_SPEECH_PROBABILITY_THRESHOLD = 0.5; const DEFAULT_VAD_ASSET_BASE_PATH = '/vad-web/'; const DEFAULT_ONNX_WASM_BASE_PATH = '/onnxruntime-web/'; +export function resolveVadAssetBasePaths(sessionPathname = ''): { + baseAssetPath: string; + onnxWASMBasePath: string; +} { + const sessionPath = `/${sessionPathname}`.replace(/\/{2,}/g, '/').replace(/\/+$/, ''); + if (!sessionPath) { + return { + baseAssetPath: DEFAULT_VAD_ASSET_BASE_PATH, + onnxWASMBasePath: DEFAULT_ONNX_WASM_BASE_PATH, + }; + } + return { + baseAssetPath: `${sessionPath}/vad-web/`, + onnxWASMBasePath: `${sessionPath}/onnxruntime-web/`, + }; +} + export async function startMediaTrackVadObserver({ mediaStreamTrack, onSpeechStart, diff --git a/lib/observability.ts b/lib/observability.ts index 3dac10f80..9c98e41a0 100644 --- a/lib/observability.ts +++ b/lib/observability.ts @@ -4,7 +4,17 @@ export const OBSERVABILITY_EVENT_TYPES = { } as const; export const FRONTEND_EVENTS = { + CONNECTION_DETAILS_STARTED: 'frontend.connection_details.started', + CONNECTION_DETAILS_FINISHED: 'frontend.connection_details.finished', + ROOM_CONNECT_STARTED: 'frontend.room_connect.started', + ROOM_CONNECT_FINISHED: 'frontend.room_connect.finished', ROOM_CONNECTED: 'frontend.room.connected', + DISPATCH_STARTED: 'frontend.dispatch.started', + DISPATCH_FINISHED: 'frontend.dispatch.finished', + BROWSER_AUDIO_CAPTURE_STARTED: 'frontend.browser_audio.capture_started', + BROWSER_AUDIO_CAPTURE_FINISHED: 'frontend.browser_audio.capture_finished', + BROWSER_AUDIO_PUBLISH_STARTED: 'frontend.browser_audio.publish_started', + BROWSER_AUDIO_PUBLISH_FINISHED: 'frontend.browser_audio.publish_finished', BROWSER_AUDIO_TRACK_PUBLISHED: 'frontend.browser_audio.track_published', BROWSER_AUDIO_TRACK_UNPUBLISHED: 'frontend.browser_audio.track_unpublished', BROWSER_AUDIO_TRACK_MUTED: 'frontend.browser_audio.track_muted', @@ -12,6 +22,11 @@ export const FRONTEND_EVENTS = { BROWSER_AUDIO_VAD_SPEECH_STARTED: 'frontend.browser_audio.vad_speech_started', BROWSER_AUDIO_VAD_SPEECH_ENDED: 'frontend.browser_audio.vad_speech_ended', BROWSER_AUDIO_VAD_PROBE_UNAVAILABLE: 'frontend.browser_audio.vad_probe_unavailable', + BROWSER_VIDEO_CAPTURE_STARTED: 'frontend.browser_video.capture_started', + BROWSER_VIDEO_CAPTURE_FINISHED: 'frontend.browser_video.capture_finished', + BROWSER_VIDEO_PUBLISH_STARTED: 'frontend.browser_video.publish_started', + BROWSER_VIDEO_PUBLISH_FINISHED: 'frontend.browser_video.publish_finished', + BROWSER_VIDEO_TRACK_PUBLISHED: 'frontend.browser_video.track_published', REPLY_AUDIO_PLAYBACK_STARTED: 'frontend.reply_audio.playback_started', REPLY_AUDIO_PLAYBACK_ENDED: 'frontend.reply_audio.playback_ended', REPLY_AUDIO_PLAYBACK_ERROR: 'frontend.reply_audio.playback_error', @@ -57,7 +72,7 @@ export interface BackendObservabilityMarker { const MAX_BACKEND_MARKER_NAME_LENGTH = 128; -type PublishableRoom = { +export type PublishableRoom = { name?: string; localParticipant?: { identity?: string; @@ -74,6 +89,7 @@ interface PublishFrontendObservabilityEventOptions { name: string; attributes?: Record; wallTimeUnixMs?: number; + performanceNowMs?: number; now?: () => number; performanceNow?: () => number; } @@ -84,6 +100,7 @@ export async function publishFrontendObservabilityEvent({ name, attributes, wallTimeUnixMs, + performanceNowMs, now = () => Date.now(), performanceNow = defaultPerformanceNow, }: PublishFrontendObservabilityEventOptions) { @@ -99,7 +116,7 @@ export async function publishFrontendObservabilityEvent({ type: OBSERVABILITY_EVENT_TYPES.FRONTEND_EVENT, name, wall_time_unix_ms: wallTimeUnixMs ?? now(), - performance_now_ms: performanceNow(), + performance_now_ms: performanceNowMs ?? performanceNow(), room_name: room.name || undefined, participant_identity: room.localParticipant?.identity || undefined, attributes: attributes ?? {}, @@ -111,6 +128,89 @@ export async function publishFrontendObservabilityEvent({ return true; } +interface BufferedFrontendObservabilityEvent { + name: string; + attributes?: Record; + wallTimeUnixMs: number; + performanceNowMs: number; +} + +interface FrontendObservabilitySession { + live: boolean; + events: BufferedFrontendObservabilityEvent[]; +} + +const frontendObservabilitySessions = new WeakMap(); + +export function beginFrontendObservabilitySession(room: PublishableRoom) { + frontendObservabilitySessions.set(room, { live: false, events: [] }); +} + +export function endFrontendObservabilitySession(room: PublishableRoom) { + frontendObservabilitySessions.delete(room); +} + +export async function recordFrontendObservabilityEvent({ + enabled, + room, + name, + attributes, + wallTimeUnixMs = Date.now(), + performanceNowMs = defaultPerformanceNow(), +}: Omit) { + if (!enabled) { + return false; + } + + const session = frontendObservabilitySessions.get(room); + if (session && !session.live) { + session.events.push({ name, attributes, wallTimeUnixMs, performanceNowMs }); + return true; + } + + return publishFrontendObservabilityEvent({ + enabled, + room, + name, + attributes, + wallTimeUnixMs, + performanceNowMs, + }); +} + +export async function flushFrontendObservabilityEvents({ + enabled, + room, +}: { + enabled: boolean; + room: PublishableRoom; +}) { + if (!enabled) { + return 0; + } + + const session = frontendObservabilitySessions.get(room); + if (!session || session.live) { + return 0; + } + + let published = 0; + while (session.events.length > 0) { + const event = session.events[0]; + try { + if (await publishFrontendObservabilityEvent({ enabled, room, ...event })) { + published += 1; + } + } catch (error) { + console.warn('[frontend-observability] failed to flush startup event', error); + } finally { + session.events.shift(); + } + } + session.live = true; + return published; +} + export function parseBackendObservabilityMarkerPayload( payload: Uint8Array | string, topic?: string diff --git a/lib/session-dispatch-client.ts b/lib/session-dispatch-client.ts index 72237306d..0d6b7ef1b 100644 --- a/lib/session-dispatch-client.ts +++ b/lib/session-dispatch-client.ts @@ -21,7 +21,7 @@ export async function requestAgentSessionDispatch( return; } - const response = await fetch('/api/session/dispatch', { + const response = await fetch('api/session/dispatch', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ diff --git a/lib/session-dispatch-readiness.ts b/lib/session-dispatch-readiness.ts index c23bfac7b..fd980d8b4 100644 --- a/lib/session-dispatch-readiness.ts +++ b/lib/session-dispatch-readiness.ts @@ -11,8 +11,10 @@ export type AgentParticipantMatchOptions = { export type ReusableAgentParticipantOptions = AgentParticipantMatchOptions & { requireRoomVideoInputReady?: boolean; + requireRoomInputParticipantsReady?: boolean; }; +const ROOM_AUDIO_INPUT_IDENTITY = 'room_audio_input'; const ROOM_VIDEO_INPUT_IDENTITY = 'room_video_input'; function readRoomInputVideoTrackName() { @@ -28,17 +30,35 @@ export function findReusableAgentParticipant( agentName: string, options: ReusableAgentParticipantOptions = {} ): ParticipantInfo | null { - const { requireRoomVideoInputReady = false, ...matchOptions } = options; + const { + requireRoomVideoInputReady = false, + requireRoomInputParticipantsReady = false, + ...matchOptions + } = options; const expectedAgent = findAgentParticipantInList(participants, agentName, matchOptions); if (!expectedAgent) { return null; } if (!requireRoomVideoInputReady) { - return expectedAgent; + return requireRoomInputParticipantsReady && !hasReadyRoomInputParticipants(participants) + ? null + : expectedAgent; + } + + if (!hasReadyRoomVideoInput(participants)) { + return null; } + return requireRoomInputParticipantsReady && !hasReadyRoomInputParticipants(participants) + ? null + : expectedAgent; +} - return hasReadyRoomVideoInput(participants) ? expectedAgent : null; +export function summarizeRoomInputReadiness(participants: ParticipantInfo[]) { + return { + audioParticipantReady: hasActiveParticipant(participants, ROOM_AUDIO_INPUT_IDENTITY), + visionParticipantReady: hasActiveParticipant(participants, ROOM_VIDEO_INPUT_IDENTITY), + }; } export function findAgentParticipantInList( @@ -76,6 +96,17 @@ function hasReadyRoomVideoInput(participants: ParticipantInfo[]) { ); } +function hasReadyRoomInputParticipants(participants: ParticipantInfo[]) { + const readiness = summarizeRoomInputReadiness(participants); + return readiness.audioParticipantReady && readiness.visionParticipantReady; +} + +function hasActiveParticipant(participants: ParticipantInfo[], identity: string) { + return participants.some( + (participant) => participant.identity === identity && isParticipantActive(participant) + ); +} + function isExpectedAgentParticipant(participant: ParticipantInfo, agentName: string) { return ( isParticipantActive(participant) && diff --git a/lib/session-stop-client.ts b/lib/session-stop-client.ts index fedbe595d..253da288a 100644 --- a/lib/session-stop-client.ts +++ b/lib/session-stop-client.ts @@ -64,7 +64,7 @@ async function sendAgentSessionStop( sessionId: string, options: AgentSessionStopOptions ): Promise { - const response = await fetch('/api/session/stop', { + const response = await fetch('api/session/stop', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ sessionId, wait: options.waitForRemote }), diff --git a/lib/utils.ts b/lib/utils.ts index 660a5512a..d8a15b629 100644 --- a/lib/utils.ts +++ b/lib/utils.ts @@ -176,6 +176,7 @@ export function getClientConfigFromEnv(): AppConfig { usesBrowserRawAudioInput: inputDeviceConfig.usesBrowserRawAudioInput, usesBrowserRawVideoInput: inputDeviceConfig.usesBrowserRawVideoInput, usesServerRoomInput: inputDeviceConfig.usesServerRoomInput, + voiceSessionId: readEnv('LIVEAVATAR_VOICE_SESSION_ID') || undefined, agentName: resolveAgentNameForInputSource(inputSource, agentName), inputSource: inputDeviceConfig.inputSource, audioInputDevice: inputDeviceConfig.audioInputDevice, @@ -334,6 +335,8 @@ export const getAppConfig = cache(async (headers: Headers): Promise = for (const [key, entry] of Object.entries(remoteConfig)) { if (entry === null) continue; + // The per-sandbox identity must stay aligned with the server-side prewarm route. + if (key === 'voiceSessionId') continue; // Only include app config entries declared in defaults and matching the // expected type. Structured fields get extra shape checks above. if (hasAppConfigKey(key) && canApplySandboxConfigEntry(key, entry)) { diff --git a/next.config.ts b/next.config.ts index 8564a5b86..cf939c7b9 100644 --- a/next.config.ts +++ b/next.config.ts @@ -1,6 +1,7 @@ import type { NextConfig } from 'next'; const nextConfig: NextConfig = { + assetPrefix: '.', allowedDevOrigins: ['liveavatar.local.lexmount.net'], }; diff --git a/styles/globals.css b/styles/globals.css index 6f39b8451..1fd151350 100644 --- a/styles/globals.css +++ b/styles/globals.css @@ -77,8 +77,8 @@ var(--font-public-sans, ui-sans-serif), system-ui, sans-serif, 'Apple Color Emoji', 'Segoe UI Emoji', 'Segoe UI Symbol', 'Noto Color Emoji'; --font-mono: - var(--font-commit-mono), ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, - 'Liberation Mono', 'Courier New', monospace; + ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, 'Liberation Mono', 'Courier New', + monospace; --color-background: var(--background); --color-foreground: var(--foreground); diff --git a/tests/browser-room-session.test.mjs b/tests/browser-room-session.test.mjs index 065fedf31..f802085ff 100644 --- a/tests/browser-room-session.test.mjs +++ b/tests/browser-room-session.test.mjs @@ -64,3 +64,27 @@ test('ending a voice session clears the reusable session id', async () => { assert.match(useRoomSource, /resetVoiceSessionId/); assert.doesNotMatch(useRoomSource, /if \(appConfig\.usesBrowserRawMediaInput\)/); }); + +test('browser room starts local media and agent dispatch concurrently after room connect', async () => { + const useRoomSource = await readFile(new URL('../hooks/useRoom.ts', import.meta.url), 'utf8'); + const browserSourceSource = await readFile( + new URL('../hooks/useBrowserSourceClient.ts', import.meta.url), + 'utf8' + ); + + assert.match( + useRoomSource, + /await room\.connect[\s\S]*connectedRoomName = room\.name;[\s\S]*await Promise\.allSettled\(\[[\s\S]*startLocalInputOrCancelDispatch\(\),[\s\S]*dispatchAgentSession\(\),[\s\S]*\]\)/ + ); + assert.match( + useRoomSource, + /const startLocalInputOrCancelDispatch = async \(\) => \{[\s\S]*await startLocalInput\(\);[\s\S]*catch \(error\) \{[\s\S]*cancelAgentSessionStart\(sessionId\);[\s\S]*throw error;/ + ); + assert.match(useRoomSource, /localInputResult\.status === 'rejected'/); + assert.match(useRoomSource, /dispatchResult\.status === 'rejected'/); + assert.match( + useRoomSource, + /localInputResult\.status === 'rejected'[\s\S]*dispatchResult\.status === 'rejected'[\s\S]*console\.warn/ + ); + assert.match(browserSourceSource, /Promise\.allSettled\(\[audioStart, videoStart\]\)/); +}); diff --git a/tests/dynamic-proxy-paths.test.mjs b/tests/dynamic-proxy-paths.test.mjs new file mode 100644 index 000000000..f2d503750 --- /dev/null +++ b/tests/dynamic-proxy-paths.test.mjs @@ -0,0 +1,45 @@ +import assert from 'node:assert/strict'; +import { readFile } from 'node:fs/promises'; +import { test } from 'node:test'; + +test('Next assets and bundled app resources use app-relative paths', async () => { + const [nextConfig, rootLayout, appConfig, styles] = await Promise.all([ + readFile('next.config.ts', 'utf8'), + readFile('app/layout.tsx', 'utf8'), + readFile('app-config.ts', 'utf8'), + readFile('styles/globals.css', 'utf8'), + ]); + + assert.match(nextConfig, /assetPrefix:\s*['"]\.['"]/); + assert.doesNotMatch(rootLayout, /next\/font/); + assert.doesNotMatch(styles, /--font-commit-mono/); + assert.match(appConfig, /logo:\s*['"]lk-logo\.png['"]/); + assert.match(appConfig, /logoDark:\s*['"]lk-logo-dark\.png['"]/); +}); + +test('browser API requests stay under the current app entry path', async () => { + const [dispatchClient, stopClient, useRoom] = await Promise.all([ + readFile('lib/session-dispatch-client.ts', 'utf8'), + readFile('lib/session-stop-client.ts', 'utf8'), + readFile('hooks/useRoom.ts', 'utf8'), + ]); + + assert.match(dispatchClient, /fetch\(['"]api\/session\/dispatch['"]/); + assert.match(stopClient, /fetch\(['"]api\/session\/stop['"]/); + assert.match(useRoom, /NEXT_PUBLIC_CONN_DETAILS_ENDPOINT \?\? ['"]api\/connection-details['"]/); + assert.match(useRoom, /window\.location\.href/); + + assert.equal( + new URL('api/connection-details', 'http://127.0.0.1:4003/').pathname, + '/api/connection-details' + ); + assert.equal( + new URL('api/connection-details', 'https://gateway.example/s/session-slug').pathname, + '/s/api/connection-details' + ); + assert.equal( + new URL('api/connection-details', 'https://sandbox.example/api/v1/sandboxes/id/proxy/4003/') + .pathname, + '/api/v1/sandboxes/id/proxy/4003/api/connection-details' + ); +}); diff --git a/tests/frontend-vad-observer.test.mjs b/tests/frontend-vad-observer.test.mjs index 088208d6a..b402e803f 100644 --- a/tests/frontend-vad-observer.test.mjs +++ b/tests/frontend-vad-observer.test.mjs @@ -1,7 +1,24 @@ import assert from 'node:assert/strict'; import { test } from 'node:test'; -const { startMediaTrackVadObserver } = await import('../lib/frontend-vad-observer.ts'); +const { resolveVadAssetBasePaths, startMediaTrackVadObserver } = await import( + '../lib/frontend-vad-observer.ts' +); + +test('vad asset paths stay local and become session scoped behind the gateway', () => { + assert.deepEqual(resolveVadAssetBasePaths(), { + baseAssetPath: '/vad-web/', + onnxWASMBasePath: '/onnxruntime-web/', + }); + assert.deepEqual(resolveVadAssetBasePaths('/sandbox-session'), { + baseAssetPath: '/sandbox-session/vad-web/', + onnxWASMBasePath: '/sandbox-session/onnxruntime-web/', + }); + assert.deepEqual(resolveVadAssetBasePaths('/s/sandbox-session/'), { + baseAssetPath: '/s/sandbox-session/vad-web/', + onnxWASMBasePath: '/s/sandbox-session/onnxruntime-web/', + }); +}); test('media track vad observer uses the supplied track and emits speech events', async () => { const originalMediaStream = globalThis.MediaStream; diff --git a/tests/local-dispatch-config.test.mjs b/tests/local-dispatch-config.test.mjs index 0c41a157b..029c4dc2b 100644 --- a/tests/local-dispatch-config.test.mjs +++ b/tests/local-dispatch-config.test.mjs @@ -37,6 +37,20 @@ test('frontend keeps explicit AGENT_NAME as an override', async () => { assert.equal(resolveAgentNameForInputSource('generic', 'custom-agent'), 'custom-agent'); }); +test('frontend exposes the server-owned voice session id to dispatch callers', async () => { + const previousEnv = { ...process.env }; + + try { + process.env.LIVEAVATAR_VOICE_SESSION_ID = 'sandbox-session-123'; + + const { getClientConfigFromEnv } = await loadUtilsModule(); + + assert.equal(getClientConfigFromEnv().voiceSessionId, 'sandbox-session-123'); + } finally { + restoreEnv(previousEnv); + } +}); + test('frontend defaults to browser input when INPUT_SOURCE is unset', async () => { const { normalizeInputSource, resolveInputDeviceConfig } = await loadAppConfigModule(); diff --git a/tests/observability.test.mjs b/tests/observability.test.mjs index 28cc69ec3..8bb647d1e 100644 --- a/tests/observability.test.mjs +++ b/tests/observability.test.mjs @@ -9,9 +9,13 @@ const { FRONTEND_EVENTS, OBSERVABILITY_ATTRS, OBSERVABILITY_EVENT_TYPES, + beginFrontendObservabilitySession, + endFrontendObservabilitySession, + flushFrontendObservabilityEvents, outputSegmentAttributesFromMarker, parseBackendObservabilityMarkerPayload, publishFrontendObservabilityEvent, + recordFrontendObservabilityEvent, } = await import('../lib/observability.ts'); test('frontend observability does not publish when disabled', async () => { @@ -135,6 +139,92 @@ test('frontend observability can publish an explicit event wall time', async () assert.equal(payload.performance_now_ms, 456.78); }); +test('frontend observability buffers startup events until the agent can receive them', async () => { + const calls = []; + const room = { + name: 'voice_assistant_room_a', + localParticipant: { + identity: 'voice_assistant_user_a', + publishData: async (...args) => { + calls.push(args); + }, + }, + }; + + beginFrontendObservabilitySession(room); + await recordFrontendObservabilityEvent({ + enabled: true, + room, + name: FRONTEND_EVENTS.ROOM_CONNECT_STARTED, + wallTimeUnixMs: 100, + performanceNowMs: 10, + }); + await recordFrontendObservabilityEvent({ + enabled: true, + room, + name: FRONTEND_EVENTS.ROOM_CONNECT_FINISHED, + wallTimeUnixMs: 200, + performanceNowMs: 20, + }); + + assert.equal(calls.length, 0); + assert.equal(await flushFrontendObservabilityEvents({ enabled: true, room }), 2); + assert.equal(calls.length, 2); + assert.deepEqual( + calls.map(([payload]) => { + const event = JSON.parse(new TextDecoder().decode(payload)); + return [event.name, event.wall_time_unix_ms, event.performance_now_ms]; + }), + [ + [FRONTEND_EVENTS.ROOM_CONNECT_STARTED, 100, 10], + [FRONTEND_EVENTS.ROOM_CONNECT_FINISHED, 200, 20], + ] + ); + + await recordFrontendObservabilityEvent({ + enabled: true, + room, + name: FRONTEND_EVENTS.DISPATCH_FINISHED, + wallTimeUnixMs: 300, + performanceNowMs: 30, + }); + assert.equal(calls.length, 3); + endFrontendObservabilitySession(room); +}); + +test('frontend observability flush is best effort and switches to live publishing', async () => { + let publishCalls = 0; + const room = { + localParticipant: { + publishData: async () => { + publishCalls += 1; + if (publishCalls === 1) throw new Error('participant not ready'); + }, + }, + }; + beginFrontendObservabilitySession(room); + await recordFrontendObservabilityEvent({ + enabled: true, + room, + name: FRONTEND_EVENTS.ROOM_CONNECT_STARTED, + }); + await recordFrontendObservabilityEvent({ + enabled: true, + room, + name: FRONTEND_EVENTS.ROOM_CONNECT_FINISHED, + }); + + assert.equal(await flushFrontendObservabilityEvents({ enabled: true, room }), 1); + await recordFrontendObservabilityEvent({ + enabled: true, + room, + name: FRONTEND_EVENTS.DISPATCH_FINISHED, + }); + + assert.equal(publishCalls, 3); + endFrontendObservabilitySession(room); +}); + test('frontend observability parses backend output segment markers', () => { const payload = { schema_version: 1, @@ -341,9 +431,18 @@ test('frontend audio observer reuses shared observability attribute types', asyn test('room hook publishes room connected frontend observability event', async () => { const source = await readFile('hooks/useRoom.ts', 'utf8'); + const recoverySource = source.slice( + source.indexOf('const recoverFromStartError'), + source.indexOf('const handleStartError') + ); assert.match(source, /FRONTEND_EVENTS\.ROOM_CONNECTED/); - assert.match(source, /publishFrontendObservabilityEvent/); + assert.match(source, /recordFrontendObservabilityEvent/); + assert.match(source, /flushFrontendObservabilityEvents/); + assert.ok( + recoverySource.indexOf('flushFrontendObservabilityEvents') < + recoverySource.indexOf('room.disconnect()') + ); assert.match( source, /const recoverFromStartError = async[\s\S]*try \{[\s\S]*await browserSourceClient\.stop\(\);[\s\S]*\} catch \(stopError\)[\s\S]*\} finally \{[\s\S]*room\.disconnect\(\);[\s\S]*\}/ diff --git a/tests/session-dispatch-readiness.test.mjs b/tests/session-dispatch-readiness.test.mjs index dfb4e8a39..675131550 100644 --- a/tests/session-dispatch-readiness.test.mjs +++ b/tests/session-dispatch-readiness.test.mjs @@ -83,6 +83,44 @@ test('dispatch can reuse an active agent once room video input is publishing', ( ); }); +test('prewarm readiness requires both room input participants without requiring a video frame', () => { + const agent = participant({ + identity: 'agent-AJ_running', + kind: ParticipantInfo_Kind.AGENT, + attributes: { 'lk.agent.name': 'frontdesk-browser-agent' }, + }); + const participants = [ + agent, + participant({ identity: 'room_audio_input' }), + participant({ identity: 'room_video_input' }), + ]; + + assert.equal( + findReusableAgentParticipant(participants, 'frontdesk-browser-agent', { + requireRoomInputParticipantsReady: true, + }), + agent + ); +}); + +test('prewarm readiness rejects a room missing either input participant', () => { + const participants = [ + participant({ + identity: 'agent-AJ_running', + kind: ParticipantInfo_Kind.AGENT, + attributes: { 'lk.agent.name': 'frontdesk-browser-agent' }, + }), + participant({ identity: 'room_video_input' }), + ]; + + assert.equal( + findReusableAgentParticipant(participants, 'frontdesk-browser-agent', { + requireRoomInputParticipantsReady: true, + }), + null + ); +}); + test('dispatch does not reuse disconnected agents', () => { const participants = [ participant({ diff --git a/tests/session-prewarm.test.mjs b/tests/session-prewarm.test.mjs new file mode 100644 index 000000000..32e66cdaf --- /dev/null +++ b/tests/session-prewarm.test.mjs @@ -0,0 +1,1474 @@ +import assert from 'node:assert/strict'; +import { readFile } from 'node:fs/promises'; +import { test } from 'node:test'; +import { ParticipantInfo_Kind, ParticipantInfo_State, TrackType } from '@livekit/protocol'; +import { + beginPrewarmUse, + buildPrewarmUseKey, + completePrewarmUse, + failPrewarmUse, + releasePrewarmUseAfterFailure, +} from '@/app/api/session/prewarm/prewarm-use-guard'; +import { + resolveAgentWorkerReadyFile, + waitForAgentWorkerReady, +} from '../app/api/session/agent-worker-readiness.ts'; +import { POST as prewarmRoute } from '../app/api/session/prewarm/route.ts'; +import { + PrewarmRoomSessionError, + dispatchRoomSession, + prewarmRoomSession, +} from '../app/api/session/session-dispatch-service.ts'; +import { getRoomSessionSnapshot } from '../app/api/session/session-registry.ts'; + +function activeParticipant(identity, attributes = {}) { + return { + identity, + kind: identity.startsWith('agent-') + ? ParticipantInfo_Kind.AGENT + : ParticipantInfo_Kind.STANDARD, + state: ParticipantInfo_State.ACTIVE, + attributes, + tracks: [], + }; +} + +function readyParticipants(agentName, { videoReady = false } = {}) { + const videoParticipant = activeParticipant('room_video_input'); + if (videoReady) { + videoParticipant.tracks = [{ name: 'room_video', type: TrackType.VIDEO, muted: false }]; + } + return [ + activeParticipant('agent-ready', { 'lk.agent.name': agentName }), + activeParticipant('room_audio_input'), + videoParticipant, + ]; +} + +test('prewarm route rejects requests without the per-sandbox secret', async () => { + const previous = process.env.LIVEAVATAR_PREWARM_SECRET; + process.env.LIVEAVATAR_PREWARM_SECRET = 'expected-prewarm-secret'; + try { + const missing = await prewarmRoute( + new Request('http://sandbox.example.test/api/session/prewarm', { method: 'POST' }) + ); + const wrong = await prewarmRoute( + new Request('http://sandbox.example.test/api/session/prewarm', { + method: 'POST', + headers: { 'x-liveavatar-prewarm-secret': 'wrong-prewarm-secret' }, + }) + ); + + assert.equal(missing.status, 401); + assert.equal(wrong.status, 401); + } finally { + if (previous === undefined) { + delete process.env.LIVEAVATAR_PREWARM_SECRET; + } else { + process.env.LIVEAVATAR_PREWARM_SECRET = previous; + } + } +}); + +test('prewarm authorization is single-use after success and retryable after failure', () => { + const completedKey = buildPrewarmUseKey( + 'completed-session', + 'voice_assistant_room_completed-session', + 'frontdesk-browser-agent-completed-session' + ); + assert.equal(beginPrewarmUse(completedKey), 'started'); + assert.equal(beginPrewarmUse(completedKey), 'in_progress'); + completePrewarmUse(completedKey); + assert.equal(beginPrewarmUse(completedKey), 'completed'); + + const retryableKey = buildPrewarmUseKey( + 'retryable-session', + 'voice_assistant_room_retryable-session', + 'frontdesk-browser-agent-retryable-session' + ); + assert.equal(beginPrewarmUse(retryableKey), 'started'); + failPrewarmUse(retryableKey); + assert.equal(beginPrewarmUse(retryableKey), 'started'); + failPrewarmUse(retryableKey); +}); + +test('prewarm route returns 409 after its server-owned authorization is consumed', async () => { + const sessionId = 'a16e0a10-4f28-4a78-8f1f-019c25a273cb'; + const roomName = `voice_assistant_room_${sessionId}`; + const agentName = 'frontdesk-browser-agent-consumed'; + const secret = 'consumed-prewarm-secret'; + const envNames = [ + 'LIVEAVATAR_PREWARM_SECRET', + 'LIVEAVATAR_VOICE_SESSION_ID', + 'LIVEAVATAR_LIVEKIT_ROOM_NAME', + 'AGENT_NAME', + ]; + const previous = Object.fromEntries(envNames.map((name) => [name, process.env[name]])); + Object.assign(process.env, { + LIVEAVATAR_PREWARM_SECRET: secret, + LIVEAVATAR_VOICE_SESSION_ID: sessionId, + LIVEAVATAR_LIVEKIT_ROOM_NAME: roomName, + AGENT_NAME: agentName, + }); + + const useKey = buildPrewarmUseKey(sessionId, roomName, agentName); + assert.equal(beginPrewarmUse(useKey), 'started'); + completePrewarmUse(useKey); + + try { + const response = await prewarmRoute( + new Request('http://sandbox.example.test/api/session/prewarm', { + method: 'POST', + headers: { 'x-liveavatar-prewarm-secret': secret }, + }) + ); + + assert.equal(response.status, 409); + assert.deepEqual(await response.json(), { + status: 'error', + error: 'prewarm authorization already consumed', + }); + } finally { + for (const name of envNames) { + if (previous[name] === undefined) { + delete process.env[name]; + } else { + process.env[name] = previous[name]; + } + } + } +}); + +test('prewarm route returns a structured retryable 502 without leaking its secret', async () => { + const sessionId = 'c5b8c624-7f55-4acf-bbe5-a7ddc634a101'; + const roomName = `voice_assistant_room_${sessionId}`; + const agentName = 'frontdesk-browser-agent-room-failure'; + const secret = 'room-failure-prewarm-secret'; + const envNames = [ + 'LIVEAVATAR_PREWARM_SECRET', + 'LIVEAVATAR_VOICE_SESSION_ID', + 'LIVEAVATAR_LIVEKIT_ROOM_NAME', + 'AGENT_NAME', + 'LIVEKIT_URL', + 'LIVEKIT_API_KEY', + 'LIVEKIT_API_SECRET', + ]; + const previous = Object.fromEntries(envNames.map((name) => [name, process.env[name]])); + const originalConsoleError = console.error; + const errorLogs = []; + console.error = (...args) => { + errorLogs.push(args); + }; + Object.assign(process.env, { + LIVEAVATAR_PREWARM_SECRET: secret, + LIVEAVATAR_VOICE_SESSION_ID: sessionId, + LIVEAVATAR_LIVEKIT_ROOM_NAME: roomName, + AGENT_NAME: agentName, + }); + delete process.env.LIVEKIT_URL; + delete process.env.LIVEKIT_API_KEY; + delete process.env.LIVEKIT_API_SECRET; + + try { + for (let attempt = 0; attempt < 2; attempt += 1) { + const response = await prewarmRoute( + new Request('http://sandbox.example.test/api/session/prewarm', { + method: 'POST', + headers: { 'x-liveavatar-prewarm-secret': secret }, + }) + ); + const payload = await response.json(); + + assert.equal(response.status, 502); + assert.equal(response.headers.get('Cache-Control'), 'no-store'); + assert.equal(payload.phase, 'room'); + assert.equal(Number.isInteger(payload.timings.totalPrewarmMs), true); + assert.equal(Number.isInteger(payload.timings.roomEnsureMs), true); + assert.deepEqual(Object.keys(payload.timings).sort(), ['roomEnsureMs', 'totalPrewarmMs']); + assert.equal(JSON.stringify(payload).includes(secret), false); + assert.equal(JSON.stringify(payload).includes('attributes'), false); + } + assert.equal(errorLogs.length, 2); + assert.equal( + errorLogs.every((entry) => entry[0] === 'session prewarm failed'), + true + ); + assert.equal( + errorLogs.every((entry) => entry[1]?.phase === 'room'), + true + ); + assert.equal(JSON.stringify(errorLogs).includes(secret), false); + } finally { + console.error = originalConsoleError; + for (const name of envNames) { + if (previous[name] === undefined) { + delete process.env[name]; + } else { + process.env[name] = previous[name]; + } + } + } +}); + +test('prewarm phase errors preserve their original cause', () => { + const cause = Object.assign(new Error('livekit request failed'), { code: 'ECONNRESET' }); + const error = new PrewarmRoomSessionError('room', { totalPrewarmMs: 5, roomEnsureMs: 5 }, cause); + + assert.equal(error.cause, cause); +}); + +test('prewarm authorization remains in progress until timed-out work settles', async () => { + const key = buildPrewarmUseKey( + 'settling-session', + 'voice_assistant_room_settling-session', + 'frontdesk-browser-agent-settling-session' + ); + let releaseSettlement; + const retryReady = new Promise((resolve) => { + releaseSettlement = resolve; + }); + const error = new PrewarmRoomSessionError( + 'room', + { totalPrewarmMs: 5, roomEnsureMs: 5 }, + new Error('room deadline expired'), + retryReady + ); + + assert.equal(beginPrewarmUse(key), 'started'); + releasePrewarmUseAfterFailure(key, error); + assert.equal(beginPrewarmUse(key), 'in_progress'); + + releaseSettlement(); + await retryReady; + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(beginPrewarmUse(key), 'started'); + failPrewarmUse(key); +}); + +test('prewarm route allowlists readiness and timings instead of spreading internal results', async () => { + const source = await readFile( + new URL('../app/api/session/prewarm/route.ts', import.meta.url), + 'utf8' + ); + const successSource = source.slice( + source.indexOf('const result = await prewarmRoomSession'), + source.indexOf('} catch (error)') + ); + + assert.doesNotMatch(successSource, /\.\.\.result/); + assert.match(successSource, /readiness:\s*result\.readiness/); + assert.match(successSource, /timings:\s*result\.timings/); + assert.doesNotMatch(successSource, /workerReadiness:\s*result\.workerReadiness/); + assert.doesNotMatch(successSource, /dispatch:\s*result\.dispatch/); +}); + +test('missing LiveKit configuration fails before registering a room session', async () => { + const names = ['LIVEKIT_URL', 'LIVEKIT_API_KEY', 'LIVEKIT_API_SECRET']; + const previous = Object.fromEntries(names.map((name) => [name, process.env[name]])); + names.forEach((name) => delete process.env[name]); + + const request = { + roomName: 'voice_assistant_room_missing_config', + sessionId: 'missing-config', + agentName: 'frontdesk-browser-agent-missing-config', + }; + try { + await assert.rejects(dispatchRoomSession(request), /LiveKit API configuration is required/); + assert.equal(getRoomSessionSnapshot(request.roomName), undefined); + } finally { + for (const name of names) { + if (previous[name] === undefined) { + delete process.env[name]; + } else { + process.env[name] = previous[name]; + } + } + } +}); + +test('regular dispatch keeps its 8s timeout while prewarm gets the default 45s total budget', async () => { + const originalNow = Date.now; + const originalTimeout = process.env.AGENT_DISPATCH_TIMEOUT_MS; + const originalPrewarmTimeout = process.env.LIVEAVATAR_PREWARM_TOTAL_TIMEOUT_MS; + let now = 1_000; + let dispatchCount = 0; + Date.now = () => now; + delete process.env.AGENT_DISPATCH_TIMEOUT_MS; + delete process.env.LIVEAVATAR_PREWARM_TOTAL_TIMEOUT_MS; + + const dispatchClient = { + async createDispatch() { + dispatchCount += 1; + return { id: `dispatch-timeout-${dispatchCount}` }; + }, + async deleteDispatch() {}, + }; + const roomClient = { + async listRooms([roomName]) { + return [{ name: roomName }]; + }, + async createRoom({ name }) { + return { name }; + }, + async listParticipants() { + return []; + }, + async deleteRoom() {}, + }; + const dependencies = { + dispatchClient, + roomClient, + dispatchPollMs: 1_000, + dispatchRetryMs: 1_000, + sleep: async (ms) => { + now += ms; + }, + waitForAgentWorkerReady: async () => ({ state: 'not_required' }), + }; + + try { + const regularStartedAt = now; + await assert.rejects( + dispatchRoomSession( + { + roomName: 'voice_assistant_room_regular_timeout', + sessionId: 'regular-timeout', + agentName: 'frontdesk-browser-agent-regular-timeout', + }, + dependencies + ), + /agent dispatch failed/ + ); + assert.equal(now - regularStartedAt, 8_000); + + const prewarmStartedAt = now; + await assert.rejects( + prewarmRoomSession( + { + roomName: 'voice_assistant_room_prewarm_timeout', + sessionId: 'prewarm-timeout', + agentName: 'frontdesk-browser-agent-prewarm-timeout', + }, + dependencies + ), + /agent dispatch failed/ + ); + assert.equal(now - prewarmStartedAt, 45_000); + } finally { + Date.now = originalNow; + if (originalTimeout === undefined) { + delete process.env.AGENT_DISPATCH_TIMEOUT_MS; + } else { + process.env.AGENT_DISPATCH_TIMEOUT_MS = originalTimeout; + } + if (originalPrewarmTimeout === undefined) { + delete process.env.LIVEAVATAR_PREWARM_TOTAL_TIMEOUT_MS; + } else { + process.env.LIVEAVATAR_PREWARM_TOTAL_TIMEOUT_MS = originalPrewarmTimeout; + } + } +}); + +test('prewarm shares its 45s total budget across worker readiness and dispatch', async () => { + const originalNow = Date.now; + const originalPrewarmTimeout = process.env.LIVEAVATAR_PREWARM_TOTAL_TIMEOUT_MS; + const originalWorkerTimeout = process.env.LIVEAVATAR_AGENT_WORKER_READY_TIMEOUT_MS; + let now = 1_000; + let workerMaxWaitMs; + Date.now = () => now; + delete process.env.LIVEAVATAR_PREWARM_TOTAL_TIMEOUT_MS; + delete process.env.LIVEAVATAR_AGENT_WORKER_READY_TIMEOUT_MS; + + const dispatchClient = { + async createDispatch() { + return { id: 'dispatch-shared-prewarm-timeout' }; + }, + async deleteDispatch() {}, + }; + const roomClient = { + async listRooms() { + return [{ name: 'voice_assistant_room_shared_prewarm_timeout' }]; + }, + async createRoom({ name }) { + return { name }; + }, + async listParticipants() { + return []; + }, + async deleteRoom() {}, + }; + + try { + const startedAt = now; + await assert.rejects( + prewarmRoomSession( + { + roomName: 'voice_assistant_room_shared_prewarm_timeout', + sessionId: 'shared-prewarm-timeout', + agentName: 'frontdesk-browser-agent-shared-prewarm-timeout', + }, + { + dispatchClient, + roomClient, + waitForAgentWorkerReady: async (_agentName, options) => { + workerMaxWaitMs = options.maxWaitMs; + now += 12_000; + return { + state: 'skipped', + agentName: _agentName, + reason: 'not_sandbox', + waitedMs: 0, + }; + }, + dispatchPollMs: 1_000, + dispatchRetryMs: 1_000, + sleep: async (ms) => { + now += ms; + }, + } + ), + /agent dispatch failed/ + ); + + assert.equal(workerMaxWaitMs, 30_000); + assert.equal(now - startedAt, 45_000); + } finally { + Date.now = originalNow; + if (originalPrewarmTimeout === undefined) { + delete process.env.LIVEAVATAR_PREWARM_TOTAL_TIMEOUT_MS; + } else { + process.env.LIVEAVATAR_PREWARM_TOTAL_TIMEOUT_MS = originalPrewarmTimeout; + } + if (originalWorkerTimeout === undefined) { + delete process.env.LIVEAVATAR_AGENT_WORKER_READY_TIMEOUT_MS; + } else { + process.env.LIVEAVATAR_AGENT_WORKER_READY_TIMEOUT_MS = originalWorkerTimeout; + } + } +}); + +test('room 2s plus worker 17s still completes dispatch readiness before the deadline', async () => { + const originalNow = Date.now; + const originalTotalTimeout = process.env.LIVEAVATAR_PREWARM_TOTAL_TIMEOUT_MS; + const originalWorkerTimeout = process.env.LIVEAVATAR_AGENT_WORKER_READY_TIMEOUT_MS; + let now = 1_000; + const startedAt = now; + const absoluteDeadline = startedAt + 45_000; + const agentName = 'frontdesk-browser-agent-cold-start-model'; + let roomEnsureStarted = false; + let workerMaxWaitMs; + Date.now = () => now; + delete process.env.LIVEAVATAR_PREWARM_TOTAL_TIMEOUT_MS; + delete process.env.LIVEAVATAR_AGENT_WORKER_READY_TIMEOUT_MS; + + const dispatchClient = { + async createDispatch() { + return { id: 'dispatch-cold-start-model' }; + }, + async deleteDispatch() {}, + }; + const roomClient = { + async listRooms() { + if (!roomEnsureStarted) { + roomEnsureStarted = true; + now += 2_000; + } + return [{ name: 'voice_assistant_room_cold_start_model' }]; + }, + async createRoom({ name }) { + return { name }; + }, + async listParticipants() { + return now >= absoluteDeadline - 1_000 ? readyParticipants(agentName) : []; + }, + async deleteRoom() {}, + }; + + try { + const result = await prewarmRoomSession( + { + roomName: 'voice_assistant_room_cold_start_model', + sessionId: 'cold-start-model', + agentName, + }, + { + dispatchClient, + roomClient, + waitForAgentWorkerReady: async (requestedAgentName, options) => { + assert.equal(requestedAgentName, agentName); + workerMaxWaitMs = options.maxWaitMs; + now += 17_000; + return { + state: 'ready', + agentName, + workerId: 'AW_cold_start_model', + registeredAt: '2026-07-24T00:00:00Z', + waitedMs: 17_000, + }; + }, + dispatchPollMs: 1_000, + dispatchRetryMs: 1_000, + sleep: async (ms) => { + now += ms; + }, + } + ); + + assert.equal(workerMaxWaitMs, 30_000); + assert.equal(now, absoluteDeadline - 1_000); + assert.deepEqual(result.timings, { + totalPrewarmMs: 44_000, + roomEnsureMs: 2_000, + workerReadyWaitMs: 17_000, + dispatchReadinessMs: 25_000, + }); + assert.deepEqual(result.readiness, { + audioParticipantReady: true, + visionParticipantReady: true, + }); + } finally { + Date.now = originalNow; + if (originalTotalTimeout === undefined) { + delete process.env.LIVEAVATAR_PREWARM_TOTAL_TIMEOUT_MS; + } else { + process.env.LIVEAVATAR_PREWARM_TOTAL_TIMEOUT_MS = originalTotalTimeout; + } + if (originalWorkerTimeout === undefined) { + delete process.env.LIVEAVATAR_AGENT_WORKER_READY_TIMEOUT_MS; + } else { + process.env.LIVEAVATAR_AGENT_WORKER_READY_TIMEOUT_MS = originalWorkerTimeout; + } + } +}); + +test('dispatch deadline never starts participant IO, dispatch creation, or sleep at zero budget', async () => { + const originalNow = Date.now; + let now = 1_000; + const deadline = now + 1_000; + let participantReadsAtDeadline = 0; + let dispatchCreatesAtDeadline = 0; + let sleepsAtDeadline = 0; + Date.now = () => now; + + try { + await assert.rejects( + prewarmRoomSession( + { + roomName: 'voice_assistant_room_hard_deadline', + sessionId: 'hard-deadline', + agentName: 'frontdesk-browser-agent-hard-deadline', + }, + { + dispatchClient: { + async createDispatch() { + if (now >= deadline) { + dispatchCreatesAtDeadline += 1; + } + return { id: 'dispatch-hard-deadline' }; + }, + async deleteDispatch() {}, + }, + roomClient: { + async listRooms() { + return [{ name: 'voice_assistant_room_hard_deadline' }]; + }, + async createRoom({ name }) { + return { name }; + }, + async listParticipants() { + if (now >= deadline) { + participantReadsAtDeadline += 1; + } + return []; + }, + async deleteRoom() {}, + }, + waitForAgentWorkerReady: async (requestedAgentName) => ({ + state: 'ready', + agentName: requestedAgentName, + workerId: 'AW_hard_deadline', + registeredAt: '2026-07-24T00:00:00Z', + waitedMs: 0, + }), + dispatchTimeoutMs: 1_000, + dispatchPollMs: 1_000, + dispatchRetryMs: 1_000, + sleep: async (ms) => { + if (now >= deadline) { + sleepsAtDeadline += 1; + } + now += ms; + }, + } + ), + /prewarm failed during dispatch_readiness/ + ); + + assert.equal(participantReadsAtDeadline, 0); + assert.equal(dispatchCreatesAtDeadline, 0); + assert.equal(sleepsAtDeadline, 0); + } finally { + Date.now = originalNow; + } +}); + +test('room timeout cannot create a room after a delayed list operation finishes', async () => { + const originalNow = Date.now; + Date.now = () => 1_000; + let releaseListRooms; + let markListRoomsStarted; + let createRoomCalls = 0; + const listRoomsStarted = new Promise((resolve) => { + markListRoomsStarted = resolve; + }); + const listRoomsGate = new Promise((resolve) => { + releaseListRooms = resolve; + }); + try { + const pending = prewarmRoomSession( + { + roomName: 'voice_assistant_room_late_list', + sessionId: 'late-list', + agentName: 'frontdesk-browser-agent-late-list', + }, + { + dispatchClient: { + async createDispatch() { + throw new Error('dispatch should not start'); + }, + async deleteDispatch() {}, + }, + roomClient: { + async listRooms() { + markListRoomsStarted(); + await listRoomsGate; + return []; + }, + async createRoom({ name }) { + createRoomCalls += 1; + return { name }; + }, + async listParticipants() { + return []; + }, + async deleteRoom() {}, + }, + dispatchTimeoutMs: 5, + } + ); + + await listRoomsStarted; + await assert.rejects(pending, /prewarm failed during room/); + releaseListRooms(); + await new Promise((resolve) => setImmediate(resolve)); + + assert.equal(createRoomCalls, 0); + } finally { + releaseListRooms?.(); + Date.now = originalNow; + } +}); + +test('room timeout removes a room whose delayed creation completes after the deadline', async () => { + const originalNow = Date.now; + Date.now = () => 1_000; + let releaseCreateRoom; + let markCreateRoomStarted; + let deleteRoomCalls = 0; + const createRoomStarted = new Promise((resolve) => { + markCreateRoomStarted = resolve; + }); + const createRoomGate = new Promise((resolve) => { + releaseCreateRoom = resolve; + }); + try { + const pending = prewarmRoomSession( + { + roomName: 'voice_assistant_room_late_create', + sessionId: 'late-create', + agentName: 'frontdesk-browser-agent-late-create', + }, + { + dispatchClient: { + async createDispatch() { + throw new Error('dispatch should not start'); + }, + async deleteDispatch() {}, + }, + roomClient: { + async listRooms() { + return []; + }, + async createRoom({ name }) { + markCreateRoomStarted(); + await createRoomGate; + return { name }; + }, + async listParticipants() { + return []; + }, + async deleteRoom() { + deleteRoomCalls += 1; + }, + }, + dispatchTimeoutMs: 5, + } + ); + + await createRoomStarted; + await assert.rejects(pending, /prewarm failed during room/); + releaseCreateRoom(); + await new Promise((resolve) => setImmediate(resolve)); + + assert.equal(deleteRoomCalls, 1); + } finally { + releaseCreateRoom?.(); + Date.now = originalNow; + } +}); + +test('prewarm total timeout and worker cap honor their environment overrides', async () => { + const originalNow = Date.now; + const originalTotalTimeout = process.env.LIVEAVATAR_PREWARM_TOTAL_TIMEOUT_MS; + const originalWorkerTimeout = process.env.LIVEAVATAR_AGENT_WORKER_READY_TIMEOUT_MS; + let now = 1_000; + let workerMaxWaitMs; + let roomEnsureStarted = false; + Date.now = () => now; + process.env.LIVEAVATAR_PREWARM_TOTAL_TIMEOUT_MS = '12000'; + process.env.LIVEAVATAR_AGENT_WORKER_READY_TIMEOUT_MS = '7000'; + const agentName = 'frontdesk-browser-agent-env-budget'; + + try { + const result = await prewarmRoomSession( + { + roomName: 'voice_assistant_room_env_budget', + sessionId: 'env-budget', + agentName, + }, + { + dispatchClient: { + async createDispatch() { + throw new Error('ready participants should be reused'); + }, + async deleteDispatch() {}, + }, + roomClient: { + async listRooms() { + if (!roomEnsureStarted) { + roomEnsureStarted = true; + now += 2_000; + } + return [{ name: 'voice_assistant_room_env_budget' }]; + }, + async createRoom({ name }) { + return { name }; + }, + async listParticipants() { + return readyParticipants(agentName); + }, + async deleteRoom() {}, + }, + waitForAgentWorkerReady: async (requestedAgentName, options) => { + workerMaxWaitMs = options.maxWaitMs; + return { + state: 'ready', + agentName: requestedAgentName, + workerId: 'AW_env_budget', + registeredAt: '2026-07-24T00:00:00Z', + waitedMs: 0, + }; + }, + } + ); + + assert.equal(workerMaxWaitMs, 7_000); + assert.deepEqual(result.timings, { + totalPrewarmMs: 2_000, + roomEnsureMs: 2_000, + workerReadyWaitMs: 0, + dispatchReadinessMs: 0, + }); + } finally { + Date.now = originalNow; + if (originalTotalTimeout === undefined) { + delete process.env.LIVEAVATAR_PREWARM_TOTAL_TIMEOUT_MS; + } else { + process.env.LIVEAVATAR_PREWARM_TOTAL_TIMEOUT_MS = originalTotalTimeout; + } + if (originalWorkerTimeout === undefined) { + delete process.env.LIVEAVATAR_AGENT_WORKER_READY_TIMEOUT_MS; + } else { + process.env.LIVEAVATAR_AGENT_WORKER_READY_TIMEOUT_MS = originalWorkerTimeout; + } + } +}); + +test('worker max wait is capped by the total time remaining after room ensure', async () => { + const originalNow = Date.now; + const originalTotalTimeout = process.env.LIVEAVATAR_PREWARM_TOTAL_TIMEOUT_MS; + const originalWorkerTimeout = process.env.LIVEAVATAR_AGENT_WORKER_READY_TIMEOUT_MS; + let now = 1_000; + let workerMaxWaitMs; + let roomEnsureStarted = false; + Date.now = () => now; + process.env.LIVEAVATAR_PREWARM_TOTAL_TIMEOUT_MS = '5000'; + process.env.LIVEAVATAR_AGENT_WORKER_READY_TIMEOUT_MS = '7000'; + const agentName = 'frontdesk-browser-agent-remaining-budget'; + + try { + await prewarmRoomSession( + { + roomName: 'voice_assistant_room_remaining_budget', + sessionId: 'remaining-budget', + agentName, + }, + { + dispatchClient: { + async createDispatch() { + throw new Error('ready participants should be reused'); + }, + async deleteDispatch() {}, + }, + roomClient: { + async listRooms() { + if (!roomEnsureStarted) { + roomEnsureStarted = true; + now += 2_000; + } + return [{ name: 'voice_assistant_room_remaining_budget' }]; + }, + async createRoom({ name }) { + return { name }; + }, + async listParticipants() { + return readyParticipants(agentName); + }, + async deleteRoom() {}, + }, + waitForAgentWorkerReady: async (requestedAgentName, options) => { + workerMaxWaitMs = options.maxWaitMs; + return { + state: 'ready', + agentName: requestedAgentName, + workerId: 'AW_remaining_budget', + registeredAt: '2026-07-24T00:00:00Z', + waitedMs: 0, + }; + }, + } + ); + + assert.equal(workerMaxWaitMs, 3_000); + } finally { + Date.now = originalNow; + if (originalTotalTimeout === undefined) { + delete process.env.LIVEAVATAR_PREWARM_TOTAL_TIMEOUT_MS; + } else { + process.env.LIVEAVATAR_PREWARM_TOTAL_TIMEOUT_MS = originalTotalTimeout; + } + if (originalWorkerTimeout === undefined) { + delete process.env.LIVEAVATAR_AGENT_WORKER_READY_TIMEOUT_MS; + } else { + process.env.LIVEAVATAR_AGENT_WORKER_READY_TIMEOUT_MS = originalWorkerTimeout; + } + } +}); + +test('prewarm failures identify the active phase and completed timings', async () => { + const originalNow = Date.now; + let now = 1_000; + let roomEnsureStarted = false; + Date.now = () => now; + + try { + const error = await prewarmRoomSession( + { + roomName: 'voice_assistant_room_worker_failure', + sessionId: 'worker-failure', + agentName: 'frontdesk-browser-agent-worker-failure', + }, + { + dispatchClient: { + async createDispatch() { + throw new Error('dispatch should not start'); + }, + async deleteDispatch() {}, + }, + roomClient: { + async listRooms() { + if (!roomEnsureStarted) { + roomEnsureStarted = true; + now += 2_000; + } + return [{ name: 'voice_assistant_room_worker_failure' }]; + }, + async createRoom({ name }) { + return { name }; + }, + async listParticipants() { + return []; + }, + async deleteRoom() {}, + }, + waitForAgentWorkerReady: async () => { + now += 3_000; + throw new Error('worker marker failed'); + }, + } + ).then( + () => null, + (reason) => reason + ); + + assert.ok(error instanceof Error); + assert.equal(error.phase, 'worker_readiness'); + assert.match(error.message, /worker marker failed/); + assert.deepEqual(error.timings, { + totalPrewarmMs: 5_000, + roomEnsureMs: 2_000, + workerReadyWaitMs: 3_000, + }); + } finally { + Date.now = originalNow; + } +}); + +test('dispatch readiness failures report their phase and all elapsed timings', async () => { + const originalNow = Date.now; + let now = 1_000; + Date.now = () => now; + + try { + const error = await prewarmRoomSession( + { + roomName: 'voice_assistant_room_dispatch_failure', + sessionId: 'dispatch-failure', + agentName: 'frontdesk-browser-agent-dispatch-failure', + }, + { + dispatchClient: { + async createDispatch() { + return { id: 'dispatch-phase-failure' }; + }, + async deleteDispatch() {}, + }, + roomClient: { + async listRooms() { + return [{ name: 'voice_assistant_room_dispatch_failure' }]; + }, + async createRoom({ name }) { + return { name }; + }, + async listParticipants() { + return []; + }, + async deleteRoom() {}, + }, + waitForAgentWorkerReady: async (requestedAgentName) => ({ + state: 'skipped', + agentName: requestedAgentName, + reason: 'not_sandbox', + waitedMs: 0, + }), + dispatchTimeoutMs: 100, + dispatchPollMs: 10, + dispatchRetryMs: 10, + sleep: async (ms) => { + now += ms; + }, + } + ).then( + () => null, + (reason) => reason + ); + + assert.ok(error instanceof Error); + assert.equal(error.phase, 'dispatch_readiness'); + assert.deepEqual(error.timings, { + totalPrewarmMs: 100, + roomEnsureMs: 0, + workerReadyWaitMs: 0, + dispatchReadinessMs: 100, + }); + } finally { + Date.now = originalNow; + } +}); + +test('concurrent prewarm dispatch calls share one LiveKit dispatch', async () => { + const firstService = await import( + '../app/api/session/session-dispatch-service.ts?chunk=dispatch-a' + ); + const secondService = await import( + '../app/api/session/session-dispatch-service.ts?chunk=dispatch-b' + ); + const agentName = 'frontdesk-browser-agent-concurrent'; + let dispatchCalls = 0; + let ready = false; + let releaseDispatch; + const dispatchGate = new Promise((resolve) => { + releaseDispatch = resolve; + }); + const dispatchClient = { + async createDispatch() { + dispatchCalls += 1; + await dispatchGate; + ready = true; + return { id: 'dispatch-concurrent' }; + }, + async deleteDispatch() {}, + }; + const roomClient = { + async listParticipants() { + return ready ? readyParticipants(agentName) : []; + }, + async listRooms() { + return [{ name: 'voice_assistant_room_concurrent' }]; + }, + async createRoom() { + throw new Error('room already exists'); + }, + async deleteRoom() {}, + }; + const request = { + roomName: 'voice_assistant_room_concurrent', + sessionId: 'concurrent', + agentName, + readiness: { requireRoomInputParticipantsReady: true }, + }; + + const first = firstService.dispatchRoomSession(request, { dispatchClient, roomClient }); + const second = secondService.dispatchRoomSession(request, { dispatchClient, roomClient }); + releaseDispatch(); + const [firstResult, secondResult] = await Promise.all([first, second]); + + assert.equal(dispatchCalls, 1); + assert.deepEqual(secondResult, firstResult); + assert.equal(firstResult.dispatchId, 'dispatch-concurrent'); +}); + +test('a concurrent prewarm budget extends the shared in-flight dispatch', async () => { + const originalNow = Date.now; + let now = 1_000; + let dispatchCalls = 0; + Date.now = () => now; + + const dispatchClient = { + async createDispatch() { + dispatchCalls += 1; + return { id: 'dispatch-shared-budget' }; + }, + async deleteDispatch() {}, + }; + const roomClient = { + async listParticipants() { + return []; + }, + async deleteRoom() {}, + }; + const dependencies = { + dispatchClient, + roomClient, + dispatchPollMs: 1_000, + dispatchRetryMs: 1_000, + sleep: async (ms) => { + now += ms; + }, + }; + const request = { + roomName: 'voice_assistant_room_shared_budget', + sessionId: 'shared-budget', + agentName: 'frontdesk-browser-agent-shared-budget', + }; + + try { + const startedAt = now; + const regularDispatch = dispatchRoomSession(request, { + ...dependencies, + dispatchTimeoutMs: 8_000, + }); + const prewarmDispatch = dispatchRoomSession(request, { + ...dependencies, + dispatchTimeoutMs: 20_000, + }); + const results = await Promise.allSettled([regularDispatch, prewarmDispatch]); + + assert.equal(dispatchCalls, 1); + assert.equal(now - startedAt, 20_000); + for (const result of results) { + assert.equal(result.status, 'rejected'); + assert.match(result.reason.message, /agent dispatch failed/); + } + } finally { + Date.now = originalNow; + } +}); + +test('a prewarm budget can extend the dispatch during the old deadline check', async () => { + const originalNow = Date.now; + let now = 1_000; + let dispatchCalls = 0; + let releaseDeadlineCheck; + let markDeadlineCheckStarted; + let deadlineCheckBlocked = false; + Date.now = () => now; + + const deadlineCheckStarted = new Promise((resolve) => { + markDeadlineCheckStarted = resolve; + }); + const deadlineCheckGate = new Promise((resolve) => { + releaseDeadlineCheck = resolve; + }); + const dispatchClient = { + async createDispatch() { + dispatchCalls += 1; + return { id: 'dispatch-late-shared-budget' }; + }, + async deleteDispatch() {}, + }; + const roomClient = { + async listParticipants() { + if (now === 8_000 && !deadlineCheckBlocked) { + deadlineCheckBlocked = true; + markDeadlineCheckStarted(); + await deadlineCheckGate; + } + return []; + }, + async deleteRoom() {}, + }; + const dependencies = { + dispatchClient, + roomClient, + dispatchPollMs: 1_000, + dispatchRetryMs: 1_000, + sleep: async (ms) => { + now += ms; + }, + }; + const request = { + roomName: 'voice_assistant_room_late_shared_budget', + sessionId: 'late-shared-budget', + agentName: 'frontdesk-browser-agent-late-shared-budget', + }; + + try { + const regularDispatch = dispatchRoomSession(request, { + ...dependencies, + dispatchTimeoutMs: 8_000, + }); + await deadlineCheckStarted; + const prewarmDispatch = dispatchRoomSession(request, { + ...dependencies, + dispatchTimeoutMs: 20_000, + }); + const resultsPromise = Promise.allSettled([regularDispatch, prewarmDispatch]); + releaseDeadlineCheck(); + const results = await resultsPromise; + + assert.equal(dispatchCalls, 1); + assert.equal(now, 28_000); + assert.equal( + results.every((result) => result.status === 'rejected'), + true + ); + } finally { + releaseDeadlineCheck?.(); + Date.now = originalNow; + } +}); + +test('concurrent dispatch callers wait for their own readiness contract', async () => { + const agentName = 'frontdesk-browser-agent-readiness-contract'; + let dispatchCalls = 0; + let agentReady = false; + let videoReady = false; + let releaseDispatch; + let releaseVideo; + let markVideoWaitStarted; + const dispatchGate = new Promise((resolve) => { + releaseDispatch = resolve; + }); + const videoGate = new Promise((resolve) => { + releaseVideo = resolve; + }); + const videoWaitStarted = new Promise((resolve) => { + markVideoWaitStarted = resolve; + }); + const dispatchClient = { + async createDispatch() { + dispatchCalls += 1; + await dispatchGate; + agentReady = true; + return { id: 'dispatch-readiness-contract' }; + }, + async deleteDispatch() {}, + }; + const roomClient = { + async listParticipants() { + return agentReady ? readyParticipants(agentName, { videoReady }) : []; + }, + async listRooms() { + return [{ name: 'voice_assistant_room_readiness_contract' }]; + }, + async createRoom() { + throw new Error('room already exists'); + }, + async deleteRoom() {}, + }; + const request = { + roomName: 'voice_assistant_room_readiness_contract', + sessionId: 'readiness-contract', + agentName, + }; + + const prewarm = dispatchRoomSession( + { ...request, readiness: { requireRoomInputParticipantsReady: true } }, + { dispatchClient, roomClient, dispatchTimeoutMs: 100 } + ); + const browserDispatch = dispatchRoomSession( + { ...request, readiness: { requireRoomVideoInputReady: true } }, + { + dispatchClient, + roomClient, + dispatchTimeoutMs: 100, + dispatchPollMs: 1, + sleep: async () => { + markVideoWaitStarted(); + await videoGate; + }, + } + ); + + releaseDispatch(); + const prewarmResult = await prewarm; + await videoWaitStarted; + assert.equal(dispatchCalls, 1); + assert.equal(prewarmResult.dispatchId, 'dispatch-readiness-contract'); + + videoReady = true; + releaseVideo(); + const browserResult = await browserDispatch; + + assert.equal(dispatchCalls, 1); + assert.equal(browserResult.dispatchId, 'dispatch-readiness-contract'); +}); + +test('shared dispatch token stays active through per-caller readiness waits', async () => { + const source = await readFile( + new URL('../app/api/session/session-dispatch-service.ts', import.meta.url), + 'utf8' + ); + const dispatchSource = source.slice( + source.indexOf('export async function dispatchRoomSession'), + source.indexOf('async function waitForRequestedRoomSessionReadiness') + ); + const readinessSource = source.slice( + source.indexOf('async function waitForRequestedRoomSessionReadiness'), + source.indexOf('export async function prewarmRoomSession') + ); + + assert.match( + dispatchSource, + /inFlight\.callers === 0[\s\S]*finishRoomSessionDispatch\(inFlight\.session\)/ + ); + assert.doesNotMatch(readinessSource, /beginRoomSessionDispatch|finishRoomSessionDispatch/); +}); + +test('prewarm creates the room and waits for both room input participants', async () => { + const agentName = 'frontdesk-browser-agent-readiness'; + let roomCreated = false; + let workerReady = false; + let dispatchCreated = false; + let visionReady = false; + const dispatchClient = { + async createDispatch() { + assert.equal(workerReady, true); + dispatchCreated = true; + return { id: 'dispatch-readiness' }; + }, + async deleteDispatch() {}, + }; + const roomClient = { + async listRooms() { + return roomCreated ? [{ name: 'voice_assistant_room_readiness' }] : []; + }, + async createRoom({ name }) { + roomCreated = true; + return { name }; + }, + async listParticipants() { + if (!dispatchCreated) { + return []; + } + return readyParticipants(agentName).filter( + (participant) => visionReady || participant.identity !== 'room_video_input' + ); + }, + async deleteRoom() {}, + }; + + const result = await prewarmRoomSession( + { + roomName: 'voice_assistant_room_readiness', + sessionId: 'readiness', + agentName, + }, + { + dispatchClient, + roomClient, + waitForAgentWorkerReady: async (requestedAgentName) => { + assert.equal(roomCreated, true); + assert.equal(requestedAgentName, agentName); + workerReady = true; + return { + state: 'ready', + agentName, + workerId: 'AW_readiness', + registeredAt: '2026-07-13T00:00:00Z', + waitedMs: 7, + }; + }, + dispatchTimeoutMs: 100, + dispatchPollMs: 1, + sleep: async () => { + visionReady = true; + }, + } + ); + + assert.equal(roomCreated, true); + assert.equal(result.workerReadiness.workerId, 'AW_readiness'); + assert.equal(result.dispatch.dispatchId, 'dispatch-readiness'); + assert.deepEqual(result.readiness, { + audioParticipantReady: true, + visionParticipantReady: true, + }); +}); + +test('repeated prewarm reuses ready participants without creating another dispatch', async () => { + const agentName = 'frontdesk-browser-agent-idempotent'; + let dispatchCalls = 0; + const dispatchClient = { + async createDispatch() { + dispatchCalls += 1; + return { id: 'unexpected-dispatch' }; + }, + async deleteDispatch() {}, + }; + const roomClient = { + async listRooms() { + return [{ name: 'voice_assistant_room_idempotent' }]; + }, + async createRoom() { + throw new Error('room already exists'); + }, + async listParticipants() { + return readyParticipants(agentName); + }, + async deleteRoom() {}, + }; + const request = { + roomName: 'voice_assistant_room_idempotent', + sessionId: 'idempotent', + agentName, + }; + const dependencies = { + dispatchClient, + roomClient, + waitForAgentWorkerReady: async () => ({ state: 'not_required' }), + }; + + const first = await prewarmRoomSession(request, dependencies); + const second = await prewarmRoomSession(request, dependencies); + + assert.equal(first.dispatch.alreadyJoined, true); + assert.equal(second.dispatch.alreadyJoined, true); + assert.equal(dispatchCalls, 0); +}); + +test('sandbox worker readiness resolves to the shared workspace marker', () => { + assert.equal( + resolveAgentWorkerReadyFile({ + LIVEAVATAR_RUNTIME_MODE: 'sandbox', + LIVEAVATAR_SANDBOX_WORKSPACE_DATA_DIR: '/workspace/test-data', + }), + '/workspace/test-data/logs/sandbox/agent-worker-ready.json' + ); + assert.equal(resolveAgentWorkerReadyFile({ LIVEAVATAR_RUNTIME_MODE: 'local' }), ''); +}); + +test('worker readiness ignores stale agent markers and waits for the expected worker', async () => { + let reads = 0; + const readiness = await waitForAgentWorkerReady('frontdesk-browser-agent-current', { + readyFile: '/tmp/agent-worker-ready.json', + timeoutMs: 100, + pollMs: 1, + readFile: async () => { + reads += 1; + return JSON.stringify({ + version: 1, + agentName: + reads === 1 ? 'frontdesk-browser-agent-stale' : 'frontdesk-browser-agent-current', + workerId: reads === 1 ? 'AW_stale' : 'AW_current', + registeredAt: '2026-07-13T00:00:00Z', + }); + }, + sleep: async () => undefined, + }); + + assert.equal(reads, 2); + assert.equal(readiness.state, 'ready'); + assert.equal(readiness.workerId, 'AW_current'); +}); + +test('worker readiness respects a caller-owned maximum wait budget', async () => { + const originalNow = Date.now; + let now = 1_000; + Date.now = () => now; + + try { + await assert.rejects( + waitForAgentWorkerReady('frontdesk-browser-agent-timeout', { + readyFile: '/tmp/agent-worker-ready.json', + timeoutMs: 100, + maxWaitMs: 30, + pollMs: 10, + readFile: async () => '{}', + sleep: async (ms) => { + now += ms; + }, + }), + /agent worker did not register before prewarm timeout/ + ); + assert.equal(now, 1_030); + } finally { + Date.now = originalNow; + } +}); + +test('worker readiness with zero remaining budget does not read or sleep', async () => { + let reads = 0; + let sleeps = 0; + + await assert.rejects( + waitForAgentWorkerReady('frontdesk-browser-agent-no-budget', { + readyFile: '/tmp/agent-worker-ready.json', + timeoutMs: 100, + maxWaitMs: 0, + pollMs: 10, + readFile: async () => { + reads += 1; + return '{}'; + }, + sleep: async () => { + sleeps += 1; + }, + }), + /agent worker did not register before prewarm timeout/ + ); + + assert.equal(reads, 0); + assert.equal(sleeps, 0); +}); diff --git a/tests/session-registry.test.mjs b/tests/session-registry.test.mjs index 02bffbc1b..62ee7a27d 100644 --- a/tests/session-registry.test.mjs +++ b/tests/session-registry.test.mjs @@ -112,4 +112,20 @@ test('session registry documents its process-local deployment constraint', async assert.match(source, /process-local/); assert.match(source, /sticky routing/); + assert.match(source, /globalThis/); +}); + +test('session registry state is shared across server module instances', async () => { + const first = await import('../app/api/session/session-registry.ts?chunk=registry-a'); + const second = await import('../app/api/session/session-registry.ts?chunk=registry-b'); + const roomName = 'room-cross-chunk'; + const sessionId = 'session-cross-chunk'; + const token = first.beginRoomSessionDispatch(roomName, sessionId, 'agent-cross-chunk'); + + assert.equal(second.getRoomSessionSnapshot(roomName).sessionId, sessionId); + second.markRoomSessionStopping(roomName, sessionId); + assert.equal(first.isRoomSessionCancelled(token), true); + + first.finishRoomSessionDispatch(token); + second.markRoomSessionStopped(roomName, sessionId); }); diff --git a/tests/session-start-dispatch.test.mjs b/tests/session-start-dispatch.test.mjs index adcde2fb7..dec14e411 100644 --- a/tests/session-start-dispatch.test.mjs +++ b/tests/session-start-dispatch.test.mjs @@ -34,24 +34,28 @@ test('session dispatch route retries explicit agent dispatch after the browser j new URL('../app/api/session/dispatch/route.ts', import.meta.url), 'utf8' ); + const serviceSource = await readFile( + new URL('../app/api/session/session-dispatch-service.ts', import.meta.url), + 'utf8' + ); - assert.match(routeSource, /AgentDispatchClient/); - assert.match(routeSource, /RoomServiceClient/); - assert.match(routeSource, /AGENT_DISPATCH_TIMEOUT_MS/); - assert.match(routeSource, /AGENT_DISPATCH_RETRY_MS/); - assert.match(routeSource, /calculateDispatchRetryDelay/); - assert.match(routeSource, /findAgentParticipant/); - assert.match(routeSource, /summarizeAgentParticipant/); - assert.match(routeSource, /deleteDispatchQuietly/); - assert.match(routeSource, /dispatchClient\.createDispatch/); + assert.match(serviceSource, /AgentDispatchClient/); + assert.match(serviceSource, /RoomServiceClient/); + assert.match(serviceSource, /AGENT_DISPATCH_TIMEOUT_MS/); + assert.match(serviceSource, /AGENT_DISPATCH_RETRY_MS/); + assert.match(serviceSource, /calculateDispatchRetryDelay/); + assert.match(serviceSource, /findReusableAgentParticipant/); + assert.match(serviceSource, /summarizeAgentParticipant/); + assert.match(serviceSource, /deleteDispatchQuietly/); + assert.match(serviceSource, /dispatchClient\.createDispatch/); assert.match(routeSource, /roomName is required/); assert.match(routeSource, /agentName is required/); assert.match(routeSource, /sessionId is required/); - assert.match(routeSource, /beginRoomSessionDispatch/); - assert.match(routeSource, /registerRoomSessionDispatchId/); - assert.match(routeSource, /isRoomSessionCancelled/); - assert.match(routeSource, /markRoomSessionRunning/); - assert.match(routeSource, /finishRoomSessionDispatch/); + assert.match(serviceSource, /beginRoomSessionDispatch/); + assert.match(serviceSource, /registerRoomSessionDispatchId/); + assert.match(serviceSource, /isRoomSessionCancelled/); + assert.match(serviceSource, /markRoomSessionRunning/); + assert.match(serviceSource, /finishRoomSessionDispatch/); assert.match(routeSource, /deriveLiveKitRoomName/); assert.match(routeSource, /deriveSessionIdFromLiveKitRoomName/); assert.match(routeSource, /isValidConnectionRoomId/); @@ -63,15 +67,17 @@ test('session dispatch route retries explicit agent dispatch after the browser j }); test('session dispatch retry backs off between repeated attempts', async () => { - const routeSource = await readFile( - new URL('../app/api/session/dispatch/route.ts', import.meta.url), + const serviceSource = await readFile( + new URL('../app/api/session/session-dispatch-service.ts', import.meta.url), 'utf8' ); - assert.match(routeSource, /function calculateDispatchRetryDelay/); - assert.match(routeSource, /2 \*\* Math\.max\(0, attempts - 1\)/); - assert.match(routeSource, /Math\.min\(AGENT_DISPATCH_RETRY_MS \* multiplier/); - assert.match(routeSource, /calculateDispatchRetryDelay\(attempts/); + assert.match(serviceSource, /function calculateDispatchRetryDelay/); + assert.match(serviceSource, /2 \*\* Math\.max\(0, attempts - 1\)/); + assert.match( + serviceSource, + /Math\.min\([\s\S]*calculateDispatchRetryDelay\(attempts, retryMs\)[\s\S]*remainingDispatchTime\(getDeadline\(\)\)[\s\S]*\)/ + ); }); test('session dispatch route pins the Next.js runtime to nodejs', async () => { @@ -88,37 +94,41 @@ test('session dispatch route cleans up dispatch when the room session is cancell new URL('../app/api/session/dispatch/route.ts', import.meta.url), 'utf8' ); + const serviceSource = await readFile( + new URL('../app/api/session/session-dispatch-service.ts', import.meta.url), + 'utf8' + ); - assert.match(routeSource, /class RoomSessionCancelledError extends Error/); - assert.match(routeSource, /constructor\(session: RoomSessionToken\)/); - assert.match(routeSource, /session\.sessionId/); - assert.match(routeSource, /throwIfSessionCancelled/); + assert.match(serviceSource, /class RoomSessionCancelledError extends Error/); + assert.match(serviceSource, /constructor\(session: RoomSessionToken\)/); + assert.match(serviceSource, /session\.sessionId/); + assert.match(serviceSource, /throwIfSessionCancelled/); assert.match( - routeSource, - /await deleteDispatchQuietly\(dispatchClient,\s*dispatch\.id,\s*roomName\)/ + serviceSource, + /await deleteDispatchQuietly\(dispatchClient, dispatchId, roomName\)/ ); - assert.match(routeSource, /await deleteLiveKitRoomQuietly\(roomClient,\s*roomName\)/); + assert.match(serviceSource, /await deleteLiveKitRoomQuietly\(roomClient, roomName\)/); assert.match(routeSource, /status: 409/); }); test('session dispatch route logs successful dispatch with canonical session identity', async () => { - const routeSource = await readFile( - new URL('../app/api/session/dispatch/route.ts', import.meta.url), + const serviceSource = await readFile( + new URL('../app/api/session/session-dispatch-service.ts', import.meta.url), 'utf8' ); - assert.match(routeSource, /console\.info\('agent session dispatch completed'/); - assert.match(routeSource, /sessionId/); - assert.match(routeSource, /roomName/); + assert.match(serviceSource, /console\.info\('agent session dispatch completed'/); + assert.match(serviceSource, /sessionId/); + assert.match(serviceSource, /roomName/); }); test('session dispatch response does not expose raw agent attributes', async () => { - const routeSource = await readFile( - new URL('../app/api/session/dispatch/route.ts', import.meta.url), + const serviceSource = await readFile( + new URL('../app/api/session/session-dispatch-service.ts', import.meta.url), 'utf8' ); const summarySource = - routeSource.match( + serviceSource.match( /function summarizeAgentParticipant[\s\S]*?\n}\n\nasync function deleteDispatchQuietly/ )?.[0] ?? ''; @@ -151,14 +161,18 @@ test('session dispatch route only accepts anonymous LiveKit agent fallback after new URL('../app/api/session/dispatch/route.ts', import.meta.url), 'utf8' ); + const serviceSource = await readFile( + new URL('../app/api/session/session-dispatch-service.ts', import.meta.url), + 'utf8' + ); const readinessSource = await readFile( new URL('../lib/session-dispatch-readiness.ts', import.meta.url), 'utf8' ); - assert.match(routeSource, /findReusableAgentParticipant/); + assert.match(serviceSource, /findReusableAgentParticipant/); assert.match( - routeSource, + serviceSource, /const alreadyJoined = await findReusableAgentParticipant\(\s*roomClient,\s*roomName,\s*agentName,\s*reusableAgentOptions\s*\);/ ); assert.match(routeSource, /requireRoomVideoInputReady/); @@ -166,10 +180,7 @@ test('session dispatch route only accepts anonymous LiveKit agent fallback after assert.match(readinessSource, /type AgentParticipantMatchOptions/); assert.match(readinessSource, /type ReusableAgentParticipantOptions/); assert.match(readinessSource, /allowAnonymousLiveKitAgentFallback/); - assert.match( - routeSource, - /findAgentParticipant\(\s*roomClient,\s*roomName,\s*agentName,\s*\{\s*allowAnonymousLiveKitAgentFallback: true,?\s*\}\s*\)/ - ); + assert.match(serviceSource, /allowAnonymousLiveKitAgentFallback: true/); assert.match(readinessSource, /function isAnonymousLiveKitAgentParticipant/); assert.match(readinessSource, /ParticipantInfo_Kind\.AGENT/); assert.match(readinessSource, /identity\.startsWith\(['"]agent-['"]\)/); @@ -318,6 +329,20 @@ test('browser video input shows the camera control as enabled by default', async ); }); +test('browser source reports video failure even when audio capture also fails', async () => { + const browserSourceSource = await readFile( + new URL('../hooks/useBrowserSourceClient.ts', import.meta.url), + 'utf8' + ); + const audioFailureBranch = + browserSourceSource.match( + /if \(audioResult\.status === 'rejected'\) \{[\s\S]*?throw audioResult\.reason;\n \}/ + )?.[0] ?? ''; + + assert.match(audioFailureBranch, /if \(videoResult\.status === 'rejected'\)/); + assert.match(audioFailureBranch, /onVideoError\?\.\(videoResult\.reason as Error\)/); +}); + test('microphone device selector remains visible before media permission is granted', async () => { const trackSelectorSource = await readFile( new URL('../components/livekit/agent-control-bar/track-selector.tsx', import.meta.url), diff --git a/tests/session-stop-client.test.mjs b/tests/session-stop-client.test.mjs index 61b62ab25..34b69e661 100644 --- a/tests/session-stop-client.test.mjs +++ b/tests/session-stop-client.test.mjs @@ -90,7 +90,7 @@ test('agent session stop does not release gateway sandbox sessions by default on await requestAgentSessionStop('11111111-2222-4333-8444-555555555555'); - assert.deepEqual(calls, [{ url: '/api/session/stop', method: 'POST' }]); + assert.deepEqual(calls, [{ url: 'api/session/stop', method: 'POST' }]); } finally { globalThis.fetch = originalFetch; if (originalWindow === undefined) { @@ -124,7 +124,7 @@ test('agent session stop ignores public sandbox paths during local cleanup', asy await requestAgentSessionStop(sessionId); - assert.deepEqual(calls, [{ url: '/api/session/stop', method: 'POST' }]); + assert.deepEqual(calls, [{ url: 'api/session/stop', method: 'POST' }]); } finally { globalThis.fetch = originalFetch; if (originalWindow === undefined) { @@ -161,7 +161,7 @@ test('background agent session stop does not release gateway sandbox sessions', await Promise.resolve(); } - assert.deepEqual(calls, [{ url: '/api/session/stop', method: 'POST' }]); + assert.deepEqual(calls, [{ url: 'api/session/stop', method: 'POST' }]); } finally { globalThis.fetch = originalFetch; if (originalWindow === undefined) { @@ -191,7 +191,7 @@ test('agent session stop skips gateway release outside public sandbox paths', as await requestAgentSessionStop('11111111-2222-4333-8444-555555555555'); - assert.deepEqual(calls, [{ url: '/api/session/stop', method: 'POST' }]); + assert.deepEqual(calls, [{ url: 'api/session/stop', method: 'POST' }]); } finally { globalThis.fetch = originalFetch; if (originalWindow === undefined) {