diff --git a/app/api/session/stop/route.ts b/app/api/session/stop/route.ts index 8f7b54d26..bde8ec247 100644 --- a/app/api/session/stop/route.ts +++ b/app/api/session/stop/route.ts @@ -9,6 +9,7 @@ import { isValidConnectionRoomId, } from '@/lib/connection-room-id'; import { + executeRoomInputStopsSequentially, resolveRoomInputStopUrls as resolveConfiguredRoomInputStopUrls, resolveLiveKitHttpUrl, } from '@/lib/session-stop'; @@ -158,18 +159,14 @@ function resolveRoomInputStopUrls(): string[] { 'ROOM_VISION_INPUT_DEVICE', 'NEXT_PUBLIC_ROOM_VISION_INPUT_DEVICE' ), - roomAudioInputUrl: readStopEnv('ROOM_AUDIO_INPUT_URL'), - roomVisionInputUrl: readStopEnv('ROOM_VISION_INPUT_URL'), - roomInputUrl: readStopEnv('ROOM_INPUT_URL'), - frontdeskInputParticipantUrl: readStopEnv('FRONTDESK_INPUT_PARTICIPANT_URL'), - faceServiceUrl: readStopEnv('FACE_SERVICE_URL'), - genericCameraParticipantUrl: readStopEnv('GENERIC_CAMERA_PARTICIPANT_URL'), + videoProcessorUrl: readStopEnv('VIDEO_PROCESSOR_URL'), + edgeMediaUrl: readStopEnv('EDGE_MEDIA_URL'), }); } -function resolveLocalLiveKitServerLogPath(): string { +function resolveLocalAgentWorkerLogPath(): string { const runLogDir = process.env.LEXVOICE_RUN_LOG_DIR?.trim(); - return runLogDir ? path.join(runLogDir, 'server.log') : ''; + return runLogDir ? path.join(runLogDir, 'live.log') : ''; } function sleep(ms: number): Promise { @@ -187,7 +184,7 @@ async function fileExists(filePath: string): Promise { } } -async function readAgentWorkerStateFromServerLog( +async function readAgentWorkerStateFromLocalLog( logPath: string, agentName: string ): Promise { @@ -214,7 +211,7 @@ async function waitForLocalAgentWorkerReadiness(): Promise { return { target: 'agent_worker_readiness', ok: true, skipped: true }; } - const logPath = resolveLocalLiveKitServerLogPath(); + const logPath = resolveLocalAgentWorkerLogPath(); const agentName = readStopAgentName(); if (!logPath || !(await fileExists(logPath))) { return { target: 'agent_worker_readiness', ok: true, skipped: true }; @@ -222,7 +219,7 @@ async function waitForLocalAgentWorkerReadiness(): Promise { const deadline = Date.now() + AGENT_WORKER_READINESS_TIMEOUT_MS; while (Date.now() <= deadline) { - const state = await readAgentWorkerStateFromServerLog(logPath, agentName); + const state = await readAgentWorkerStateFromLocalLog(logPath, agentName); if (state === 'available') { return { target: 'agent_worker_readiness', ok: true }; } @@ -341,12 +338,26 @@ async function postRoomInputStop( } async function stopRoomInput(roomName: string, sessionId: string): Promise { - const stopUrls = resolveRoomInputStopUrls(); + let stopUrls: string[]; + try { + stopUrls = resolveRoomInputStopUrls(); + } catch (error) { + return [ + { + target: 'room_input_configuration', + ok: false, + fatal: true, + error: error instanceof Error ? error.message : String(error), + }, + ]; + } if (stopUrls.length === 0) { return [{ target: 'room_input', ok: true, skipped: true }]; } - return Promise.all(stopUrls.map((stopUrl) => postRoomInputStop(stopUrl, roomName, sessionId))); + return executeRoomInputStopsSequentially(stopUrls, (stopUrl) => + postRoomInputStop(stopUrl, roomName, sessionId) + ); } async function runRemoteSessionCleanup( diff --git a/lib/agent-worker-readiness.ts b/lib/agent-worker-readiness.ts index 4f022e71f..39708fa05 100644 --- a/lib/agent-worker-readiness.ts +++ b/lib/agent-worker-readiness.ts @@ -9,8 +9,19 @@ export function readAgentWorkerStateFromLog(source: string, agentName: string): const agentNamePattern = new RegExp(`"agentName"\\s*:\\s*"${escapeRegExp(agentName)}"`); const availablePattern = /"status"\s*:\s*"WS_AVAILABLE"/; const unavailablePattern = /"status"\s*:\s*"WS_FULL"/; + // The run-scoped live.log contains one local worker; these SDK capacity lines omit agentName. + const localAvailablePattern = /worker is below capacity, marking as available/; + const localUnavailablePattern = /worker is at full capacity, marking as unavailable/; for (const line of source.split(/\r?\n/)) { + if (localAvailablePattern.test(line)) { + state = 'available'; + continue; + } + if (localUnavailablePattern.test(line)) { + state = 'unavailable'; + continue; + } if (!agentNamePattern.test(line)) { continue; } diff --git a/lib/session-stop.ts b/lib/session-stop.ts index ee3a23b6d..d2a0d1898 100644 --- a/lib/session-stop.ts +++ b/lib/session-stop.ts @@ -6,17 +6,8 @@ export interface ResolveRoomInputStopUrlsOptions { inputSource?: string | null; audioInputDevice?: string | null; visionInputDevice?: string | null; - /** - * Room-input control URLs are configured as base endpoint paths. The - * normalizer intentionally strips query/hash fragments when switching - * between /start and /stop so stop calls do not inherit start-only params. - */ - roomAudioInputUrl?: string | null; - roomVisionInputUrl?: string | null; - roomInputUrl?: string | null; - frontdeskInputParticipantUrl?: string | null; - faceServiceUrl?: string | null; - genericCameraParticipantUrl?: string | null; + videoProcessorUrl?: string | null; + edgeMediaUrl?: string | null; } export function resolveLiveKitHttpUrl(liveKitUrl?: string | null): string | undefined { @@ -33,13 +24,6 @@ export function resolveLiveKitHttpUrl(liveKitUrl?: string | null): string | unde return normalized; } -function addRoomInputStopUrl(urls: Set, rawUrl?: string | null): void { - const stopUrl = normalizeRoomInputControlUrl(rawUrl || '', 'stop'); - if (stopUrl) { - urls.add(stopUrl); - } -} - export function normalizeRoomInputControlUrl( rawUrl: string, action: RoomInputControlAction @@ -77,16 +61,23 @@ export function normalizeRoomInputControlUrl( } } +export async function executeRoomInputStopsSequentially( + stopUrls: readonly string[], + stop: (stopUrl: string) => Promise +): Promise { + const results: T[] = []; + for (const stopUrl of stopUrls) { + results.push(await stop(stopUrl)); + } + return results; +} + export function resolveRoomInputStopUrls({ inputSource, audioInputDevice, visionInputDevice, - roomAudioInputUrl, - roomVisionInputUrl, - roomInputUrl, - frontdeskInputParticipantUrl, - faceServiceUrl, - genericCameraParticipantUrl, + videoProcessorUrl, + edgeMediaUrl, }: ResolveRoomInputStopUrlsOptions): string[] { const { audioInputDevice: resolvedAudioInputDevice, @@ -97,28 +88,23 @@ export function resolveRoomInputStopUrls({ visionInputDevice, }); - const urls = new Set(); - const selectedServerDevices = new Set(); - - if (usesServerRoomInputDevice(resolvedAudioInputDevice)) { - selectedServerDevices.add(resolvedAudioInputDevice); - addRoomInputStopUrl(urls, roomAudioInputUrl || roomInputUrl); - } - if (usesServerRoomInputDevice(resolvedVisionInputDevice)) { - selectedServerDevices.add(resolvedVisionInputDevice); - addRoomInputStopUrl(urls, roomVisionInputUrl || roomInputUrl); - } - if (selectedServerDevices.size === 0) { + const usesServerInput = + usesServerRoomInputDevice(resolvedAudioInputDevice) || + usesServerRoomInputDevice(resolvedVisionInputDevice); + if (!usesServerInput) { return []; } - if (selectedServerDevices.has('xunfei')) { - addRoomInputStopUrl(urls, frontdeskInputParticipantUrl); - addRoomInputStopUrl(urls, faceServiceUrl); + const videoProcessorStopUrl = normalizeRoomInputControlUrl(videoProcessorUrl || '', 'stop'); + const edgeMediaStopUrl = normalizeRoomInputControlUrl(edgeMediaUrl || '', 'stop'); + if (!videoProcessorStopUrl || !edgeMediaStopUrl) { + throw new Error('VIDEO_PROCESSOR_URL and EDGE_MEDIA_URL are required for server room input'); } - if (selectedServerDevices.has('generic')) { - addRoomInputStopUrl(urls, genericCameraParticipantUrl); + if (videoProcessorStopUrl === edgeMediaStopUrl) { + throw new Error( + 'VIDEO_PROCESSOR_URL and EDGE_MEDIA_URL must resolve to distinct stop endpoints' + ); } - return [...urls]; + return [videoProcessorStopUrl, edgeMediaStopUrl]; } diff --git a/tests/session-stop.test.mjs b/tests/session-stop.test.mjs index a6dbdf7e5..4ee4c06a5 100644 --- a/tests/session-stop.test.mjs +++ b/tests/session-stop.test.mjs @@ -1,8 +1,22 @@ import assert from 'node:assert/strict'; import { readFile } from 'node:fs/promises'; import { test } from 'node:test'; +import { POST as stopSession } from '../app/api/session/stop/route.ts'; import { readAgentWorkerStateFromLog } from '../lib/agent-worker-readiness.ts'; -import { resolveLiveKitHttpUrl, resolveRoomInputStopUrls } from '../lib/session-stop.ts'; +import { + executeRoomInputStopsSequentially, + resolveLiveKitHttpUrl, + resolveRoomInputStopUrls, +} from '../lib/session-stop.ts'; + +function restoreEnv(previousEnv) { + for (const key of Object.keys(process.env)) { + if (!(key in previousEnv)) { + delete process.env[key]; + } + } + Object.assign(process.env, previousEnv); +} test('parses the latest target agent worker state from LiveKit server logs', () => { const source = [ @@ -15,59 +29,182 @@ test('parses the latest target agent worker state from LiveKit server logs', () assert.equal(readAgentWorkerStateFromLog(source, 'missing-agent'), 'unknown'); }); +test('parses the latest local worker capacity state from the agent log', () => { + const source = [ + 'worker is at full capacity, marking as unavailable', + 'worker is below capacity, marking as available', + ].join('\n'); + + assert.equal(readAgentWorkerStateFromLog(source, 'frontdesk-agent'), 'available'); +}); + test('maps livekit websocket URLs to server API URLs', () => { assert.equal(resolveLiveKitHttpUrl('ws://localhost:7818'), 'http://localhost:7818'); assert.equal(resolveLiveKitHttpUrl('wss://livekit.example'), 'https://livekit.example'); assert.equal(resolveLiveKitHttpUrl('https://livekit.example'), 'https://livekit.example'); }); -test('room input stop URL resolver ignores primebot non-server input', () => { +test('room input stop URL resolver skips browser input', () => { assert.deepEqual( resolveRoomInputStopUrls({ - inputSource: 'primebot', - roomInputUrl: 'http://room-input.local/start', - roomAudioInputUrl: 'http://audio.local/start', - roomVisionInputUrl: 'http://vision.local/start', - frontdeskInputParticipantUrl: 'http://xunfei.local/start', - faceServiceUrl: 'http://face.local/start', - genericCameraParticipantUrl: 'http://generic.local/start', + inputSource: 'browser', + edgeMediaUrl: 'http://edge.local/start', + videoProcessorUrl: 'http://processor.local/start', }), [] ); }); -test('room input stop URL resolver only stops selected mixed server roles', () => { +test('room input stop URL resolver returns processor then edge for server input', () => { assert.deepEqual( resolveRoomInputStopUrls({ - inputSource: 'mixed', - audioInputDevice: 'xunfei', - visionInputDevice: 'browser', - roomAudioInputUrl: 'http://xunfei-audio.local/start', - roomVisionInputUrl: 'http://unused-vision.local/start', - roomInputUrl: 'http://fallback.local/start', - frontdeskInputParticipantUrl: 'http://frontdesk.local/start', - faceServiceUrl: 'http://face.local/start', - genericCameraParticipantUrl: 'http://generic.local/start', + inputSource: 'xunfei', + edgeMediaUrl: 'http://edge.local/start', + videoProcessorUrl: 'http://processor.local/start', }), - ['http://xunfei-audio.local/stop', 'http://frontdesk.local/stop', 'http://face.local/stop'] + ['http://processor.local/stop', 'http://edge.local/stop'] ); }); -test('session stop route can call the room-input control endpoint before deleting the room', async () => { +test('mixed input keeps the complete split topology when either role uses server input', () => { + for (const roleDevices of [ + { audioInputDevice: 'xunfei', visionInputDevice: 'browser' }, + { audioInputDevice: 'browser', visionInputDevice: 'generic' }, + ]) { + assert.deepEqual( + resolveRoomInputStopUrls({ + inputSource: 'mixed', + ...roleDevices, + edgeMediaUrl: 'http://edge.local/start', + videoProcessorUrl: 'http://processor.local/start', + }), + ['http://processor.local/stop', 'http://edge.local/stop'] + ); + } +}); + +test('room input stop URL resolver rejects incomplete split media configuration', () => { + for (const options of [ + {}, + { videoProcessorUrl: 'http://processor.local/start' }, + { edgeMediaUrl: 'http://edge.local/start' }, + ]) { + assert.throws( + () => resolveRoomInputStopUrls({ inputSource: 'xunfei', ...options }), + /VIDEO_PROCESSOR_URL and EDGE_MEDIA_URL are required/ + ); + } +}); + +test('room input stop URL resolver rejects duplicate split media endpoints', () => { + assert.throws( + () => + resolveRoomInputStopUrls({ + inputSource: 'xunfei', + videoProcessorUrl: 'http://media.local/start', + edgeMediaUrl: 'http://media.local/stop', + }), + /must resolve to distinct stop endpoints/ + ); +}); + +test('session stop reports invalid split media configuration and continues room cleanup', async () => { + const previousEnv = { ...process.env }; + + process.env.INPUT_SOURCE = 'xunfei'; + delete process.env.VIDEO_PROCESSOR_URL; + delete process.env.EDGE_MEDIA_URL; + delete process.env.LIVEKIT_URL; + delete process.env.LIVEKIT_API_KEY; + delete process.env.LIVEKIT_API_SECRET; + delete process.env.LEXVOICE_RUN_LOG_DIR; + + try { + const response = await stopSession( + new Request('http://localhost/api/session/stop', { + method: 'POST', + body: JSON.stringify({ + sessionId: '00000000-0000-4000-8000-000000000020', + wait: true, + }), + }) + ); + const payload = await response.json(); + + assert.equal(response.status, 502); + assert.equal(payload.status, 'partial'); + assert.deepEqual( + payload.results.find((result) => result.target === 'room_input_configuration'), + { + target: 'room_input_configuration', + ok: false, + fatal: true, + error: 'VIDEO_PROCESSOR_URL and EDGE_MEDIA_URL are required for server room input', + } + ); + assert.deepEqual( + payload.results.find((result) => result.target === 'livekit_room'), + { + target: 'livekit_room', + ok: true, + skipped: true, + } + ); + } finally { + restoreEnv(previousEnv); + } +}); + +test('room input stop executor waits for each stop before starting the next', async () => { + const processorUrl = 'http://processor.local/stop'; + const edgeUrl = 'http://edge.local/stop'; + const events = []; + let releaseProcessorStop = () => {}; + const processorStopPending = new Promise((resolve) => { + releaseProcessorStop = resolve; + }); + + const execution = executeRoomInputStopsSequentially([processorUrl, edgeUrl], async (stopUrl) => { + events.push(`start:${stopUrl}`); + if (stopUrl === processorUrl) { + await processorStopPending; + } + events.push(`finish:${stopUrl}`); + return stopUrl; + }); + + await Promise.resolve(); + assert.deepEqual(events, [`start:${processorUrl}`]); + + releaseProcessorStop(); + assert.deepEqual(await execution, [processorUrl, edgeUrl]); + assert.deepEqual(events, [ + `start:${processorUrl}`, + `finish:${processorUrl}`, + `start:${edgeUrl}`, + `finish:${edgeUrl}`, + ]); +}); + +test('session stop route stops room input before deleting the room', async () => { const routeSource = await readFile( new URL('../app/api/session/stop/route.ts', import.meta.url), 'utf8' ); const cleanupSource = routeSource.match(/async function runRemoteSessionCleanup[\s\S]*?\n}/)?.[0]; + const stopUrlResolverSource = routeSource.match( + /function resolveRoomInputStopUrls[\s\S]*?\n}/ + )?.[0]; + const stopRoomInputSource = routeSource.match(/async function stopRoomInput[\s\S]*?\n}/)?.[0]; assert.ok(cleanupSource, 'runRemoteSessionCleanup should be defined'); - assert.match(routeSource, /readStopEnv\('ROOM_INPUT_URL'\)/); - assert.match(routeSource, /resolveRoomInputStopUrls/); - assert.match(routeSource, /stopRoomInput/); - assert.match(routeSource, /FRONTDESK_INPUT_PARTICIPANT_URL/); - assert.match(routeSource, /FACE_SERVICE_URL/); - assert.match(routeSource, /GENERIC_CAMERA_PARTICIPANT_URL/); + assert.ok(stopUrlResolverSource, 'resolveRoomInputStopUrls should be defined'); + assert.match(stopUrlResolverSource, /videoProcessorUrl: readStopEnv\('VIDEO_PROCESSOR_URL'\)/); + assert.match(stopUrlResolverSource, /edgeMediaUrl: readStopEnv\('EDGE_MEDIA_URL'\)/); + assert.equal((stopUrlResolverSource.match(/readStopEnv\(/g) ?? []).length, 2); + assert.ok(stopRoomInputSource, 'stopRoomInput should be defined'); + assert.match(stopRoomInputSource, /executeRoomInputStopsSequentially\(stopUrls,/); assert.match( cleanupSource, /const roomInputResults = await stopRoomInput\(roomName, sessionId\);[\s\S]*const liveKitRoomResult = await deleteLiveKitRoom\(roomName\);/ @@ -125,7 +262,8 @@ test('session stop route waits for local agent worker readiness before finishing assert.ok(cleanupSource, 'runRemoteSessionCleanup should be defined'); assert.match(routeSource, /function waitForLocalAgentWorkerReadiness/); assert.match(routeSource, /process\.env\.LEXVOICE_RUN_LOG_DIR/); - assert.match(routeSource, /server\.log/); + assert.match(routeSource, /live\.log/); + assert.doesNotMatch(routeSource, /path\.join\(runLogDir, 'server\.log'\)/); assert.match(routeSource, /AGENT_WORKER_READINESS_TIMEOUT_MS/); assert.match(routeSource, /readFileTail\(logPath/); assert.doesNotMatch(routeSource, /readFile\(logPath,\s*'utf8'\)/);