From 733a4b42d6c6ab468eb60dbcd36928c4e3f77d3b Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 10 Aug 2026 11:57:44 +0800 Subject: [PATCH 1/5] fix: align session cleanup with split media topology --- app/api/session/stop/route.ts | 18 ++++++++++++------ lib/agent-worker-readiness.ts | 10 ++++++++++ lib/session-stop.ts | 19 +++++++++++++++++-- tests/session-stop.test.mjs | 35 ++++++++++++++++++++++++++++++++++- 4 files changed, 73 insertions(+), 9 deletions(-) diff --git a/app/api/session/stop/route.ts b/app/api/session/stop/route.ts index 8f7b54d26..9210a0384 100644 --- a/app/api/session/stop/route.ts +++ b/app/api/session/stop/route.ts @@ -158,6 +158,8 @@ function resolveRoomInputStopUrls(): string[] { 'ROOM_VISION_INPUT_DEVICE', 'NEXT_PUBLIC_ROOM_VISION_INPUT_DEVICE' ), + edgeMediaUrl: readStopEnv('EDGE_MEDIA_URL'), + videoProcessorUrl: readStopEnv('VIDEO_PROCESSOR_URL'), roomAudioInputUrl: readStopEnv('ROOM_AUDIO_INPUT_URL'), roomVisionInputUrl: readStopEnv('ROOM_VISION_INPUT_URL'), roomInputUrl: readStopEnv('ROOM_INPUT_URL'), @@ -167,9 +169,9 @@ function resolveRoomInputStopUrls(): string[] { }); } -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 +189,7 @@ async function fileExists(filePath: string): Promise { } } -async function readAgentWorkerStateFromServerLog( +async function readAgentWorkerStateFromLocalLog( logPath: string, agentName: string ): Promise { @@ -214,7 +216,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 +224,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 }; } @@ -346,7 +348,11 @@ async function stopRoomInput(roomName: string, sessionId: string): Promise postRoomInputStop(stopUrl, roomName, sessionId))); + const results: StopResult[] = []; + for (const stopUrl of stopUrls) { + results.push(await postRoomInputStop(stopUrl, roomName, sessionId)); + } + return results; } async function runRemoteSessionCleanup( diff --git a/lib/agent-worker-readiness.ts b/lib/agent-worker-readiness.ts index 4f022e71f..be003c54f 100644 --- a/lib/agent-worker-readiness.ts +++ b/lib/agent-worker-readiness.ts @@ -9,8 +9,18 @@ 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"/; + 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..96c2d535c 100644 --- a/lib/session-stop.ts +++ b/lib/session-stop.ts @@ -6,6 +6,8 @@ export interface ResolveRoomInputStopUrlsOptions { inputSource?: string | null; audioInputDevice?: string | null; visionInputDevice?: string | null; + edgeMediaUrl?: string | null; + videoProcessorUrl?: string | null; /** * Room-input control URLs are configured as base endpoint paths. The * normalizer intentionally strips query/hash fragments when switching @@ -81,6 +83,8 @@ export function resolveRoomInputStopUrls({ inputSource, audioInputDevice, visionInputDevice, + edgeMediaUrl, + videoProcessorUrl, roomAudioInputUrl, roomVisionInputUrl, roomInputUrl, @@ -102,16 +106,27 @@ export function resolveRoomInputStopUrls({ if (usesServerRoomInputDevice(resolvedAudioInputDevice)) { selectedServerDevices.add(resolvedAudioInputDevice); - addRoomInputStopUrl(urls, roomAudioInputUrl || roomInputUrl); } if (usesServerRoomInputDevice(resolvedVisionInputDevice)) { selectedServerDevices.add(resolvedVisionInputDevice); - addRoomInputStopUrl(urls, roomVisionInputUrl || roomInputUrl); } if (selectedServerDevices.size === 0) { return []; } + if (edgeMediaUrl || videoProcessorUrl) { + addRoomInputStopUrl(urls, videoProcessorUrl); + addRoomInputStopUrl(urls, edgeMediaUrl); + return [...urls]; + } + + if (usesServerRoomInputDevice(resolvedAudioInputDevice)) { + addRoomInputStopUrl(urls, roomAudioInputUrl || roomInputUrl); + } + if (usesServerRoomInputDevice(resolvedVisionInputDevice)) { + addRoomInputStopUrl(urls, roomVisionInputUrl || roomInputUrl); + } + if (selectedServerDevices.has('xunfei')) { addRoomInputStopUrl(urls, frontdeskInputParticipantUrl); addRoomInputStopUrl(urls, faceServiceUrl); diff --git a/tests/session-stop.test.mjs b/tests/session-stop.test.mjs index a6dbdf7e5..f20458171 100644 --- a/tests/session-stop.test.mjs +++ b/tests/session-stop.test.mjs @@ -15,6 +15,15 @@ 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'); @@ -53,6 +62,22 @@ test('room input stop URL resolver only stops selected mixed server roles', () = ); }); +test('room input stop URL resolver prefers the split topology in dependency order', () => { + assert.deepEqual( + resolveRoomInputStopUrls({ + inputSource: 'xunfei', + edgeMediaUrl: 'http://edge.local/start', + videoProcessorUrl: 'http://processor.local/start', + roomAudioInputUrl: 'http://legacy-audio.local/start', + roomVisionInputUrl: 'http://legacy-vision.local/start', + roomInputUrl: 'http://legacy-room-input.local/start', + frontdeskInputParticipantUrl: 'http://legacy-frontdesk.local/start', + faceServiceUrl: 'http://legacy-face.local/start', + }), + ['http://processor.local/stop', 'http://edge.local/stop'] + ); +}); + test('session stop route can call the room-input control endpoint before deleting the room', async () => { const routeSource = await readFile( new URL('../app/api/session/stop/route.ts', import.meta.url), @@ -63,6 +88,8 @@ test('session stop route can call the room-input control endpoint before deletin assert.ok(cleanupSource, 'runRemoteSessionCleanup should be defined'); assert.match(routeSource, /readStopEnv\('ROOM_INPUT_URL'\)/); + assert.match(routeSource, /readStopEnv\('EDGE_MEDIA_URL'\)/); + assert.match(routeSource, /readStopEnv\('VIDEO_PROCESSOR_URL'\)/); assert.match(routeSource, /resolveRoomInputStopUrls/); assert.match(routeSource, /stopRoomInput/); assert.match(routeSource, /FRONTDESK_INPUT_PARTICIPANT_URL/); @@ -72,6 +99,11 @@ test('session stop route can call the room-input control endpoint before deletin cleanupSource, /const roomInputResults = await stopRoomInput\(roomName, sessionId\);[\s\S]*const liveKitRoomResult = await deleteLiveKitRoom\(roomName\);/ ); + assert.match( + routeSource, + /for \(const stopUrl of stopUrls\) \{[\s\S]*await postRoomInputStop\(stopUrl, roomName, sessionId\)/ + ); + assert.doesNotMatch(routeSource, /Promise\.all\(stopUrls\.map\(\(stopUrl\) => postRoomInputStop/); }); test('session stop route cancels room session before remote cleanup', async () => { @@ -125,7 +157,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'\)/); From 87969304130b70ae9622b4baf84b6ffcd817b223 Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 10 Aug 2026 14:34:24 +0800 Subject: [PATCH 2/5] fix: remove legacy room-input stop fallback --- app/api/session/stop/route.ts | 8 +---- lib/session-stop.ts | 58 ++++++----------------------------- tests/session-stop.test.mjs | 52 +++++++++---------------------- 3 files changed, 25 insertions(+), 93 deletions(-) diff --git a/app/api/session/stop/route.ts b/app/api/session/stop/route.ts index 9210a0384..c4047a8fe 100644 --- a/app/api/session/stop/route.ts +++ b/app/api/session/stop/route.ts @@ -158,14 +158,8 @@ function resolveRoomInputStopUrls(): string[] { 'ROOM_VISION_INPUT_DEVICE', 'NEXT_PUBLIC_ROOM_VISION_INPUT_DEVICE' ), - edgeMediaUrl: readStopEnv('EDGE_MEDIA_URL'), videoProcessorUrl: readStopEnv('VIDEO_PROCESSOR_URL'), - 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'), + edgeMediaUrl: readStopEnv('EDGE_MEDIA_URL'), }); } diff --git a/lib/session-stop.ts b/lib/session-stop.ts index 96c2d535c..c0012fb6d 100644 --- a/lib/session-stop.ts +++ b/lib/session-stop.ts @@ -6,19 +6,8 @@ export interface ResolveRoomInputStopUrlsOptions { inputSource?: string | null; audioInputDevice?: string | null; visionInputDevice?: string | null; - edgeMediaUrl?: string | null; videoProcessorUrl?: 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; + edgeMediaUrl?: string | null; } export function resolveLiveKitHttpUrl(liveKitUrl?: string | null): string | undefined { @@ -83,14 +72,8 @@ export function resolveRoomInputStopUrls({ inputSource, audioInputDevice, visionInputDevice, - edgeMediaUrl, videoProcessorUrl, - roomAudioInputUrl, - roomVisionInputUrl, - roomInputUrl, - frontdeskInputParticipantUrl, - faceServiceUrl, - genericCameraParticipantUrl, + edgeMediaUrl, }: ResolveRoomInputStopUrlsOptions): string[] { const { audioInputDevice: resolvedAudioInputDevice, @@ -101,39 +84,16 @@ export function resolveRoomInputStopUrls({ visionInputDevice, }); - const urls = new Set(); - const selectedServerDevices = new Set(); - - if (usesServerRoomInputDevice(resolvedAudioInputDevice)) { - selectedServerDevices.add(resolvedAudioInputDevice); - } - if (usesServerRoomInputDevice(resolvedVisionInputDevice)) { - selectedServerDevices.add(resolvedVisionInputDevice); - } - if (selectedServerDevices.size === 0) { + const usesServerInput = + usesServerRoomInputDevice(resolvedAudioInputDevice) || + usesServerRoomInputDevice(resolvedVisionInputDevice); + if (!usesServerInput) { return []; } - if (edgeMediaUrl || videoProcessorUrl) { - addRoomInputStopUrl(urls, videoProcessorUrl); - addRoomInputStopUrl(urls, edgeMediaUrl); - return [...urls]; - } - - if (usesServerRoomInputDevice(resolvedAudioInputDevice)) { - addRoomInputStopUrl(urls, roomAudioInputUrl || roomInputUrl); - } - if (usesServerRoomInputDevice(resolvedVisionInputDevice)) { - addRoomInputStopUrl(urls, roomVisionInputUrl || roomInputUrl); - } - - if (selectedServerDevices.has('xunfei')) { - addRoomInputStopUrl(urls, frontdeskInputParticipantUrl); - addRoomInputStopUrl(urls, faceServiceUrl); - } - if (selectedServerDevices.has('generic')) { - addRoomInputStopUrl(urls, genericCameraParticipantUrl); - } + const urls = new Set(); + addRoomInputStopUrl(urls, videoProcessorUrl); + addRoomInputStopUrl(urls, edgeMediaUrl); return [...urls]; } diff --git a/tests/session-stop.test.mjs b/tests/session-stop.test.mjs index f20458171..1751ac337 100644 --- a/tests/session-stop.test.mjs +++ b/tests/session-stop.test.mjs @@ -30,55 +30,44 @@ test('maps livekit websocket URLs to server API URLs', () => { 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 ignores unsupported legacy fallback options', () => { 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', + roomAudioInputUrl: 'http://legacy-audio.local/start', + roomVisionInputUrl: 'http://legacy-vision.local/start', + roomInputUrl: 'http://legacy-room-input.local/start', + frontdeskInputParticipantUrl: 'http://legacy-frontdesk.local/start', + faceServiceUrl: 'http://legacy-face.local/start', + genericCameraParticipantUrl: 'http://legacy-generic.local/start', }), - ['http://xunfei-audio.local/stop', 'http://frontdesk.local/stop', 'http://face.local/stop'] + [] ); }); -test('room input stop URL resolver prefers the split topology in dependency order', () => { +test('room input stop URL resolver returns processor then edge for server input', () => { assert.deepEqual( resolveRoomInputStopUrls({ inputSource: 'xunfei', edgeMediaUrl: 'http://edge.local/start', videoProcessorUrl: 'http://processor.local/start', - roomAudioInputUrl: 'http://legacy-audio.local/start', - roomVisionInputUrl: 'http://legacy-vision.local/start', - roomInputUrl: 'http://legacy-room-input.local/start', - frontdeskInputParticipantUrl: 'http://legacy-frontdesk.local/start', - faceServiceUrl: 'http://legacy-face.local/start', }), ['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('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' @@ -87,23 +76,12 @@ test('session stop route can call the room-input control endpoint before deletin const cleanupSource = routeSource.match(/async function runRemoteSessionCleanup[\s\S]*?\n}/)?.[0]; assert.ok(cleanupSource, 'runRemoteSessionCleanup should be defined'); - assert.match(routeSource, /readStopEnv\('ROOM_INPUT_URL'\)/); - assert.match(routeSource, /readStopEnv\('EDGE_MEDIA_URL'\)/); - assert.match(routeSource, /readStopEnv\('VIDEO_PROCESSOR_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.match( cleanupSource, /const roomInputResults = await stopRoomInput\(roomName, sessionId\);[\s\S]*const liveKitRoomResult = await deleteLiveKitRoom\(roomName\);/ ); - assert.match( - routeSource, - /for \(const stopUrl of stopUrls\) \{[\s\S]*await postRoomInputStop\(stopUrl, roomName, sessionId\)/ - ); - assert.doesNotMatch(routeSource, /Promise\.all\(stopUrls\.map\(\(stopUrl\) => postRoomInputStop/); }); test('session stop route cancels room session before remote cleanup', async () => { From e240b97b85ad90d329b22288f6298b8cf50ab0d0 Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 10 Aug 2026 14:50:00 +0800 Subject: [PATCH 3/5] test: cover sequential room-input stop execution --- app/api/session/stop/route.ts | 9 +++-- lib/session-stop.ts | 11 ++++++ tests/session-stop.test.mjs | 64 +++++++++++++++++++++++++---------- 3 files changed, 61 insertions(+), 23 deletions(-) diff --git a/app/api/session/stop/route.ts b/app/api/session/stop/route.ts index c4047a8fe..01f84066d 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'; @@ -342,11 +343,9 @@ async function stopRoomInput(roomName: string, sessionId: string): Promise + postRoomInputStop(stopUrl, roomName, sessionId) + ); } async function runRemoteSessionCleanup( diff --git a/lib/session-stop.ts b/lib/session-stop.ts index c0012fb6d..fcd42e892 100644 --- a/lib/session-stop.ts +++ b/lib/session-stop.ts @@ -68,6 +68,17 @@ 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, diff --git a/tests/session-stop.test.mjs b/tests/session-stop.test.mjs index 1751ac337..ddd434ae5 100644 --- a/tests/session-stop.test.mjs +++ b/tests/session-stop.test.mjs @@ -2,7 +2,11 @@ import assert from 'node:assert/strict'; import { readFile } from 'node:fs/promises'; import { test } from 'node:test'; 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'; test('parses the latest target agent worker state from LiveKit server logs', () => { const source = [ @@ -41,21 +45,6 @@ test('room input stop URL resolver skips browser input', () => { ); }); -test('room input stop URL resolver ignores unsupported legacy fallback options', () => { - assert.deepEqual( - resolveRoomInputStopUrls({ - inputSource: 'xunfei', - roomAudioInputUrl: 'http://legacy-audio.local/start', - roomVisionInputUrl: 'http://legacy-vision.local/start', - roomInputUrl: 'http://legacy-room-input.local/start', - frontdeskInputParticipantUrl: 'http://legacy-frontdesk.local/start', - faceServiceUrl: 'http://legacy-face.local/start', - genericCameraParticipantUrl: 'http://legacy-generic.local/start', - }), - [] - ); -}); - test('room input stop URL resolver returns processor then edge for server input', () => { assert.deepEqual( resolveRoomInputStopUrls({ @@ -67,6 +56,37 @@ test('room input stop URL resolver returns processor then edge for server input' ); }); +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), @@ -74,10 +94,18 @@ test('session stop route stops room input before deleting the room', async () => ); 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, /resolveRoomInputStopUrls/); - assert.match(routeSource, /stopRoomInput/); + 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\);/ From 805492ec2c3345de79ed50d03d081a4596168b80 Mon Sep 17 00:00:00 2001 From: Codex Date: Thu, 13 Aug 2026 13:32:21 +0800 Subject: [PATCH 4/5] fix: reject incomplete split media cleanup --- app/api/session/stop/route.ts | 14 +++++- lib/session-stop.ts | 22 +++++----- tests/session-stop.test.mjs | 82 +++++++++++++++++++++++++++++++++++ 3 files changed, 106 insertions(+), 12 deletions(-) diff --git a/app/api/session/stop/route.ts b/app/api/session/stop/route.ts index 01f84066d..bde8ec247 100644 --- a/app/api/session/stop/route.ts +++ b/app/api/session/stop/route.ts @@ -338,7 +338,19 @@ 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 }]; } diff --git a/lib/session-stop.ts b/lib/session-stop.ts index fcd42e892..d2a0d1898 100644 --- a/lib/session-stop.ts +++ b/lib/session-stop.ts @@ -24,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 @@ -102,9 +95,16 @@ export function resolveRoomInputStopUrls({ return []; } - const urls = new Set(); - addRoomInputStopUrl(urls, videoProcessorUrl); - addRoomInputStopUrl(urls, edgeMediaUrl); + 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 (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 ddd434ae5..2a8a4c574 100644 --- a/tests/session-stop.test.mjs +++ b/tests/session-stop.test.mjs @@ -1,6 +1,7 @@ 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 { executeRoomInputStopsSequentially, @@ -8,6 +9,15 @@ import { 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 = [ '{"agentName":"other-agent","status":"WS_AVAILABLE"}', @@ -56,6 +66,78 @@ test('room input stop URL resolver returns processor then edge for server input' ); }); +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'; From fb193312a9bc4b19ac278deee134ea6a10a8280a Mon Sep 17 00:00:00 2001 From: Codex Date: Thu, 13 Aug 2026 13:36:52 +0800 Subject: [PATCH 5/5] test: lock mixed split cleanup contract --- lib/agent-worker-readiness.ts | 1 + tests/session-stop.test.mjs | 17 +++++++++++++++++ 2 files changed, 18 insertions(+) diff --git a/lib/agent-worker-readiness.ts b/lib/agent-worker-readiness.ts index be003c54f..39708fa05 100644 --- a/lib/agent-worker-readiness.ts +++ b/lib/agent-worker-readiness.ts @@ -9,6 +9,7 @@ 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/; diff --git a/tests/session-stop.test.mjs b/tests/session-stop.test.mjs index 2a8a4c574..4ee4c06a5 100644 --- a/tests/session-stop.test.mjs +++ b/tests/session-stop.test.mjs @@ -66,6 +66,23 @@ test('room input stop URL resolver returns processor then edge for server input' ); }); +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 [ {},