From 014235f02b3918e62679e4b74925d614cb648f54 Mon Sep 17 00:00:00 2001 From: why-tomato Date: Tue, 25 Aug 2026 20:14:29 +0800 Subject: [PATCH 1/2] feat: orchestrate Generic endpoints through heartbeat leases --- README.md | 26 +- app/api/connection-details/route.ts | 7 +- app/api/endpoint/connectivity/route.ts | 75 +++ app/api/session/dispatch/route.ts | 15 +- app/api/session/generic-edge-media-pairing.ts | 275 +++++++++++ .../session/generic-failed-start-cleanup.ts | 12 + app/api/session/generic-session-dispatch.ts | 123 +++++ app/api/session/session-dispatch-service.ts | 50 +- app/api/session/stop/route.ts | 58 ++- components/app/session-provider.tsx | 17 + .../livekit/agent-control-bar/chat-input.tsx | 13 +- lib/browser-room-session.ts | 23 +- lib/browser-runtime-compat.ts | 45 ++ lib/chat-send.ts | 27 + lib/endpoint-connectivity.ts | 82 +++ lib/generic-endpoint-lease.ts | 383 ++++++++++++++ lib/session-dispatch-readiness.ts | 33 ++ tests/browser-room-session.test.mjs | 68 +++ tests/browser-runtime-compat.test.mjs | 67 +++ tests/chat-send.test.mjs | 53 ++ tests/connection-details.test.mjs | 52 ++ tests/endpoint-connectivity.test.mjs | 134 +++++ tests/generic-edge-media.test.mjs | 466 ++++++++++++++++++ tests/generic-endpoint-lease.test.mjs | 330 +++++++++++++ tests/project-config.test.mjs | 29 ++ tests/session-dispatch-readiness.test.mjs | 163 +++++- tests/session-start-dispatch.test.mjs | 106 ++++ tests/session-stop.test.mjs | 135 ++++- 28 files changed, 2838 insertions(+), 29 deletions(-) create mode 100644 app/api/endpoint/connectivity/route.ts create mode 100644 app/api/session/generic-edge-media-pairing.ts create mode 100644 app/api/session/generic-failed-start-cleanup.ts create mode 100644 app/api/session/generic-session-dispatch.ts create mode 100644 lib/browser-runtime-compat.ts create mode 100644 lib/chat-send.ts create mode 100644 lib/endpoint-connectivity.ts create mode 100644 lib/generic-endpoint-lease.ts create mode 100644 tests/browser-runtime-compat.test.mjs create mode 100644 tests/chat-send.test.mjs create mode 100644 tests/endpoint-connectivity.test.mjs create mode 100644 tests/generic-edge-media.test.mjs create mode 100644 tests/generic-endpoint-lease.test.mjs diff --git a/README.md b/README.md index 54ebe2f90..0eee36cd2 100644 --- a/README.md +++ b/README.md @@ -68,6 +68,30 @@ If you replace the custom connection details endpoint, it must echo the requeste `sessionId` and derive the same room name so dispatch and stop calls coordinate with the connected room. +Integrated Generic configuration belongs to the LexVoice unified Mac startup authority. +Do not add integrated Generic settings to this repository's `.env.example` or `.env.local`. +The LexVoice lifecycle injects the server-only settings into this Next.js process. + +The endpoint POSTs its fixed device ID, stable instance UUID, hostname, and +route-discovered private IPv4 to +`/api/endpoint/connectivity` every 10 seconds using the +`X-Endpoint-Connectivity-Token` header. The Next.js owner writes per-instance +mode-0600 records under a mode-0700 directory and expires them after 45 seconds. +One current instance may change address; zero, multiple, expired, malformed, or +out-of-CIDR leases fail closed. The registry is shared by same-host workers and +survives a Next.js restart; multi-host Next.js deployment is explicitly unsupported. +Generic has no static `EDGE_MEDIA_URL` authority: only server code may resolve +the current lease and it fixes the control target scheme, port, and paths. + +For a Generic Start Call, the Agent joins first. The server then resolves one +active lease, reclaims stale endpoint state, and sends one authenticated start +request with a 15-minute `room_audio_input` token. Dispatch succeeds only after +the exact unmuted tracks are present: `room_audio_input/room_audio`, +`room_audio_input/room_video_raw`, and `room_video_input/room_video`. Stop Call +resolves the current lease, stops the processor and endpoint, cancels dispatch +work, and deletes the Room. The browser cannot supply a device target or either +server token. + ### LiveAvatar Gateway Deployments Sandbox-backed public deployments are owned by the LexVoice repository. Set @@ -83,7 +107,7 @@ server directly: ```bash pnpm install -pnpm dev +pnpm exec next dev --hostname 0.0.0.0 --port 3000 ``` And open http://localhost:3000 in your browser. diff --git a/app/api/connection-details/route.ts b/app/api/connection-details/route.ts index 93f8e0aca..45671e011 100644 --- a/app/api/connection-details/route.ts +++ b/app/api/connection-details/route.ts @@ -16,14 +16,15 @@ type ConnectionDetails = { const API_KEY = process.env.LIVEKIT_API_KEY; const API_SECRET = process.env.LIVEKIT_API_SECRET; const LIVEKIT_URL = process.env.LIVEKIT_URL; +const LIVEKIT_BROWSER_URL = process.env.LIVEKIT_BROWSER_URL?.trim() || LIVEKIT_URL?.trim(); // don't cache the results export const revalidate = 0; export async function POST(req: Request) { try { - if (LIVEKIT_URL === undefined) { - throw new Error('LIVEKIT_URL is not defined'); + if (LIVEKIT_BROWSER_URL === undefined) { + throw new Error('LIVEKIT_BROWSER_URL and LIVEKIT_URL are not defined'); } if (API_KEY === undefined) { throw new Error('LIVEKIT_API_KEY is not defined'); @@ -53,7 +54,7 @@ export async function POST(req: Request) { // Return connection details const data: ConnectionDetails = { - serverUrl: LIVEKIT_URL, + serverUrl: LIVEKIT_BROWSER_URL, sessionId, roomName, participantToken: participantToken, diff --git a/app/api/endpoint/connectivity/route.ts b/app/api/endpoint/connectivity/route.ts new file mode 100644 index 000000000..2b020422f --- /dev/null +++ b/app/api/endpoint/connectivity/route.ts @@ -0,0 +1,75 @@ +import { NextResponse } from 'next/server'; +import { + parseEndpointConnectivityPayload, + readConnectivityToken, + secretsMatch, +} from '@/lib/endpoint-connectivity'; +import { + EndpointLeaseConflictError, + loadGenericEndpointLeaseConfig, + renewGenericEndpointLease, +} from '@/lib/generic-endpoint-lease'; + +export const runtime = 'nodejs'; +export const revalidate = 0; + +const NO_STORE_HEADERS = { 'Cache-Control': 'no-store' }; + +export async function POST(request: Request) { + const expectedToken = (process.env.ENDPOINT_CONNECTIVITY_TOKEN || '').trim(); + const actualToken = readConnectivityToken(request); + if (!expectedToken) { + return NextResponse.json( + { status: 'error', error: 'endpoint connectivity probe is not configured' }, + { status: 503, headers: NO_STORE_HEADERS } + ); + } + if (!actualToken || !secretsMatch(actualToken, expectedToken)) { + return NextResponse.json( + { status: 'error', error: 'unauthorized' }, + { status: 401, headers: NO_STORE_HEADERS } + ); + } + + let input: unknown; + try { + input = await request.json(); + } catch { + return NextResponse.json( + { status: 'error', error: 'valid JSON body is required' }, + { status: 400, headers: NO_STORE_HEADERS } + ); + } + + const parsed = parseEndpointConnectivityPayload(input); + if (!parsed.ok) { + return NextResponse.json( + { status: 'error', error: parsed.error }, + { status: 400, headers: NO_STORE_HEADERS } + ); + } + + try { + const lease = await renewGenericEndpointLease(parsed.payload, loadGenericEndpointLeaseConfig()); + console.info('Generic endpoint heartbeat accepted', { + deviceId: lease.deviceId, + instanceId: lease.instanceId, + address: lease.address, + receivedAt: lease.receivedAt, + expiresAt: lease.expiresAt, + }); + return NextResponse.json( + { status: 'leased', ...lease, hostname: parsed.payload.hostname }, + { headers: NO_STORE_HEADERS } + ); + } catch (error) { + const conflict = error instanceof EndpointLeaseConflictError; + return NextResponse.json( + { + status: 'error', + error: conflict ? 'active endpoint instance conflict' : 'endpoint heartbeat rejected', + }, + { status: conflict ? 409 : 400, headers: NO_STORE_HEADERS } + ); + } +} diff --git a/app/api/session/dispatch/route.ts b/app/api/session/dispatch/route.ts index 982817d87..00c2f950e 100644 --- a/app/api/session/dispatch/route.ts +++ b/app/api/session/dispatch/route.ts @@ -1,8 +1,9 @@ import { NextResponse } from 'next/server'; import { RoomSessionCancelledError, - dispatchRoomSession, -} from '@/app/api/session/session-dispatch-service'; + formatSessionDispatchError, + runSessionDispatch, +} from '@/app/api/session/generic-session-dispatch'; import { deriveLiveKitRoomName, deriveSessionIdFromLiveKitRoomName, @@ -52,14 +53,12 @@ export async function POST(req: Request) { } try { - const dispatch = await dispatchRoomSession({ + const dispatch = await runSessionDispatch({ roomName, sessionId, agentName, - readiness: { - requireRoomVideoInputReady: - body.requireRoomVideoInputReady === true || body.require_room_video_input_ready === true, - }, + requireRoomVideoInputReady: + body.requireRoomVideoInputReady === true || body.require_room_video_input_ready === true, }); return NextResponse.json({ status: 'dispatched', roomName, agentName, sessionId, dispatch }); } catch (error) { @@ -75,7 +74,7 @@ export async function POST(req: Request) { roomName, agentName, sessionId, - error: error instanceof Error ? error.message : String(error), + error: formatSessionDispatchError(error), }, { status: 502 } ); diff --git a/app/api/session/generic-edge-media-pairing.ts b/app/api/session/generic-edge-media-pairing.ts new file mode 100644 index 000000000..b2d718092 --- /dev/null +++ b/app/api/session/generic-edge-media-pairing.ts @@ -0,0 +1,275 @@ +import { AccessToken, type VideoGrant } from 'livekit-server-sdk'; +import { + buildGenericEdgeControlUrl, + loadGenericEndpointLeaseConfig, + resolveActiveGenericEndpointLease, +} from '@/lib/generic-endpoint-lease'; + +export type GenericEdgeTargetSnapshot = Readonly<{ + startUrl: string; + stopUrl: string; + controlToken: string; + deviceId: string; + address: string; +}>; + +export type GenericPairingRequest = { + roomUrl: string; + roomName: string; + sessionId: string; + controlSenderIdentity: string; +}; + +type ControlAction = 'start' | 'stop'; + +type Environment = Record; + +type TargetResolutionDependencies = { + environment?: Environment; + loadLeaseConfig?: typeof loadGenericEndpointLeaseConfig; + resolveLease?: typeof resolveActiveGenericEndpointLease; +}; + +type ControlRequestDependencies = { + fetchImpl?: typeof fetch; + timeoutMs?: number; +}; + +type GenericSessionCoordinatorRequest = Omit & { + agentName: string; +}; + +type GenericSessionCoordinatorDependencies = { + dispatchAgent: () => Promise<{ agentParticipant?: { identity?: string } }>; + resolveTarget: () => Promise; + pairEndpoint: ( + request: GenericPairingRequest, + target: GenericEdgeTargetSnapshot + ) => Promise<{ deviceId: string; address: string }>; + cleanupSession?: () => Promise; +}; + +type GenericStopDependencies = { + resolveTarget: () => Promise; + requestControl: ( + action: ControlAction, + payload: Record, + target: GenericEdgeTargetSnapshot + ) => Promise; +}; + +type GenericPairingDependencies = { + config: GenericEdgeTargetSnapshot; + createRoomToken: () => Promise; + requestControl: ( + action: ControlAction, + payload: Record, + config: GenericEdgeTargetSnapshot + ) => Promise; + waitForReadiness: () => Promise; + isCancelled?: () => boolean; +}; + +export async function pairGenericEdgeMedia( + request: GenericPairingRequest, + dependencies: GenericPairingDependencies +) { + const target = Object.freeze({ ...dependencies.config }); + const stopPayload = { + room_name: request.roomName, + session_id: request.sessionId, + }; + await dependencies.requestControl('stop', stopPayload, target); + + const roomToken = await dependencies.createRoomToken(); + const startPayload = { + room_url: request.roomUrl, + room_token: roomToken, + room_name: request.roomName, + session_id: request.sessionId, + service_instance_id: target.deviceId, + source_type: 'generic', + control_sender_identity: request.controlSenderIdentity, + participant_identity: 'room_audio_input', + track_names: { audio: 'room_audio', video: 'room_video_raw' }, + }; + + let startAttempted = false; + try { + startAttempted = true; + await dependencies.requestControl('start', startPayload, target); + await dependencies.waitForReadiness(); + if (dependencies.isCancelled?.()) { + throw new Error('Generic endpoint pairing was cancelled'); + } + return { deviceId: target.deviceId, address: target.address }; + } catch (error) { + if (startAttempted) { + try { + await dependencies.requestControl('stop', stopPayload, target); + } catch { + throw new Error('Generic endpoint pairing failed and rollback could not be confirmed', { + cause: error, + }); + } + } + throw error; + } +} + +export function isGenericEndpointPairingEnabled(environment: Environment = process.env): boolean { + const inputSource = readEnvironmentValue(environment, [ + 'INPUT_SOURCE', + 'NEXT_PUBLIC_INPUT_SOURCE', + 'NEXT_PUBLIC_LEXVOICE_DEVICE', + ]).toLowerCase(); + if (inputSource === 'generic') { + return true; + } + if (inputSource !== 'mixed') { + return false; + } + return ['ROOM_AUDIO_INPUT_DEVICE', 'ROOM_VISION_INPUT_DEVICE'].some( + (name) => + readEnvironmentValue(environment, [name, `NEXT_PUBLIC_${name}`]).toLowerCase() === 'generic' + ); +} + +export async function resolveGenericEdgeTargetSnapshot( + dependencies: TargetResolutionDependencies = {} +): Promise { + const environment = dependencies.environment ?? process.env; + const loadLeaseConfig = dependencies.loadLeaseConfig ?? loadGenericEndpointLeaseConfig; + const resolveLease = dependencies.resolveLease ?? resolveActiveGenericEndpointLease; + const lease = await resolveLease(loadLeaseConfig(environment)); + const controlToken = (environment.EDGE_MEDIA_CONTROL_TOKEN ?? '').trim(); + if (!controlToken) { + throw new Error('EDGE_MEDIA_CONTROL_TOKEN is required for Generic endpoint control'); + } + return Object.freeze({ + startUrl: buildGenericEdgeControlUrl(lease, 'start'), + stopUrl: buildGenericEdgeControlUrl(lease, 'stop'), + controlToken, + deviceId: lease.deviceId, + address: lease.address, + }); +} + +export async function requestGenericEdgeControl( + action: ControlAction, + payload: Record, + target: GenericEdgeTargetSnapshot, + dependencies: ControlRequestDependencies = {} +): Promise { + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), dependencies.timeoutMs ?? 3_000); + try { + const response = await (dependencies.fetchImpl ?? fetch)( + buildGenericEdgeControlUrl(target, action), + { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'X-Lexvoice-Control-Token': target.controlToken, + }, + body: JSON.stringify(payload), + redirect: 'manual', + signal: controller.signal, + } + ); + if (!response.ok) { + throw new Error(`Generic endpoint ${action} returned HTTP ${response.status}`); + } + } catch (error) { + if ( + error instanceof Error && + /^Generic endpoint (start|stop) returned HTTP/.test(error.message) + ) { + throw error; + } + throw new Error(`Generic endpoint ${action} request failed`); + } finally { + clearTimeout(timeout); + } +} + +export async function createGenericRoomInputToken( + roomName: string, + environment: Environment = process.env +): Promise { + const apiKey = (environment.LIVEKIT_API_KEY ?? '').trim(); + const apiSecret = (environment.LIVEKIT_API_SECRET ?? '').trim(); + if (!apiKey || !apiSecret) { + throw new Error('LiveKit API configuration is required for Generic endpoint pairing'); + } + const token = new AccessToken(apiKey, apiSecret, { + identity: 'room_audio_input', + name: 'Generic Edge Media', + ttl: '15m', + }); + const grant: VideoGrant = { + room: roomName, + roomJoin: true, + canPublish: true, + canPublishData: true, + canSubscribe: true, + }; + token.addGrant(grant); + return token.toJwt(); +} + +export async function coordinateGenericRoomSession( + request: GenericSessionCoordinatorRequest, + dependencies: GenericSessionCoordinatorDependencies +) { + const dispatch = await dependencies.dispatchAgent(); + try { + const controlSenderIdentity = dispatch.agentParticipant?.identity?.trim(); + if (!controlSenderIdentity) { + throw new Error('Generic session Agent participant is unavailable'); + } + const target = await dependencies.resolveTarget(); + const edge = await dependencies.pairEndpoint( + { + roomUrl: request.roomUrl, + roomName: request.roomName, + sessionId: request.sessionId, + controlSenderIdentity, + }, + target + ); + return { dispatch, edge }; + } catch (error) { + try { + await dependencies.cleanupSession?.(); + } catch { + throw new Error('Generic session startup failed and cloud cleanup could not be confirmed', { + cause: error, + }); + } + throw error; + } +} + +export async function stopGenericEdgeMedia( + request: { roomName: string; sessionId: string }, + dependencies: GenericStopDependencies +) { + const target = await dependencies.resolveTarget(); + await dependencies.requestControl( + 'stop', + { room_name: request.roomName, session_id: request.sessionId }, + target + ); + return { deviceId: target.deviceId, address: target.address }; +} + +function readEnvironmentValue(environment: Environment, names: string[]): string { + for (const name of names) { + const value = environment[name]?.trim(); + if (value) { + return value; + } + } + return ''; +} diff --git a/app/api/session/generic-failed-start-cleanup.ts b/app/api/session/generic-failed-start-cleanup.ts new file mode 100644 index 000000000..296d767db --- /dev/null +++ b/app/api/session/generic-failed-start-cleanup.ts @@ -0,0 +1,12 @@ +const failedStartCleanupRequests = new WeakSet(); + +export function markGenericFailedStartCleanupRequest(request: Request): Request { + failedStartCleanupRequests.add(request); + return request; +} + +export function consumeGenericFailedStartCleanupRequest(request: Request): boolean { + const marked = failedStartCleanupRequests.has(request); + failedStartCleanupRequests.delete(request); + return marked; +} diff --git a/app/api/session/generic-session-dispatch.ts b/app/api/session/generic-session-dispatch.ts new file mode 100644 index 000000000..72a399b03 --- /dev/null +++ b/app/api/session/generic-session-dispatch.ts @@ -0,0 +1,123 @@ +import { + type GenericEdgeTargetSnapshot, + type GenericPairingRequest, + coordinateGenericRoomSession, + createGenericRoomInputToken, + isGenericEndpointPairingEnabled, + pairGenericEdgeMedia, + requestGenericEdgeControl, + resolveGenericEdgeTargetSnapshot, +} from '@/app/api/session/generic-edge-media-pairing'; +import { markGenericFailedStartCleanupRequest } from '@/app/api/session/generic-failed-start-cleanup'; +import { + RoomSessionCancelledError, + dispatchRoomSession, + waitForExistingRoomSessionReadiness, +} from '@/app/api/session/session-dispatch-service'; +import { getRoomSessionSnapshot } from '@/app/api/session/session-registry'; +import { POST as stopSession } from '@/app/api/session/stop/route'; + +type Environment = Record; + +type RunSessionDispatchRequest = { + roomName: string; + sessionId: string; + agentName: string; + requireRoomVideoInputReady?: boolean; +}; + +type RunSessionDispatchDependencies = { + environment?: Environment; + dispatchAgent?: () => Promise<{ agentParticipant?: { identity?: string } }>; + resolveTarget?: () => Promise; + pairEndpoint?: ( + request: GenericPairingRequest, + target: GenericEdgeTargetSnapshot + ) => Promise<{ deviceId: string; address: string }>; +}; + +export async function cleanupFailedGenericRoomSession( + roomName: string, + sessionId: string +): Promise { + const request = markGenericFailedStartCleanupRequest( + new Request('http://localhost/api/session/stop', { + method: 'POST', + body: JSON.stringify({ roomName, sessionId, wait: true }), + }) + ); + const response = await stopSession(request); + const payload = (await response.json()) as { + results?: Array<{ ok?: boolean; skipped?: boolean }>; + }; + const cleanupUnconfirmed = + !Array.isArray(payload.results) || + payload.results.some((result) => result.ok !== true && result.skipped !== true); + if (!response.ok || cleanupUnconfirmed) { + throw new Error('Generic failed-start cloud cleanup could not be confirmed'); + } +} + +export async function runSessionDispatch( + request: RunSessionDispatchRequest, + dependencies: RunSessionDispatchDependencies = {} +) { + const environment = dependencies.environment ?? process.env; + const dispatchAgent = + dependencies.dispatchAgent ?? + (() => + dispatchRoomSession({ + roomName: request.roomName, + sessionId: request.sessionId, + agentName: request.agentName, + readiness: isGenericEndpointPairingEnabled(environment) + ? {} + : { requireRoomVideoInputReady: request.requireRoomVideoInputReady === true }, + })); + if (!isGenericEndpointPairingEnabled(environment)) { + return dispatchAgent(); + } + + const roomUrl = + (environment.LIVEKIT_BROWSER_URL ?? '').trim() || (environment.LIVEKIT_URL ?? '').trim(); + if (!roomUrl) { + throw new Error('LIVEKIT_BROWSER_URL or LIVEKIT_URL is required for Generic endpoint pairing'); + } + return coordinateGenericRoomSession( + { ...request, roomUrl }, + { + dispatchAgent, + resolveTarget: + dependencies.resolveTarget ?? (() => resolveGenericEdgeTargetSnapshot({ environment })), + pairEndpoint: + dependencies.pairEndpoint ?? + ((pairingRequest, target) => + pairGenericEdgeMedia(pairingRequest, { + config: target, + createRoomToken: () => createGenericRoomInputToken(request.roomName, environment), + requestControl: requestGenericEdgeControl, + waitForReadiness: () => + waitForExistingRoomSessionReadiness( + { roomName: request.roomName, agentName: request.agentName }, + { + isCancelled: () => getRoomSessionSnapshot(request.roomName)?.cancelled === true, + } + ), + isCancelled: () => getRoomSessionSnapshot(request.roomName)?.cancelled === true, + })), + cleanupSession: () => cleanupFailedGenericRoomSession(request.roomName, request.sessionId), + } + ); +} + +export function formatSessionDispatchError( + error: unknown, + environment: Environment = process.env +): string { + if (isGenericEndpointPairingEnabled(environment)) { + return 'Generic session startup failed'; + } + return error instanceof Error ? error.message : String(error); +} + +export { RoomSessionCancelledError }; diff --git a/app/api/session/session-dispatch-service.ts b/app/api/session/session-dispatch-service.ts index 067468acc..bfb6c424a 100644 --- a/app/api/session/session-dispatch-service.ts +++ b/app/api/session/session-dispatch-service.ts @@ -41,6 +41,14 @@ type DispatchDependencies = { ) => Promise; }; +type ExistingRoomReadinessDependencies = { + roomClient?: Pick; + timeoutMs?: number; + pollMs?: number; + sleep?: (ms: number) => Promise; + isCancelled?: () => boolean; +}; + export type DispatchRoomSessionRequest = { roomName: string; sessionId: string; @@ -177,6 +185,35 @@ export async function dispatchRoomSession( } } +export async function waitForExistingRoomSessionReadiness( + request: { roomName: string; agentName: string }, + dependencies: ExistingRoomReadinessDependencies = {} +) { + const roomClient = dependencies.roomClient ?? resolveRoomClient(); + const deadline = + Date.now() + + (dependencies.timeoutMs ?? + readPositiveIntEnv('GENERIC_EDGE_MEDIA_READY_TIMEOUT_MS', DEFAULT_PREWARM_TOTAL_TIMEOUT_MS)); + const pollMs = dependencies.pollMs ?? readPositiveIntEnv('AGENT_DISPATCH_POLL_MS', 200); + const sleepFn = dependencies.sleep ?? sleep; + + while (Date.now() < deadline) { + if (dependencies.isCancelled?.()) { + throw new Error('Generic endpoint readiness wait was cancelled'); + } + const participants = await roomClient.listParticipants(request.roomName); + const participant = findReusableAgentParticipantInList(participants, request.agentName, { + allowAnonymousLiveKitAgentFallback: true, + requireExactRoomInputTracksReady: true, + }); + if (participant) { + return summarizeAgentParticipant(participant); + } + await sleepFn(Math.min(pollMs, Math.max(0, deadline - Date.now()))); + } + throw new Error('Generic endpoint media readiness timeout'); +} + async function waitForRequestedRoomSessionReadiness( request: DispatchRoomSessionRequest, dependencies: DispatchDependencies, @@ -188,7 +225,8 @@ async function waitForRequestedRoomSessionReadiness( if ( readiness.requireAgentSessionReady !== true && readiness.requireRoomInputParticipantsReady !== true && - readiness.requireRoomVideoInputReady !== true + readiness.requireRoomVideoInputReady !== true && + readiness.requireExactRoomInputTracksReady !== true ) { return dispatch; } @@ -411,6 +449,16 @@ function resolveClients(dependencies: DispatchDependencies): { }; } +function resolveRoomClient(): Pick { + 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 new RoomServiceClient(liveKitHttpUrl, apiKey, apiSecret); +} + async function ensureLiveKitRoom( roomClient: RoomClient, roomName: string, diff --git a/app/api/session/stop/route.ts b/app/api/session/stop/route.ts index bde8ec247..d92d74c50 100644 --- a/app/api/session/stop/route.ts +++ b/app/api/session/stop/route.ts @@ -10,9 +10,17 @@ import { } from '@/lib/connection-room-id'; import { executeRoomInputStopsSequentially, + normalizeRoomInputControlUrl, resolveRoomInputStopUrls as resolveConfiguredRoomInputStopUrls, resolveLiveKitHttpUrl, } from '@/lib/session-stop'; +import { + isGenericEndpointPairingEnabled, + requestGenericEdgeControl, + resolveGenericEdgeTargetSnapshot, + stopGenericEdgeMedia, +} from '../generic-edge-media-pairing'; +import { consumeGenericFailedStartCleanupRequest } from '../generic-failed-start-cleanup'; import { markRoomSessionStopped, markRoomSessionStopping, @@ -149,6 +157,17 @@ function resolveRoomInputStopUrls(): string[] { return []; } + if (isGenericEndpointPairingEnabled()) { + const processorStopUrl = normalizeRoomInputControlUrl( + readStopEnv('VIDEO_PROCESSOR_URL'), + 'stop' + ); + if (!processorStopUrl) { + throw new Error('VIDEO_PROCESSOR_URL is required for Generic room input'); + } + return [processorStopUrl]; + } + return resolveConfiguredRoomInputStopUrls({ inputSource: readStopInputSource(), audioInputDevice: readStopRoleDevice( @@ -337,7 +356,11 @@ async function postRoomInputStop( } } -async function stopRoomInput(roomName: string, sessionId: string): Promise { +async function stopRoomInput( + roomName: string, + sessionId: string, + options: { includeGenericEdge?: boolean } = {} +): Promise { let stopUrls: string[]; try { stopUrls = resolveRoomInputStopUrls(); @@ -355,19 +378,42 @@ async function stopRoomInput(roomName: string, sessionId: string): Promise + const results = await executeRoomInputStopsSequentially(stopUrls, (stopUrl) => postRoomInputStop(stopUrl, roomName, sessionId) ); + if (!isGenericEndpointPairingEnabled() || options.includeGenericEdge === false) { + return results; + } + + try { + await stopGenericEdgeMedia( + { roomName, sessionId }, + { + resolveTarget: resolveGenericEdgeTargetSnapshot, + requestControl: requestGenericEdgeControl, + } + ); + results.push({ target: 'generic_edge_media', ok: true, status: 200 }); + } catch { + results.push({ + target: 'generic_edge_media', + ok: false, + fatal: true, + error: 'Generic endpoint cleanup could not be confirmed', + }); + } + return results; } async function runRemoteSessionCleanup( roomName: string, sessionId: string, dispatchResult: StopResult, - dispatchIds: string[] + dispatchIds: string[], + options: { includeGenericEdge?: boolean } = {} ): Promise<{ results: StopResult[]; failures: StopResult[] }> { const dispatchBarrierResult = await waitForPendingDispatches(roomName, sessionId); - const roomInputResults = await stopRoomInput(roomName, sessionId); + const roomInputResults = await stopRoomInput(roomName, sessionId, options); const liveKitRoomResult = await deleteLiveKitRoom(roomName); const agentWorkerReadinessResult = await waitForLocalAgentWorkerReadiness(); const cleanupResults = [ @@ -403,6 +449,7 @@ async function runRemoteSessionCleanup( } export async function POST(req: Request) { + const isGenericFailedStartCleanup = consumeGenericFailedStartCleanupRequest(req); let body: StopRequestBody; try { body = await req.json(); @@ -470,7 +517,8 @@ export async function POST(req: Request) { roomName, sessionId, dispatchResult, - stoppingSession.dispatchIds + stoppingSession.dispatchIds, + { includeGenericEdge: !isGenericFailedStartCleanup } ); return NextResponse.json( { diff --git a/components/app/session-provider.tsx b/components/app/session-provider.tsx index 51a42b4bb..30f1af851 100644 --- a/components/app/session-provider.tsx +++ b/components/app/session-provider.tsx @@ -6,6 +6,7 @@ import { APP_CONFIG_DEFAULTS, type AppConfig } from '@/app-config'; import type { BrowserSourceClient } from '@/hooks/useBrowserSourceClient'; import { useRoom } from '@/hooks/useRoom'; import { SelectedVideoTrackProvider } from '@/hooks/useSelectedVideoTrack'; +import { ensureBrowserRandomUuid } from '@/lib/browser-runtime-compat'; const DEFAULT_BROWSER_SOURCE_CLIENT: BrowserSourceClient = { enabled: false, @@ -43,6 +44,22 @@ interface SessionProviderProps { } export const SessionProvider = ({ appConfig, children }: SessionProviderProps) => { + const compatibility = ensureBrowserRandomUuid(); + if (!compatibility.ok) { + return ( +
+
+

Browser compatibility error

+

{compatibility.message}

+
+
+ ); + } + + return {children}; +}; + +const CompatibleSessionProvider = ({ appConfig, children }: SessionProviderProps) => { const { room, isSessionActive, diff --git a/components/livekit/agent-control-bar/chat-input.tsx b/components/livekit/agent-control-bar/chat-input.tsx index b352fb100..cede8de99 100644 --- a/components/livekit/agent-control-bar/chat-input.tsx +++ b/components/livekit/agent-control-bar/chat-input.tsx @@ -1,7 +1,9 @@ import { useEffect, useRef, useState } from 'react'; import { motion } from 'motion/react'; import { PaperPlaneRightIcon, SpinnerIcon } from '@phosphor-icons/react/dist/ssr'; +import { toastAlert } from '@/components/livekit/alert-toast'; import { Button } from '@/components/livekit/button'; +import { ChatSendTimeoutError, sendChatMessageWithTimeout } from '@/lib/chat-send'; const MOTION_PROPS = { variants: { @@ -26,7 +28,7 @@ const MOTION_PROPS = { interface ChatInputProps { chatOpen: boolean; isAgentAvailable?: boolean; - onSend?: (message: string) => void; + onSend?: (message: string) => Promise | unknown; } export function ChatInput({ @@ -43,10 +45,17 @@ export function ChatInput({ try { setIsSending(true); - await onSend(message); + await sendChatMessageWithTimeout(onSend, message); setMessage(''); } catch (error) { console.error(error); + toastAlert({ + title: 'Message could not be sent', + description: + error instanceof ChatSendTimeoutError + ? 'Sending timed out. Check the connection and try again.' + : 'The message was not sent. Please try again.', + }); } finally { setIsSending(false); } diff --git a/lib/browser-room-session.ts b/lib/browser-room-session.ts index face980a4..7df53bb11 100644 --- a/lib/browser-room-session.ts +++ b/lib/browser-room-session.ts @@ -6,7 +6,7 @@ let fallbackSessionId: string | null = null; export function getVoiceSessionId( storage: Pick | null | undefined = getSessionStorage(), - createSessionId: () => string = () => crypto.randomUUID() + createSessionId: () => string = createBrowserRandomUuid ) { if (storage) { try { @@ -50,6 +50,27 @@ export function resetVoiceSessionId( export const getBrowserRoomSessionId = getVoiceSessionId; export const resetBrowserRoomSessionId = resetVoiceSessionId; +export function createBrowserRandomUuid( + cryptoProvider: Pick & Partial> = crypto +) { + if (typeof cryptoProvider.randomUUID === 'function') { + return cryptoProvider.randomUUID(); + } + + return createBrowserRandomUuidFromRandomValues(cryptoProvider); +} + +export function createBrowserRandomUuidFromRandomValues( + cryptoProvider: Pick +) { + const bytes = cryptoProvider.getRandomValues(new Uint8Array(16)); + bytes[6] = (bytes[6] & 0x0f) | 0x40; + bytes[8] = (bytes[8] & 0x3f) | 0x80; + + const hex = Array.from(bytes, (byte) => byte.toString(16).padStart(2, '0')); + return `${hex.slice(0, 4).join('')}-${hex.slice(4, 6).join('')}-${hex.slice(6, 8).join('')}-${hex.slice(8, 10).join('')}-${hex.slice(10).join('')}`; +} + function getSessionStorage() { if (typeof window === 'undefined') { return null; diff --git a/lib/browser-runtime-compat.ts b/lib/browser-runtime-compat.ts new file mode 100644 index 000000000..7a01924f1 --- /dev/null +++ b/lib/browser-runtime-compat.ts @@ -0,0 +1,45 @@ +import { createBrowserRandomUuidFromRandomValues } from './browser-room-session'; + +type BrowserCryptoProvider = Pick & Partial>; + +export type BrowserRandomUuidStatus = + | { ok: true; installed: boolean } + | { ok: false; message: string }; + +export function ensureBrowserRandomUuid( + cryptoProvider: BrowserCryptoProvider | undefined = globalThis.crypto +): BrowserRandomUuidStatus { + if (!cryptoProvider || typeof cryptoProvider.getRandomValues !== 'function') { + return { + ok: false, + message: 'This browser does not provide the secure random values required by randomUUID.', + }; + } + + if (typeof cryptoProvider.randomUUID === 'function') { + return { ok: true, installed: false }; + } + + try { + Object.defineProperty(cryptoProvider, 'randomUUID', { + configurable: true, + enumerable: false, + writable: false, + value: () => createBrowserRandomUuidFromRandomValues(cryptoProvider), + }); + } catch { + return { + ok: false, + message: 'This browser could not install the required randomUUID compatibility support.', + }; + } + + if (typeof cryptoProvider.randomUUID !== 'function') { + return { + ok: false, + message: 'This browser could not install the required randomUUID compatibility support.', + }; + } + + return { ok: true, installed: true }; +} diff --git a/lib/chat-send.ts b/lib/chat-send.ts new file mode 100644 index 000000000..ec9aaa09c --- /dev/null +++ b/lib/chat-send.ts @@ -0,0 +1,27 @@ +export class ChatSendTimeoutError extends Error { + constructor(timeoutMs: number) { + super(`Message send timed out after ${timeoutMs}ms`); + this.name = 'ChatSendTimeoutError'; + } +} + +export const CHAT_SEND_TIMEOUT_MS = 8_000; + +export async function sendChatMessageWithTimeout( + send: (message: string) => Promise | unknown, + message: string, + timeoutMs: number = CHAT_SEND_TIMEOUT_MS +): Promise { + let timeout: ReturnType | undefined; + const timeoutPromise = new Promise((_, reject) => { + timeout = setTimeout(() => reject(new ChatSendTimeoutError(timeoutMs)), timeoutMs); + }); + + try { + await Promise.race([Promise.resolve().then(() => send(message)), timeoutPromise]); + } finally { + if (timeout !== undefined) { + clearTimeout(timeout); + } + } +} diff --git a/lib/endpoint-connectivity.ts b/lib/endpoint-connectivity.ts new file mode 100644 index 000000000..3b1efec72 --- /dev/null +++ b/lib/endpoint-connectivity.ts @@ -0,0 +1,82 @@ +import { timingSafeEqual } from 'node:crypto'; + +export const ENDPOINT_CONNECTIVITY_TOKEN_HEADER = 'x-endpoint-connectivity-token'; + +export type EndpointConnectivityPayload = { + deviceId: string; + instanceId: string; + hostname: string; + address: string; +}; + +type ParseResult = + | { ok: true; payload: EndpointConnectivityPayload } + | { ok: false; error: string }; + +const DEVICE_ID_PATTERN = /^[A-Za-z0-9._:-]{1,128}$/; +const INSTANCE_ID_PATTERN = + /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i; + +export 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 function readConnectivityToken(request: Request): string { + const direct = (request.headers.get(ENDPOINT_CONNECTIVITY_TOKEN_HEADER) || '').trim(); + if (direct) { + return direct; + } + + const authorization = (request.headers.get('authorization') || '').trim(); + const [scheme, token] = authorization.split(/\s+/, 2); + return scheme?.toLowerCase() === 'bearer' ? (token || '').trim() : ''; +} + +export function parseEndpointConnectivityPayload(input: unknown): ParseResult { + if (!input || typeof input !== 'object' || Array.isArray(input)) { + return { ok: false, error: 'JSON object is required' }; + } + + const record = input as Record; + if (Object.keys(record).sort().join(',') !== 'address,deviceId,hostname,instanceId') { + return { ok: false, error: 'heartbeat fields do not match the contract' }; + } + const deviceId = readOptionalString(record.deviceId); + if (!deviceId || !DEVICE_ID_PATTERN.test(deviceId)) { + return { + ok: false, + error: 'deviceId must contain 1-128 letters, numbers, dots, underscores, colons, or hyphens', + }; + } + + const instanceId = readOptionalString(record.instanceId); + if (!INSTANCE_ID_PATTERN.test(instanceId)) { + return { ok: false, error: 'instanceId must be an RFC 4122 version 4 UUID' }; + } + const hostname = readOptionalString(record.hostname); + const address = readOptionalString(record.address); + if (!hostname || hostname.length > 255) { + return { ok: false, error: 'hostname must contain 1-255 characters' }; + } + if (!address || address.length > 15) { + return { ok: false, error: 'address must be a canonical IPv4 address' }; + } + + return { + ok: true, + payload: { + deviceId, + instanceId: instanceId.toLowerCase(), + hostname, + address, + }, + }; +} + +function readOptionalString(value: unknown): string { + return typeof value === 'string' ? value.trim() : ''; +} diff --git a/lib/generic-endpoint-lease.ts b/lib/generic-endpoint-lease.ts new file mode 100644 index 000000000..837550bfc --- /dev/null +++ b/lib/generic-endpoint-lease.ts @@ -0,0 +1,383 @@ +import { randomUUID } from 'node:crypto'; +import { chmod, lstat, mkdir, open, readFile, readdir, rename, unlink } from 'node:fs/promises'; +import path from 'node:path'; +import type { EndpointConnectivityPayload } from './endpoint-connectivity'; + +const LEASE_TTL_MS = 45_000; +const REGISTRY_LOCK_FILE = '.generic-endpoint-lease.lock'; +const REGISTRY_LOCK_TIMEOUT_MS = 3_000; +const REGISTRY_STALE_LOCK_MS = 10_000; +const DEVICE_ID_PATTERN = /^[A-Za-z0-9._:-]{1,128}$/; +const INSTANCE_ID_PATTERN = + /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i; +const LEASE_FILE_PATTERN = /^([0-9a-f-]{36})\.lease\.json$/i; +const RECORD_FIELDS = ['address', 'deviceId', 'expiresAt', 'instanceId', 'receivedAt'] as const; + +type Environment = Record; +type ClockDependencies = { now?: () => Date }; + +type Cidr = { network: number; prefix: number; source: string }; + +export type GenericEndpointLeaseConfig = { + deviceId: string; + allowedCidrs: readonly Cidr[]; + registryDir: string; +}; + +export type GenericEndpointLease = { + deviceId: string; + instanceId: string; + address: string; + receivedAt: string; + expiresAt: string; +}; + +export class EndpointLeaseUnavailableError extends Error { + constructor(message = 'Generic endpoint lease is unavailable') { + super(message); + this.name = 'EndpointLeaseUnavailableError'; + } +} + +export class EndpointLeaseConflictError extends Error { + constructor(message = 'Generic endpoint lease has multiple active instances') { + super(message); + this.name = 'EndpointLeaseConflictError'; + } +} + +export function loadGenericEndpointLeaseConfig( + environment: Environment = process.env +): GenericEndpointLeaseConfig { + const deviceId = readEnv(environment, 'GENERIC_EDGE_MEDIA_DEVICE_ID'); + const cidrText = readEnv(environment, 'GENERIC_EDGE_MEDIA_ALLOWED_CIDRS'); + const registryDir = readEnv(environment, 'GENERIC_ENDPOINT_REGISTRY_DIR'); + if (!DEVICE_ID_PATTERN.test(deviceId)) { + throw new Error('GENERIC_EDGE_MEDIA_DEVICE_ID is required'); + } + if (!cidrText) { + throw new Error('GENERIC_EDGE_MEDIA_ALLOWED_CIDRS is required'); + } + if (!path.isAbsolute(registryDir)) { + throw new Error('GENERIC_ENDPOINT_REGISTRY_DIR must be absolute'); + } + if (readEnv(environment, 'GENERIC_ENDPOINT_MULTI_HOST') === '1') { + throw new Error('Generic endpoint lease does not support multi-host Next deployments'); + } + return { + deviceId, + allowedCidrs: cidrText.split(',').map((item) => parseAllowedCidr(item.trim())), + registryDir: path.resolve(registryDir), + }; +} + +export async function renewGenericEndpointLease( + heartbeat: EndpointConnectivityPayload, + config: GenericEndpointLeaseConfig, + dependencies: ClockDependencies = {} +): Promise { + if (heartbeat.deviceId !== config.deviceId) { + throw new Error('heartbeat deviceId does not match configured device'); + } + const instanceId = normalizeInstanceId(heartbeat.instanceId); + const address = validateAllowedAddress(heartbeat.address, config.allowedCidrs); + await ensureSecureRegistryDirectory(config.registryDir); + return withRegistryLock(config.registryDir, async () => { + const now = (dependencies.now ?? (() => new Date()))(); + const records = await readLeaseRecords(config, now, true); + if (records.some((record) => record.instanceId !== instanceId)) { + throw new EndpointLeaseConflictError(); + } + const record: GenericEndpointLease = { + deviceId: config.deviceId, + instanceId, + address, + receivedAt: now.toISOString(), + expiresAt: new Date(now.getTime() + LEASE_TTL_MS).toISOString(), + }; + await writeLeaseAtomically(config.registryDir, record); + return record; + }); +} + +export async function resolveActiveGenericEndpointLease( + config: GenericEndpointLeaseConfig, + dependencies: ClockDependencies = {} +): Promise { + await ensureSecureRegistryDirectory(config.registryDir); + return withRegistryLock(config.registryDir, async () => { + const records = await readLeaseRecords( + config, + (dependencies.now ?? (() => new Date()))(), + true + ); + if (records.length === 0) { + throw new EndpointLeaseUnavailableError(); + } + if (records.length !== 1) { + throw new EndpointLeaseConflictError(); + } + return Object.freeze({ ...records[0] }); + }); +} + +export function buildGenericEdgeControlUrl( + lease: Pick, + action: 'start' | 'stop' +): string { + const address = validateCanonicalPrivateIpv4(lease.address); + return `http://${address}:8013/${action}`; +} + +async function ensureSecureRegistryDirectory(directory: string): Promise { + try { + const info = await lstat(directory); + validateRegistryDirectory(info, directory); + } catch (error) { + if (!isNodeError(error, 'ENOENT')) { + throw error; + } + await mkdir(directory, { recursive: true, mode: 0o700 }); + await chmod(directory, 0o700); + validateRegistryDirectory(await lstat(directory), directory); + } +} + +function validateRegistryDirectory(info: Awaited>, directory: string) { + if (info.isSymbolicLink() || !info.isDirectory()) { + throw new Error('Generic endpoint registry directory must be a real directory'); + } + if ((Number(info.mode) & 0o777) !== 0o700) { + throw new Error('Generic endpoint registry directory must use mode 0700'); + } + if (typeof process.getuid === 'function' && info.uid !== process.getuid()) { + throw new Error(`Generic endpoint registry directory has the wrong owner: ${directory}`); + } +} + +async function readLeaseRecords( + config: GenericEndpointLeaseConfig, + now: Date, + cleanExpired: boolean +): Promise { + const records: GenericEndpointLease[] = []; + for (const entry of await readdir(config.registryDir)) { + const match = LEASE_FILE_PATTERN.exec(entry); + if (!match) { + continue; + } + const filePath = path.join(config.registryDir, entry); + const info = await lstat(filePath); + if (info.isSymbolicLink() || !info.isFile()) { + throw new Error('Generic endpoint lease record must be a regular file'); + } + if ((info.mode & 0o777) !== 0o600) { + throw new Error('Generic endpoint lease record must use mode 0600'); + } + if (typeof process.getuid === 'function' && info.uid !== process.getuid()) { + throw new Error('Generic endpoint lease record has the wrong owner'); + } + let parsed: unknown; + try { + parsed = JSON.parse(await readFile(filePath, 'utf8')); + } catch { + throw new Error('Generic endpoint lease record is invalid'); + } + const record = validateLeaseRecord(parsed, config); + if (record.instanceId !== match[1].toLowerCase()) { + throw new Error('Generic endpoint lease record identity mismatch'); + } + if (Date.parse(record.expiresAt) <= now.getTime()) { + if (cleanExpired) { + await unlink(filePath); + } + continue; + } + records.push(record); + } + return records; +} + +function validateLeaseRecord(value: unknown, config: GenericEndpointLeaseConfig) { + if (!value || typeof value !== 'object' || Array.isArray(value)) { + throw new Error('Generic endpoint lease record is invalid'); + } + const record = value as Record; + if (Object.keys(record).sort().join(',') !== [...RECORD_FIELDS].sort().join(',')) { + throw new Error('Generic endpoint lease record fields are invalid'); + } + if (record.deviceId !== config.deviceId) { + throw new Error('Generic endpoint lease record device is invalid'); + } + const instanceId = normalizeInstanceId(record.instanceId); + const address = validateAllowedAddress(record.address, config.allowedCidrs); + const receivedAt = validateTimestamp(record.receivedAt, 'receivedAt'); + const expiresAt = validateTimestamp(record.expiresAt, 'expiresAt'); + return { deviceId: config.deviceId, instanceId, address, receivedAt, expiresAt }; +} + +async function writeLeaseAtomically(directory: string, record: GenericEndpointLease) { + const destination = path.join(directory, `${record.instanceId}.lease.json`); + const temporary = path.join(directory, `.${record.instanceId}.${randomUUID()}.tmp`); + try { + const file = await open(temporary, 'wx', 0o600); + try { + await file.writeFile(JSON.stringify(record)); + await file.sync(); + } finally { + await file.close(); + } + await rename(temporary, destination); + await chmod(destination, 0o600); + } catch (error) { + await unlink(temporary).catch((unlinkError) => { + if (!isNodeError(unlinkError, 'ENOENT')) throw unlinkError; + }); + throw error; + } +} + +async function withRegistryLock(directory: string, operation: () => Promise): Promise { + const lockPath = path.join(directory, REGISTRY_LOCK_FILE); + const deadline = Date.now() + REGISTRY_LOCK_TIMEOUT_MS; + while (true) { + let acquired = false; + try { + const file = await open(lockPath, 'wx', 0o600); + acquired = true; + try { + await file.writeFile(`${process.pid}\n`); + await file.sync(); + } finally { + await file.close(); + } + break; + } catch (error) { + if (acquired) { + await unlink(lockPath).catch((unlinkError) => { + if (!isNodeError(unlinkError, 'ENOENT')) throw unlinkError; + }); + } + if (!isNodeError(error, 'EEXIST')) throw error; + await clearDeadRegistryLock(lockPath); + if (Date.now() >= deadline) { + throw new Error('Generic endpoint lease registry is busy'); + } + await new Promise((resolve) => setTimeout(resolve, 10)); + } + } + + try { + return await operation(); + } finally { + await unlink(lockPath).catch((error) => { + if (!isNodeError(error, 'ENOENT')) throw error; + }); + } +} + +async function clearDeadRegistryLock(lockPath: string): Promise { + try { + const info = await lstat(lockPath); + if (info.isSymbolicLink() || !info.isFile() || (Number(info.mode) & 0o777) !== 0o600) { + throw new Error('Generic endpoint lease registry lock is unsafe'); + } + if (typeof process.getuid === 'function' && info.uid !== process.getuid()) { + throw new Error('Generic endpoint lease registry lock has the wrong owner'); + } + + const owner = Number.parseInt((await readFile(lockPath, 'utf8')).trim(), 10); + let ownerAlive = Number.isInteger(owner) && owner > 0; + if (ownerAlive) { + try { + process.kill(owner, 0); + } catch (error) { + if (!isNodeError(error, 'ESRCH')) throw error; + ownerAlive = false; + } + } + if (!ownerAlive && Date.now() - info.mtimeMs >= REGISTRY_STALE_LOCK_MS) { + await unlink(lockPath).catch((error) => { + if (!isNodeError(error, 'ENOENT')) throw error; + }); + } + } catch (error) { + if (!isNodeError(error, 'ENOENT')) throw error; + } +} + +function parseAllowedCidr(value: string): Cidr { + const [address, prefixText, ...extra] = value.split('/'); + const prefix = Number(prefixText); + if (extra.length > 0 || !Number.isInteger(prefix) || prefix < 0 || prefix > 32) { + throw new Error('GENERIC_EDGE_MEDIA_ALLOWED_CIDRS contains an invalid IPv4 CIDR'); + } + const canonical = validateCanonicalPrivateIpv4(address); + const numeric = ipv4ToInt(canonical); + const mask = prefix === 0 ? 0 : (0xffffffff << (32 - prefix)) >>> 0; + if ((numeric & mask) >>> 0 !== numeric) { + throw new Error('GENERIC_EDGE_MEDIA_ALLOWED_CIDRS must use canonical network addresses'); + } + return { network: numeric, prefix, source: value }; +} + +function validateAllowedAddress(value: unknown, cidrs: readonly Cidr[]): string { + const address = validateCanonicalPrivateIpv4(value); + const numeric = ipv4ToInt(address); + const allowed = cidrs.some(({ network, prefix }) => { + const mask = prefix === 0 ? 0 : (0xffffffff << (32 - prefix)) >>> 0; + return (numeric & mask) >>> 0 === network; + }); + if (!allowed) { + throw new Error('heartbeat address is outside the allowed CIDR policy'); + } + return address; +} + +function validateCanonicalPrivateIpv4(value: unknown): string { + if (typeof value !== 'string' || !/^\d{1,3}(?:\.\d{1,3}){3}$/.test(value)) { + throw new Error('heartbeat address must be canonical private IPv4'); + } + const octets = value.split('.'); + if (octets.some((part) => String(Number(part)) !== part || Number(part) > 255)) { + throw new Error('heartbeat address must be canonical private IPv4'); + } + const [first, second] = octets.map(Number); + const isPrivate = + first === 10 || + (first === 172 && second >= 16 && second <= 31) || + (first === 192 && second === 168); + if (!isPrivate) { + throw new Error('heartbeat address must be RFC1918 private IPv4'); + } + return value; +} + +function ipv4ToInt(address: string): number { + return address + .split('.') + .map(Number) + .reduce((value, octet) => ((value << 8) | octet) >>> 0, 0); +} + +function normalizeInstanceId(value: unknown): string { + if (typeof value !== 'string' || !INSTANCE_ID_PATTERN.test(value)) { + throw new Error('heartbeat instanceId must be an RFC 4122 version 4 UUID'); + } + return value.toLowerCase(); +} + +function validateTimestamp(value: unknown, name: string): string { + if (typeof value !== 'string' || !Number.isFinite(Date.parse(value))) { + throw new Error(`Generic endpoint lease record ${name} is invalid`); + } + return value; +} + +function readEnv(environment: Environment, name: string): string { + return environment[name]?.trim() ?? ''; +} + +function isNodeError(error: unknown, code: string): boolean { + return error instanceof Error && 'code' in error && error.code === code; +} diff --git a/lib/session-dispatch-readiness.ts b/lib/session-dispatch-readiness.ts index 2c6026cc3..4e87a3e67 100644 --- a/lib/session-dispatch-readiness.ts +++ b/lib/session-dispatch-readiness.ts @@ -13,6 +13,7 @@ export type ReusableAgentParticipantOptions = AgentParticipantMatchOptions & { requireAgentSessionReady?: boolean; requireRoomVideoInputReady?: boolean; requireRoomInputParticipantsReady?: boolean; + requireExactRoomInputTracksReady?: boolean; }; export const AGENT_SESSION_READY_ATTRIBUTE = 'liveavatar.agent.session_ready'; @@ -37,12 +38,16 @@ export function findReusableAgentParticipant( requireAgentSessionReady = false, requireRoomVideoInputReady = false, requireRoomInputParticipantsReady = false, + requireExactRoomInputTracksReady = false, ...matchOptions } = options; const expectedAgent = findAgentParticipantInList(participants, agentName, matchOptions); if (!expectedAgent || (requireAgentSessionReady && !isAgentSessionReady(expectedAgent))) { return null; } + if (requireExactRoomInputTracksReady && !hasExactReadyRoomInputTracks(participants)) { + return null; + } if (!requireRoomVideoInputReady) { return requireRoomInputParticipantsReady && !hasReadyRoomInputParticipants(participants) @@ -113,6 +118,34 @@ function hasReadyRoomInputParticipants(participants: ParticipantInfo[]) { return readiness.audioParticipantReady && readiness.visionParticipantReady; } +function hasExactReadyRoomInputTracks(participants: ParticipantInfo[]) { + return ( + hasReadyRoomAudioInput(participants) && + participants.some( + (participant) => + isParticipantActive(participant) && + participant.identity === ROOM_VIDEO_INPUT_IDENTITY && + hasReadyTrack(participant, 'room_video', TrackType.VIDEO) + ) + ); +} + +function hasReadyRoomAudioInput(participants: ParticipantInfo[]) { + return participants.some( + (participant) => + isParticipantActive(participant) && + participant.identity === ROOM_AUDIO_INPUT_IDENTITY && + hasReadyTrack(participant, 'room_audio', TrackType.AUDIO) && + hasReadyTrack(participant, 'room_video_raw', TrackType.VIDEO) + ); +} + +function hasReadyTrack(participant: ParticipantInfo, name: string, type: TrackType) { + return (participant.tracks ?? []).some( + (track) => track.name === name && track.type === type && track.muted !== true + ); +} + function hasActiveParticipant(participants: ParticipantInfo[], identity: string) { return participants.some( (participant) => participant.identity === identity && isParticipantActive(participant) diff --git a/tests/browser-room-session.test.mjs b/tests/browser-room-session.test.mjs index 153264045..f94dba1b5 100644 --- a/tests/browser-room-session.test.mjs +++ b/tests/browser-room-session.test.mjs @@ -44,6 +44,74 @@ function createMemoryStorage() { }; } +async function withCryptoProvider(provider, callback) { + const descriptor = Object.getOwnPropertyDescriptor(globalThis, 'crypto'); + Object.defineProperty(globalThis, 'crypto', { + configurable: true, + value: provider, + }); + try { + return await callback(); + } finally { + if (descriptor) { + Object.defineProperty(globalThis, 'crypto', descriptor); + } else { + delete globalThis.crypto; + } + } +} + +test('creates an RFC 4122 v4 browser room id when native randomUUID is unavailable', async () => { + const { createBrowserRandomUuid, isValidConnectionRoomId } = await loadBrowserRoomSessionModule(); + + const sessionId = await withCryptoProvider( + { + getRandomValues(bytes) { + bytes.set(Array.from({ length: 16 }, (_, index) => index)); + return bytes; + }, + }, + () => createBrowserRandomUuid() + ); + + assert.equal(sessionId, '00010203-0405-4607-8809-0a0b0c0d0e0f'); + assert.equal(sessionId[14], '4'); + assert.match(sessionId[19], /[89ab]/); + assert.equal(isValidConnectionRoomId(sessionId), true); +}); + +test('prefers native randomUUID for a new browser room id', async () => { + const { createBrowserRandomUuid } = await loadBrowserRoomSessionModule(); + const nativeSessionId = '33333333-4444-4555-8666-777777777777'; + + const sessionId = await withCryptoProvider( + { + randomUUID() { + return nativeSessionId; + }, + getRandomValues() { + assert.fail('getRandomValues must not run when native randomUUID is callable'); + }, + }, + () => createBrowserRandomUuid() + ); + + assert.equal(sessionId, nativeSessionId); +}); + +test('reuses a valid stored browser room id without creating another', async () => { + const { getBrowserRoomSessionId } = await loadBrowserRoomSessionModule(); + const storage = createMemoryStorage(); + const storedSessionId = '44444444-5555-4666-8777-888888888888'; + storage.setItem('lexvoice.session_id.v1', storedSessionId); + + const sessionId = getBrowserRoomSessionId(storage, () => { + assert.fail('stored session id should be reused'); + }); + + assert.equal(sessionId, storedSessionId); +}); + test('resetting browser room session rotates the next room id', async () => { const { getBrowserRoomSessionId, resetBrowserRoomSessionId } = await loadBrowserRoomSessionModule(); diff --git a/tests/browser-runtime-compat.test.mjs b/tests/browser-runtime-compat.test.mjs new file mode 100644 index 000000000..b86f250ef --- /dev/null +++ b/tests/browser-runtime-compat.test.mjs @@ -0,0 +1,67 @@ +import assert from 'node:assert/strict'; +import { readFile } from 'node:fs/promises'; +import test from 'node:test'; +import * as browserCompat from '../lib/browser-runtime-compat.ts'; + +test('installs an RFC 4122 v4 randomUUID when the browser crypto API omits it', () => { + const cryptoProvider = { + getRandomValues(bytes) { + bytes.set(Array.from({ length: 16 }, (_, index) => index)); + return bytes; + }, + }; + + const result = browserCompat.ensureBrowserRandomUuid(cryptoProvider); + + assert.deepEqual(result, { ok: true, installed: true }); + assert.equal(cryptoProvider.randomUUID(), '00010203-0405-4607-8809-0a0b0c0d0e0f'); + assert.equal(cryptoProvider.randomUUID()[14], '4'); + assert.match(cryptoProvider.randomUUID()[19], /[89ab]/); +}); + +test('preserves the native randomUUID implementation', () => { + const nativeRandomUuid = () => '33333333-4444-4555-8666-777777777777'; + const cryptoProvider = { + randomUUID: nativeRandomUuid, + getRandomValues() { + assert.fail('getRandomValues must not run when native randomUUID exists'); + }, + }; + + const result = browserCompat.ensureBrowserRandomUuid(cryptoProvider); + + assert.deepEqual(result, { ok: true, installed: false }); + assert.equal(cryptoProvider.randomUUID, nativeRandomUuid); +}); + +test('fails safely when the browser crypto object rejects the polyfill', () => { + const cryptoProvider = Object.preventExtensions({ + getRandomValues(bytes) { + return bytes; + }, + }); + + const result = browserCompat.ensureBrowserRandomUuid(cryptoProvider); + + assert.equal(result.ok, false); + assert.match(result.message, /randomUUID/i); + assert.equal('randomUUID' in cryptoProvider, false); +}); + +test('SessionProvider exposes an explicit compatibility error before creating a Room', async () => { + const source = await readFile('components/app/session-provider.tsx', 'utf8'); + + assert.match(source, /ensureBrowserRandomUuid\(/); + assert.match(source, /role="alert"/); + assert.match(source, /Browser compatibility error/); +}); + +test('AgentControlBar retains the high-level useChat send path', async () => { + const source = await readFile( + 'components/livekit/agent-control-bar/agent-control-bar.tsx', + 'utf8' + ); + + assert.match(source, /await send\(message\)/); + assert.doesNotMatch(source, /streamText|sendBrowserChatMessage/); +}); diff --git a/tests/chat-send.test.mjs b/tests/chat-send.test.mjs new file mode 100644 index 000000000..6b728bb97 --- /dev/null +++ b/tests/chat-send.test.mjs @@ -0,0 +1,53 @@ +import assert from 'node:assert/strict'; +import { readFile } from 'node:fs/promises'; +import test from 'node:test'; +import * as chatSend from '../lib/chat-send.ts'; + +const { ChatSendTimeoutError, sendChatMessageWithTimeout } = chatSend; + +test('sendChatMessageWithTimeout returns after a successful send', async () => { + let received = ''; + + await sendChatMessageWithTimeout( + async (message) => { + received = message; + }, + 'hello', + 25 + ); + + assert.equal(received, 'hello'); +}); + +test('sendChatMessageWithTimeout surfaces an underlying send rejection', async () => { + await assert.rejects( + sendChatMessageWithTimeout( + async () => { + throw new Error('data transport unavailable'); + }, + 'hello', + 25 + ), + /data transport unavailable/ + ); +}); + +test( + 'sendChatMessageWithTimeout rejects a stalled send within the configured bound', + { timeout: 100 }, + async () => { + await assert.rejects( + sendChatMessageWithTimeout(() => new Promise(() => {}), 'hello', 10), + ChatSendTimeoutError + ); + } +); + +test('ChatInput reports send failures visibly and retains the message for retry', async () => { + const source = await readFile('components/livekit/agent-control-bar/chat-input.tsx', 'utf8'); + + assert.match(source, /sendChatMessageWithTimeout\(onSend, message\)/); + assert.match(source, /toastAlert\(/); + assert.match(source, /Message could not be sent/); + assert.ok(source.indexOf("setMessage('')") < source.indexOf('catch (error)')); +}); diff --git a/tests/connection-details.test.mjs b/tests/connection-details.test.mjs index 149100425..7e6ef6f67 100644 --- a/tests/connection-details.test.mjs +++ b/tests/connection-details.test.mjs @@ -89,3 +89,55 @@ test('connection details route logs issued token with canonical session identity assert.match(routeSource, /roomName/); assert.match(routeSource, /participantIdentity/); }); + +test('connection details advertise the explicit browser LiveKit URL without Host derivation', async () => { + process.env.LIVEKIT_BROWSER_URL = 'ws://10.2.77.108:7818'; + process.env.LIVEKIT_URL = 'ws://127.0.0.1:7818'; + const { POST } = await import( + new URL( + '../app/api/connection-details/route.ts?browser-url-contract', + import.meta.url + ).href + ); + + const response = await POST( + new Request('http://attacker.example/api/connection-details', { + method: 'POST', + headers: { + 'content-type': 'application/json', + host: 'attacker.example', + 'x-forwarded-host': 'forwarded-attacker.example', + }, + body: '{}', + }) + ); + const payload = await response.json(); + + assert.equal(response.status, 200); + assert.equal(payload.serverUrl, 'ws://10.2.77.108:7818'); + const routeSource = await readFile( + new URL('../app/api/connection-details/route.ts', import.meta.url), + 'utf8' + ); + assert.doesNotMatch(routeSource, /headers\.get\(['"](?:host|x-forwarded-host)['"]\)/i); +}); + +test('connection details fall back to internal LiveKit URL for standalone localhost', async () => { + delete process.env.LIVEKIT_BROWSER_URL; + process.env.LIVEKIT_URL = 'ws://localhost:7818'; + const { POST } = await import( + new URL('../app/api/connection-details/route.ts?localhost-fallback', import.meta.url).href + ); + + const response = await POST( + new Request('http://localhost/api/connection-details', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: '{}', + }) + ); + const payload = await response.json(); + + assert.equal(response.status, 200); + assert.equal(payload.serverUrl, 'ws://localhost:7818'); +}); diff --git a/tests/endpoint-connectivity.test.mjs b/tests/endpoint-connectivity.test.mjs new file mode 100644 index 000000000..6e4b71968 --- /dev/null +++ b/tests/endpoint-connectivity.test.mjs @@ -0,0 +1,134 @@ +import assert from 'node:assert/strict'; +import { randomUUID } from 'node:crypto'; +import { mkdir } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { test } from 'node:test'; +import { POST as endpointConnectivity } from '../app/api/endpoint/connectivity/route.ts'; +import { + ENDPOINT_CONNECTIVITY_TOKEN_HEADER, + parseEndpointConnectivityPayload, + readConnectivityToken, + secretsMatch, +} from '../lib/endpoint-connectivity.ts'; + +async function temporaryRegistry() { + const directory = path.join(tmpdir(), `generic-endpoint-route-${randomUUID()}`); + await mkdir(directory, { mode: 0o700 }); + return directory; +} + +test('endpoint connectivity payload accepts bounded endpoint identity metadata', () => { + assert.deepEqual( + parseEndpointConnectivityPayload({ + deviceId: 'yahboom-001', + instanceId: '11111111-2222-4333-8444-555555555555', + hostname: 'yahboom', + address: '10.2.2.199', + }), + { + ok: true, + payload: { + deviceId: 'yahboom-001', + instanceId: '11111111-2222-4333-8444-555555555555', + hostname: 'yahboom', + address: '10.2.2.199', + }, + } + ); +}); + +test('endpoint connectivity payload rejects unsafe device ids', () => { + const result = parseEndpointConnectivityPayload({ deviceId: '../../yahboom' }); + assert.equal(result.ok, false); +}); + +test('endpoint connectivity payload rejects missing, extra, and non-v4 identity fields', () => { + const valid = { + deviceId: 'generic-orin', + instanceId: '11111111-2222-4333-8444-555555555555', + hostname: 'orin', + address: '10.2.2.199', + }; + for (const invalid of [ + { ...valid, extra: 'rejected' }, + { ...valid, hostname: undefined }, + { ...valid, instanceId: '11111111-2222-1333-8444-555555555555' }, + ]) { + assert.equal(parseEndpointConnectivityPayload(invalid).ok, false); + } +}); + +test('endpoint connectivity token supports a dedicated header and bearer auth', () => { + assert.equal( + readConnectivityToken( + new Request('http://cloud.test/api/endpoint/connectivity', { + headers: { [ENDPOINT_CONNECTIVITY_TOKEN_HEADER]: 'direct-token' }, + }) + ), + 'direct-token' + ); + assert.equal( + readConnectivityToken( + new Request('http://cloud.test/api/endpoint/connectivity', { + headers: { Authorization: 'Bearer bearer-token' }, + }) + ), + 'bearer-token' + ); +}); + +test('endpoint connectivity token comparison requires an exact match', () => { + assert.equal(secretsMatch('probe-secret', 'probe-secret'), true); + assert.equal(secretsMatch('probe-secret', 'other-secret'), false); +}); + +test('connectivity route renews only the configured fixed device lease', async () => { + const names = [ + 'ENDPOINT_CONNECTIVITY_TOKEN', + 'GENERIC_EDGE_MEDIA_DEVICE_ID', + 'GENERIC_EDGE_MEDIA_ALLOWED_CIDRS', + 'GENERIC_ENDPOINT_REGISTRY_DIR', + ]; + const previous = Object.fromEntries(names.map((name) => [name, process.env[name]])); + Object.assign(process.env, { + ENDPOINT_CONNECTIVITY_TOKEN: 'probe-secret', + GENERIC_EDGE_MEDIA_DEVICE_ID: 'generic-orin', + GENERIC_EDGE_MEDIA_ALLOWED_CIDRS: '10.2.0.0/16', + GENERIC_ENDPOINT_REGISTRY_DIR: await temporaryRegistry(), + }); + try { + const response = await endpointConnectivity( + new Request('http://cloud.test/api/endpoint/connectivity', { + method: 'POST', + headers: { + [ENDPOINT_CONNECTIVITY_TOKEN_HEADER]: 'probe-secret', + 'Content-Type': 'application/json', + 'X-Forwarded-For': '203.0.113.55', + }, + body: JSON.stringify({ + deviceId: 'generic-orin', + instanceId: '11111111-2222-4333-8444-555555555555', + hostname: 'orin', + address: '10.2.2.199', + }), + }) + ); + assert.equal(response.status, 200); + const payload = await response.json(); + assert.equal(payload.status, 'leased'); + assert.equal(payload.deviceId, 'generic-orin'); + assert.equal(payload.instanceId, '11111111-2222-4333-8444-555555555555'); + assert.equal(payload.hostname, 'orin'); + assert.equal(payload.address, '10.2.2.199'); + assert.equal('token' in payload, false); + } finally { + for (const name of names) { + if (previous[name] === undefined) { + delete process.env[name]; + } else { + process.env[name] = previous[name]; + } + } + } +}); diff --git a/tests/generic-edge-media.test.mjs b/tests/generic-edge-media.test.mjs new file mode 100644 index 000000000..74d38b84b --- /dev/null +++ b/tests/generic-edge-media.test.mjs @@ -0,0 +1,466 @@ +import assert from 'node:assert/strict'; +import { test } from 'node:test'; +import { + coordinateGenericRoomSession, + createGenericRoomInputToken, + pairGenericEdgeMedia, + requestGenericEdgeControl, + resolveGenericEdgeTargetSnapshot, + stopGenericEdgeMedia, +} from '../app/api/session/generic-edge-media-pairing.ts'; + +test('Generic pairing reclaims stale state, starts exactly once, then waits for exact media', async () => { + const events = []; + const config = { + startUrl: 'http://10.2.2.199:8013/start', + stopUrl: 'http://10.2.2.199:8013/stop', + controlToken: 'control-secret', + deviceId: 'generic-orin', + address: '10.2.2.199', + }; + const result = await pairGenericEdgeMedia( + { + roomUrl: 'ws://10.2.77.108:7818', + roomName: 'voice_assistant_room_session-1', + sessionId: 'session-1', + controlSenderIdentity: 'agent-joined', + }, + { + config, + createRoomToken: async () => 'short-lived-room-token', + requestControl: async (action, payload, resolvedConfig) => { + events.push({ action, payload, resolvedConfig }); + }, + waitForReadiness: async () => { + events.push({ action: 'wait' }); + }, + } + ); + + assert.deepEqual( + events.map((event) => event.action), + ['stop', 'start', 'wait'] + ); + const start = events[1]; + assert.equal(start.resolvedConfig.controlToken, 'control-secret'); + assert.deepEqual(start.payload, { + room_url: 'ws://10.2.77.108:7818', + room_token: 'short-lived-room-token', + room_name: 'voice_assistant_room_session-1', + session_id: 'session-1', + service_instance_id: 'generic-orin', + source_type: 'generic', + control_sender_identity: 'agent-joined', + participant_identity: 'room_audio_input', + track_names: { audio: 'room_audio', video: 'room_video_raw' }, + }); + assert.deepEqual(result, { deviceId: 'generic-orin', address: '10.2.2.199' }); +}); + +test('Generic pairing rolls endpoint state back when readiness fails', async () => { + const actions = []; + await assert.rejects( + pairGenericEdgeMedia( + { + roomUrl: 'ws://livekit.test', + roomName: 'room-1', + sessionId: 'session-1', + controlSenderIdentity: 'agent-joined', + }, + { + config: { + startUrl: 'http://10.2.2.199:8013/start', + stopUrl: 'http://10.2.2.199:8013/stop', + controlToken: 'secret', + deviceId: 'generic-orin', + address: '10.2.2.199', + }, + createRoomToken: async () => 'token', + requestControl: async (action) => { + actions.push(action); + }, + waitForReadiness: async () => { + throw new Error('media readiness timeout'); + }, + } + ), + /media readiness timeout/ + ); + assert.deepEqual(actions, ['stop', 'start', 'stop']); +}); + +test('Generic pairing performs no start when stale reclaim fails', async () => { + const actions = []; + await assert.rejects( + pairGenericEdgeMedia( + { + roomUrl: 'ws://livekit.test', + roomName: 'room-1', + sessionId: 'session-1', + controlSenderIdentity: 'agent-joined', + }, + { + config: { + startUrl: 'http://10.2.2.199:8013/start', + stopUrl: 'http://10.2.2.199:8013/stop', + controlToken: 'secret', + deviceId: 'generic-orin', + address: '10.2.2.199', + }, + createRoomToken: async () => assert.fail('token must not be created'), + requestControl: async (action) => { + actions.push(action); + throw new Error('stale reclaim failed'); + }, + waitForReadiness: async () => assert.fail('readiness must not run'), + } + ), + /stale reclaim failed/ + ); + assert.deepEqual(actions, ['stop']); +}); + +test('Generic pairing attempts rollback when start outcome is uncertain', async () => { + const actions = []; + await assert.rejects( + pairGenericEdgeMedia( + { + roomUrl: 'ws://livekit.test', + roomName: 'room-1', + sessionId: 'session-1', + controlSenderIdentity: 'agent-joined', + }, + { + config: { + startUrl: 'http://10.2.2.199:8013/start', + stopUrl: 'http://10.2.2.199:8013/stop', + controlToken: 'secret', + deviceId: 'generic-orin', + address: '10.2.2.199', + }, + createRoomToken: async () => 'token', + requestControl: async (action) => { + actions.push(action); + if (action === 'start') throw new Error('start response lost'); + }, + waitForReadiness: async () => assert.fail('readiness must not run'), + } + ), + /start response lost/ + ); + assert.deepEqual(actions, ['stop', 'start', 'stop']); +}); + +test('Generic pairing cancellation is material and rolls back the immutable target', async () => { + const actions = []; + let cancelled = false; + await assert.rejects( + pairGenericEdgeMedia( + { + roomUrl: 'ws://livekit.test', + roomName: 'room-1', + sessionId: 'session-1', + controlSenderIdentity: 'agent-joined', + }, + { + config: Object.freeze({ + startUrl: 'http://10.2.2.199:8013/start', + stopUrl: 'http://10.2.2.199:8013/stop', + controlToken: 'secret', + deviceId: 'generic-orin', + address: '10.2.2.199', + }), + createRoomToken: async () => 'token', + requestControl: async (action, _payload, config) => { + actions.push(`${action}:${config.address}`); + }, + waitForReadiness: async () => { + cancelled = true; + }, + isCancelled: () => cancelled, + } + ), + /cancelled/ + ); + assert.deepEqual(actions, ['stop:10.2.2.199', 'start:10.2.2.199', 'stop:10.2.2.199']); +}); + +test('Generic pairing reports an unconfirmed immutable-target rollback as material', async () => { + let stopAttempts = 0; + await assert.rejects( + pairGenericEdgeMedia( + { + roomUrl: 'ws://livekit.test', + roomName: 'room-1', + sessionId: 'session-1', + controlSenderIdentity: 'agent-joined', + }, + { + config: { + startUrl: 'http://10.2.2.199:8013/start', + stopUrl: 'http://10.2.2.199:8013/stop', + controlToken: 'control-secret', + deviceId: 'generic-orin', + address: '10.2.2.199', + }, + createRoomToken: async () => 'room-token', + requestControl: async (action) => { + if (action === 'stop' && ++stopAttempts === 2) { + throw new Error('control-secret rollback body'); + } + }, + waitForReadiness: async () => { + throw new Error('media readiness timeout'); + }, + } + ), + (error) => { + assert.match(error.message, /rollback could not be confirmed/); + assert.doesNotMatch(error.message, /control-secret|room-token|rollback body/); + return true; + } + ); +}); + +test('production target resolution uses one active lease and never reads a static URL', async () => { + const events = []; + const leaseConfig = { deviceId: 'generic-orin', registryDir: '/safe/registry' }; + const target = await resolveGenericEdgeTargetSnapshot({ + environment: { + GENERIC_EDGE_MEDIA_DEVICE_ID: 'generic-orin', + GENERIC_EDGE_MEDIA_ALLOWED_CIDRS: '10.2.0.0/16', + GENERIC_ENDPOINT_REGISTRY_DIR: '/safe/registry', + EDGE_MEDIA_CONTROL_TOKEN: 'control-secret', + EDGE_MEDIA_URL: 'http://attacker.invalid:9999/ignored', + }, + loadLeaseConfig: () => leaseConfig, + resolveLease: async (config) => { + events.push(config); + return { + deviceId: 'generic-orin', + instanceId: '11111111-2222-4333-8444-555555555555', + address: '10.2.2.199', + receivedAt: '2026-08-24T10:00:00.000Z', + expiresAt: '2026-08-24T10:00:45.000Z', + }; + }, + }); + + assert.deepEqual(events, [leaseConfig]); + assert.deepEqual(target, { + startUrl: 'http://10.2.2.199:8013/start', + stopUrl: 'http://10.2.2.199:8013/stop', + controlToken: 'control-secret', + deviceId: 'generic-orin', + address: '10.2.2.199', + }); + assert.equal(Object.isFrozen(target), true); +}); + +test('control requests reconstruct the fixed target and redact response bodies', async () => { + const calls = []; + const target = { + startUrl: 'http://attacker.invalid/start', + stopUrl: 'http://attacker.invalid/stop', + controlToken: 'control-secret', + deviceId: 'generic-orin', + address: '10.2.2.199', + }; + await requestGenericEdgeControl('start', { session_id: 'session-1' }, target, { + fetchImpl: async (url, init) => { + calls.push({ url, init }); + return new Response('{}', { status: 200 }); + }, + }); + + assert.equal(calls[0].url, 'http://10.2.2.199:8013/start'); + assert.equal(calls[0].init.redirect, 'manual'); + assert.equal(calls[0].init.headers['X-Lexvoice-Control-Token'], 'control-secret'); + + await assert.rejects( + requestGenericEdgeControl('stop', { session_id: 'session-1' }, target, { + fetchImpl: async () => new Response('control-secret room-token', { status: 403 }), + }), + (error) => { + assert.match(error.message, /HTTP 403/); + assert.doesNotMatch(error.message, /control-secret|room-token/); + return true; + } + ); +}); + +test('room input token is room-scoped and expires after 15 minutes', async () => { + const token = await createGenericRoomInputToken('voice_assistant_room_session-1', { + LIVEKIT_API_KEY: 'devkey', + LIVEKIT_API_SECRET: 'devsecret-devsecret-devsecret-dev', + }); + const claims = JSON.parse(Buffer.from(token.split('.')[1], 'base64url').toString('utf8')); + + assert.equal(claims.sub, 'room_audio_input'); + assert.ok(claims.exp - Math.floor(Date.now() / 1000) >= 899); + assert.ok(claims.exp - Math.floor(Date.now() / 1000) <= 900); + assert.equal(claims.video.room, 'voice_assistant_room_session-1'); + assert.equal(claims.video.roomJoin, true); + assert.equal(claims.video.canPublish, true); + assert.equal(claims.video.canSubscribe, true); + assert.equal(claims.video.canPublishData, true); +}); + +test('production coordinator dispatches the Agent before resolving and controlling Edge', async () => { + const events = []; + const target = Object.freeze({ + startUrl: 'http://10.2.2.199:8013/start', + stopUrl: 'http://10.2.2.199:8013/stop', + controlToken: 'secret', + deviceId: 'generic-orin', + address: '10.2.2.199', + }); + const result = await coordinateGenericRoomSession( + { + roomUrl: 'ws://livekit.test', + roomName: 'room-1', + sessionId: 'session-1', + agentName: 'lexvoice-generic-agent', + }, + { + dispatchAgent: async () => { + events.push('agent'); + return { agentParticipant: { identity: 'agent-joined' } }; + }, + resolveTarget: async () => { + events.push('lease'); + return target; + }, + pairEndpoint: async (_request, resolvedTarget) => { + events.push(`edge:${resolvedTarget.address}`); + return { deviceId: resolvedTarget.deviceId, address: resolvedTarget.address }; + }, + } + ); + + assert.deepEqual(events, ['agent', 'lease', 'edge:10.2.2.199']); + assert.deepEqual(result.edge, { deviceId: 'generic-orin', address: '10.2.2.199' }); +}); + +test('production coordinator awaits cloud cleanup after a post-Agent pairing failure', async () => { + const events = []; + await assert.rejects( + coordinateGenericRoomSession( + { + roomUrl: 'ws://livekit.test', + roomName: 'room-1', + sessionId: 'session-1', + agentName: 'lexvoice-generic-agent', + }, + { + dispatchAgent: async () => { + events.push('agent'); + return { agentParticipant: { identity: 'agent-joined' } }; + }, + resolveTarget: async () => { + events.push('lease'); + return { + startUrl: 'http://10.2.2.199:8013/start', + stopUrl: 'http://10.2.2.199:8013/stop', + controlToken: 'secret', + deviceId: 'generic-orin', + address: '10.2.2.199', + }; + }, + pairEndpoint: async () => { + events.push('edge'); + throw new Error('media readiness timeout'); + }, + cleanupSession: async () => { + events.push('cloud-cleanup'); + }, + } + ), + /media readiness timeout/ + ); + assert.deepEqual(events, ['agent', 'lease', 'edge', 'cloud-cleanup']); +}); + +test('production coordinator cleans the Agent Room when lease resolution fails closed', async () => { + const events = []; + await assert.rejects( + coordinateGenericRoomSession( + { + roomUrl: 'ws://livekit.test', + roomName: 'room-1', + sessionId: 'session-1', + agentName: 'lexvoice-generic-agent', + }, + { + dispatchAgent: async () => { + events.push('agent'); + return { agentParticipant: { identity: 'agent-joined' } }; + }, + resolveTarget: async () => { + events.push('lease'); + throw new Error('Generic endpoint lease is unavailable'); + }, + pairEndpoint: async () => assert.fail('Edge must not be called without a lease'), + cleanupSession: async () => { + events.push('cloud-cleanup'); + }, + } + ), + /lease is unavailable/ + ); + assert.deepEqual(events, ['agent', 'lease', 'cloud-cleanup']); +}); + +test('production coordinator cleans the Agent Room when dispatch returns no identity', async () => { + const events = []; + await assert.rejects( + coordinateGenericRoomSession( + { + roomUrl: 'ws://livekit.test', + roomName: 'room-1', + sessionId: 'session-1', + agentName: 'lexvoice-generic-agent', + }, + { + dispatchAgent: async () => { + events.push('agent'); + return { agentParticipant: {} }; + }, + resolveTarget: async () => assert.fail('lease must not resolve without Agent identity'), + pairEndpoint: async () => assert.fail('Edge must not start without Agent identity'), + cleanupSession: async () => { + events.push('cloud-cleanup'); + }, + } + ), + /Agent participant is unavailable/ + ); + assert.deepEqual(events, ['agent', 'cloud-cleanup']); +}); + +test('explicit Generic stop resolves the current lease and treats control failure as material', async () => { + const events = []; + await assert.rejects( + stopGenericEdgeMedia( + { roomName: 'room-1', sessionId: 'session-1' }, + { + resolveTarget: async () => { + events.push('lease'); + return { + startUrl: 'http://10.2.2.200:8013/start', + stopUrl: 'http://10.2.2.200:8013/stop', + controlToken: 'secret', + deviceId: 'generic-orin', + address: '10.2.2.200', + }; + }, + requestControl: async (action, payload, target) => { + events.push(`${action}:${target.address}:${payload.session_id}`); + throw new Error('Edge stop returned HTTP 500'); + }, + } + ), + /HTTP 500/ + ); + assert.deepEqual(events, ['lease', 'stop:10.2.2.200:session-1']); +}); diff --git a/tests/generic-endpoint-lease.test.mjs b/tests/generic-endpoint-lease.test.mjs new file mode 100644 index 000000000..08d7f7451 --- /dev/null +++ b/tests/generic-endpoint-lease.test.mjs @@ -0,0 +1,330 @@ +import assert from 'node:assert/strict'; +import { execFile } from 'node:child_process'; +import { randomUUID } from 'node:crypto'; +import { + chmod, + lstat, + mkdir, + readFile, + rename, + symlink, + unlink, + utimes, + writeFile, +} from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { test } from 'node:test'; +import { pathToFileURL } from 'node:url'; +import { promisify } from 'node:util'; +import { + EndpointLeaseConflictError, + EndpointLeaseUnavailableError, + buildGenericEdgeControlUrl, + loadGenericEndpointLeaseConfig, + renewGenericEndpointLease, + resolveActiveGenericEndpointLease, +} from '../lib/generic-endpoint-lease.ts'; + +const execFileAsync = promisify(execFile); + +async function temporaryRegistry() { + const directory = path.join(tmpdir(), `generic-endpoint-lease-${randomUUID()}`); + await mkdir(directory, { mode: 0o700 }); + return directory; +} + +async function publishLeaseRecord(directory, record) { + const destination = path.join(directory, `${record.instanceId}.lease.json`); + const temporary = path.join(directory, `.${record.instanceId}.${randomUUID()}.tmp`); + await writeFile(temporary, JSON.stringify(record), { mode: 0o600 }); + await rename(temporary, destination); + await chmod(destination, 0o600); +} + +function freshLeaseRecord(overrides = {}) { + return { + deviceId: 'generic-orin', + instanceId: '11111111-2222-4333-8444-555555555555', + address: '10.2.2.200', + receivedAt: '2026-08-24T10:01:00.000Z', + expiresAt: '2026-08-24T10:01:45.000Z', + ...overrides, + }; +} + +function config(directory) { + return loadGenericEndpointLeaseConfig({ + GENERIC_EDGE_MEDIA_DEVICE_ID: 'generic-orin', + GENERIC_EDGE_MEDIA_ALLOWED_CIDRS: '10.2.0.0/16,192.168.10.0/24', + GENERIC_ENDPOINT_REGISTRY_DIR: directory, + }); +} + +function heartbeat(overrides = {}) { + return { + deviceId: 'generic-orin', + instanceId: '11111111-2222-4333-8444-555555555555', + hostname: 'orin', + address: '10.2.2.199', + ...overrides, + }; +} + +test('lease renews for 45 seconds and same instance address changes atomically', async () => { + const directory = await temporaryRegistry(); + const leaseConfig = config(directory); + const first = await renewGenericEndpointLease(heartbeat(), leaseConfig, { + now: () => new Date('2026-08-24T10:00:00.000Z'), + }); + assert.equal(first.expiresAt, '2026-08-24T10:00:45.000Z'); + assert.equal((await lstat(directory)).mode & 0o777, 0o700); + + await renewGenericEndpointLease(heartbeat({ address: '10.2.2.200' }), leaseConfig, { + now: () => new Date('2026-08-24T10:00:10.000Z'), + }); + const resolved = await resolveActiveGenericEndpointLease(leaseConfig, { + now: () => new Date('2026-08-24T10:00:44.999Z'), + }); + assert.equal(resolved.address, '10.2.2.200'); + assert.equal(buildGenericEdgeControlUrl(resolved, 'start'), 'http://10.2.2.200:8013/start'); + const persisted = JSON.parse( + await readFile(path.join(directory, `${resolved.instanceId}.lease.json`)) + ); + assert.deepEqual(Object.keys(persisted).sort(), [ + 'address', + 'deviceId', + 'expiresAt', + 'instanceId', + 'receivedAt', + ]); + assert.equal(JSON.stringify(persisted).includes('token'), false); +}); + +test('expired or conflicting leases fail closed', async () => { + const directory = await temporaryRegistry(); + const leaseConfig = config(directory); + await renewGenericEndpointLease(heartbeat(), leaseConfig, { + now: () => new Date('2026-08-24T10:00:00.000Z'), + }); + await assert.rejects( + resolveActiveGenericEndpointLease(leaseConfig, { + now: () => new Date('2026-08-24T10:00:45.000Z'), + }), + EndpointLeaseUnavailableError + ); + + await renewGenericEndpointLease(heartbeat(), leaseConfig, { + now: () => new Date('2026-08-24T11:00:00.000Z'), + }); + await assert.rejects( + renewGenericEndpointLease( + heartbeat({ instanceId: 'aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee' }), + leaseConfig, + { now: () => new Date('2026-08-24T11:00:10.000Z') } + ), + EndpointLeaseConflictError + ); +}); + +test('concurrent different instances cannot both acquire the single-device lease', async () => { + const directory = await temporaryRegistry(); + const leaseConfig = config(directory); + const results = await Promise.allSettled([ + renewGenericEndpointLease(heartbeat(), leaseConfig), + renewGenericEndpointLease( + heartbeat({ instanceId: 'aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee' }), + leaseConfig + ), + ]); + + assert.equal(results.filter((result) => result.status === 'fulfilled').length, 1); + const rejected = results.find((result) => result.status === 'rejected'); + assert.ok(rejected); + assert.ok(rejected.reason instanceof EndpointLeaseConflictError); + await resolveActiveGenericEndpointLease(leaseConfig); +}); + +test('resolver serializes an expired read with same-instance renewal in-process', async () => { + const directory = await temporaryRegistry(); + const leaseConfig = config(directory); + await renewGenericEndpointLease(heartbeat(), leaseConfig, { + now: () => new Date('2026-08-24T10:00:00.000Z'), + }); + const lockPath = path.join(directory, '.generic-endpoint-lease.lock'); + await writeFile(lockPath, `${process.pid}\n`, { mode: 0o600 }); + + const resolving = resolveActiveGenericEndpointLease(leaseConfig, { + now: () => new Date('2026-08-24T10:01:00.000Z'), + }); + await new Promise((resolve) => setTimeout(resolve, 30)); + await publishLeaseRecord(directory, freshLeaseRecord()); + await unlink(lockPath); + + const resolved = await resolving; + assert.equal(resolved.address, '10.2.2.200'); + assert.equal( + ( + await resolveActiveGenericEndpointLease(leaseConfig, { + now: () => new Date('2026-08-24T10:01:01.000Z'), + }) + ).address, + '10.2.2.200' + ); +}); + +test('same-host resolver worker cannot delete a renewal published behind the registry lock', async () => { + const directory = await temporaryRegistry(); + const leaseConfig = config(directory); + await renewGenericEndpointLease(heartbeat(), leaseConfig, { + now: () => new Date('2026-08-24T10:00:00.000Z'), + }); + const lockPath = path.join(directory, '.generic-endpoint-lease.lock'); + await writeFile(lockPath, `${process.pid}\n`, { mode: 0o600 }); + const moduleUrl = pathToFileURL(path.resolve('lib/generic-endpoint-lease.ts')).href; + const environment = { + GENERIC_EDGE_MEDIA_DEVICE_ID: 'generic-orin', + GENERIC_EDGE_MEDIA_ALLOWED_CIDRS: '10.2.0.0/16', + GENERIC_ENDPOINT_REGISTRY_DIR: directory, + }; + const reader = ` + import { loadGenericEndpointLeaseConfig, resolveActiveGenericEndpointLease } from ${JSON.stringify(moduleUrl)}; + const config = loadGenericEndpointLeaseConfig(${JSON.stringify(environment)}); + process.stdout.write(JSON.stringify(await resolveActiveGenericEndpointLease(config, { + now: () => new Date('2026-08-24T10:01:00.000Z'), + }))); + `; + const worker = execFileAsync( + process.execPath, + ['--import', 'tsx', '--input-type=module', '--eval', reader], + { cwd: process.cwd() } + ); + await new Promise((resolve) => setTimeout(resolve, 100)); + await publishLeaseRecord(directory, freshLeaseRecord()); + await unlink(lockPath); + + const { stdout } = await worker; + assert.equal(JSON.parse(stdout).address, '10.2.2.200'); + assert.equal( + ( + await resolveActiveGenericEndpointLease(leaseConfig, { + now: () => new Date('2026-08-24T10:01:01.000Z'), + }) + ).address, + '10.2.2.200' + ); +}); + +test('a stale dead-process registry lock is recovered on the next heartbeat', async () => { + const directory = await temporaryRegistry(); + const lockPath = path.join(directory, '.generic-endpoint-lease.lock'); + await writeFile(lockPath, '999999999\n', { mode: 0o600 }); + const stale = new Date(Date.now() - 11_000); + await utimes(lockPath, stale, stale); + + const lease = await renewGenericEndpointLease(heartbeat(), config(directory)); + + assert.equal(lease.instanceId, heartbeat().instanceId); + await assert.rejects(lstat(lockPath), { code: 'ENOENT' }); +}); + +test('lease validation rejects unsafe addresses and target injection', async () => { + const directory = await temporaryRegistry(); + const leaseConfig = config(directory); + for (const address of [ + '8.8.8.8', + '127.0.0.1', + '169.254.1.2', + '224.0.0.1', + '::1', + 'orin.local', + '10.2.2.199:9000', + '010.002.002.199', + ]) { + await assert.rejects(renewGenericEndpointLease(heartbeat({ address }), leaseConfig), /address/); + } + await assert.rejects( + renewGenericEndpointLease(heartbeat({ address: '172.16.1.2' }), leaseConfig), + /allowed CIDR/ + ); +}); + +test('registry rejects symlink and unsafe permissions', async () => { + const parent = await temporaryRegistry(); + const real = path.join(parent, 'real'); + const linked = path.join(parent, 'linked'); + await mkdir(real, { mode: 0o700 }); + await symlink(real, linked); + await assert.rejects( + renewGenericEndpointLease(heartbeat(), config(linked)), + /registry directory/ + ); + + const unsafe = path.join(parent, 'unsafe'); + await mkdir(unsafe, { mode: 0o755 }); + await chmod(unsafe, 0o755); + await assert.rejects(renewGenericEndpointLease(heartbeat(), config(unsafe)), /0700/); +}); + +test('corrupt and unsafe lease files fail closed', async () => { + const directory = await temporaryRegistry(); + const leaseConfig = config(directory); + const corrupt = path.join(directory, '11111111-2222-4333-8444-555555555555.lease.json'); + await writeFile(corrupt, '{}', { mode: 0o600 }); + await assert.rejects(resolveActiveGenericEndpointLease(leaseConfig), /lease record/); + await writeFile(corrupt, JSON.stringify(heartbeat()), { mode: 0o600 }); + await chmod(corrupt, 0o644); + await assert.rejects(resolveActiveGenericEndpointLease(leaseConfig), /0600/); +}); + +test('configuration has no static Generic URL and rejects unsupported multi-host mode', () => { + const directory = '/tmp/generic-lease-test'; + assert.throws(() => config('relative/path'), /absolute/); + assert.throws( + () => + loadGenericEndpointLeaseConfig({ + GENERIC_EDGE_MEDIA_DEVICE_ID: 'generic-orin', + GENERIC_EDGE_MEDIA_ALLOWED_CIDRS: '10.2.0.0/16', + GENERIC_ENDPOINT_REGISTRY_DIR: directory, + GENERIC_ENDPOINT_MULTI_HOST: '1', + }), + /multi-host/ + ); + const resolved = config(directory); + assert.equal('edgeMediaUrl' in resolved, false); +}); + +test('fresh same-host processes share the persisted active lease', async () => { + const directory = await temporaryRegistry(); + const moduleUrl = pathToFileURL(path.resolve('lib/generic-endpoint-lease.ts')).href; + const environment = { + GENERIC_EDGE_MEDIA_DEVICE_ID: 'generic-orin', + GENERIC_EDGE_MEDIA_ALLOWED_CIDRS: '10.2.0.0/16', + GENERIC_ENDPOINT_REGISTRY_DIR: directory, + }; + const writer = ` + import { loadGenericEndpointLeaseConfig, renewGenericEndpointLease } from ${JSON.stringify(moduleUrl)}; + const config = loadGenericEndpointLeaseConfig(${JSON.stringify(environment)}); + await renewGenericEndpointLease(${JSON.stringify(heartbeat())}, config); + `; + await execFileAsync( + process.execPath, + ['--import', 'tsx', '--input-type=module', '--eval', writer], + { cwd: process.cwd() } + ); + + const parentRead = await resolveActiveGenericEndpointLease(config(directory)); + assert.equal(parentRead.address, '10.2.2.199'); + + const reader = ` + import { loadGenericEndpointLeaseConfig, resolveActiveGenericEndpointLease } from ${JSON.stringify(moduleUrl)}; + const config = loadGenericEndpointLeaseConfig(${JSON.stringify(environment)}); + process.stdout.write(JSON.stringify(await resolveActiveGenericEndpointLease(config))); + `; + const { stdout } = await execFileAsync( + process.execPath, + ['--import', 'tsx', '--input-type=module', '--eval', reader], + { cwd: process.cwd() } + ); + assert.equal(JSON.parse(stdout).instanceId, heartbeat().instanceId); +}); diff --git a/tests/project-config.test.mjs b/tests/project-config.test.mjs index d2194d0d3..218120a0d 100644 --- a/tests/project-config.test.mjs +++ b/tests/project-config.test.mjs @@ -26,6 +26,35 @@ test('README matches the documented LexVoice environment source', async () => { assert.doesNotMatch(readme, /copy `\.env\.example`/i); }); +test('integrated Generic configuration is delegated to LexVoice', async () => { + const readme = await readFile('README.md', 'utf8'); + const envExample = await readFile('.env.example', 'utf8'); + + for (const name of [ + 'INPUT_SOURCE', + 'AGENT_NAME', + 'LIVEKIT_URL', + 'LIVEKIT_API_KEY', + 'LIVEKIT_API_SECRET', + 'VIDEO_PROCESSOR_URL', + 'ENDPOINT_CONNECTIVITY_TOKEN', + 'EDGE_MEDIA_CONTROL_TOKEN', + 'GENERIC_EDGE_MEDIA_DEVICE_ID', + 'GENERIC_EDGE_MEDIA_ALLOWED_CIDRS', + 'GENERIC_ENDPOINT_REGISTRY_DIR', + ]) { + assert.doesNotMatch(envExample, new RegExp(`^${name}=`, 'm')); + } + + assert.match(readme, /Generic configuration.*LexVoice unified Mac startup/i); + assert.match(readme, /do not add.*\.env\.example.*\.env\.local/i); + assert.doesNotMatch(readme, /For Generic endpoint discovery, configure the server-only/); + assert.match(readme, /Agent joins first/); + assert.match(readme, /room_audio_input.*room_audio/); + assert.match(readme, /room_audio_input.*room_video_raw/); + assert.match(readme, /room_video_input.*room_video/); +}); + test('avatar filtering excludes the current room video input identity', async () => { const source = await readFile('hooks/useSmartVoiceAssistant.ts', 'utf8'); diff --git a/tests/session-dispatch-readiness.test.mjs b/tests/session-dispatch-readiness.test.mjs index 7977a5d74..ab5ce4120 100644 --- a/tests/session-dispatch-readiness.test.mjs +++ b/tests/session-dispatch-readiness.test.mjs @@ -7,6 +7,9 @@ const { ParticipantInfo_Kind, ParticipantInfo_State, TrackType } = await import( const { AGENT_SESSION_READY_ATTRIBUTE, findReusableAgentParticipant } = await import( '../lib/session-dispatch-readiness.ts' ); +const { waitForExistingRoomSessionReadiness } = await import( + '../app/api/session/session-dispatch-service.ts' +); function participant({ identity, @@ -85,7 +88,7 @@ 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', () => { +test('room input readiness requires the exact active unmuted audio, raw-video, and processed-video tracks', () => { const agent = participant({ identity: 'agent-AJ_running', kind: ParticipantInfo_Kind.AGENT, @@ -93,18 +96,107 @@ test('prewarm readiness requires both room input participants without requiring }); const participants = [ agent, - participant({ identity: 'room_audio_input' }), - participant({ identity: 'room_video_input' }), + participant({ + identity: 'room_audio_input', + tracks: [ + { name: 'room_audio', type: TrackType.AUDIO, muted: false }, + { name: 'room_video_raw', type: TrackType.VIDEO, muted: false }, + ], + }), + participant({ + identity: 'room_video_input', + tracks: [{ name: 'room_video', type: TrackType.VIDEO, muted: false }], + }), ]; assert.equal( findReusableAgentParticipant(participants, 'frontdesk-browser-agent', { - requireRoomInputParticipantsReady: true, + requireExactRoomInputTracksReady: true, }), agent ); }); +test('exact Generic readiness keeps room_video fixed despite legacy track configuration', () => { + const previous = process.env.NEXT_PUBLIC_ROOM_VISION_TRACK_NAME; + process.env.NEXT_PUBLIC_ROOM_VISION_TRACK_NAME = 'configured_other_video'; + 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', + tracks: [ + { name: 'room_audio', type: TrackType.AUDIO, muted: false }, + { name: 'room_video_raw', type: TrackType.VIDEO, muted: false }, + ], + }), + participant({ + identity: 'room_video_input', + tracks: [{ name: 'room_video', type: TrackType.VIDEO, muted: false }], + }), + ]; + + try { + assert.equal( + findReusableAgentParticipant(participants, 'frontdesk-browser-agent', { + requireExactRoomInputTracksReady: true, + }), + agent + ); + } finally { + if (previous === undefined) delete process.env.NEXT_PUBLIC_ROOM_VISION_TRACK_NAME; + else process.env.NEXT_PUBLIC_ROOM_VISION_TRACK_NAME = previous; + } +}); + +test('room input readiness rejects active participants with missing, muted, or misnamed tracks', () => { + const agent = participant({ + identity: 'agent-AJ_running', + kind: ParticipantInfo_Kind.AGENT, + attributes: { 'lk.agent.name': 'frontdesk-browser-agent' }, + }); + for (const roomInputs of [ + [participant({ identity: 'room_audio_input' }), participant({ identity: 'room_video_input' })], + [ + participant({ + identity: 'room_audio_input', + tracks: [ + { name: 'room_audio', type: TrackType.AUDIO, muted: true }, + { name: 'room_video_raw', type: TrackType.VIDEO, muted: false }, + ], + }), + participant({ + identity: 'room_video_input', + tracks: [{ name: 'room_video', type: TrackType.VIDEO, muted: false }], + }), + ], + [ + participant({ + identity: 'room_audio_input', + tracks: [ + { name: 'room_audio', type: TrackType.AUDIO, muted: false }, + { name: 'wrong_raw_video', type: TrackType.VIDEO, muted: false }, + ], + }), + participant({ + identity: 'room_video_input', + tracks: [{ name: 'room_video', type: TrackType.VIDEO, muted: false }], + }), + ], + ]) { + assert.equal( + findReusableAgentParticipant([agent, ...roomInputs], 'frontdesk-browser-agent', { + requireExactRoomInputTracksReady: true, + }), + null + ); + } +}); + test('prewarm can require the full agent session ready marker', () => { const agent = participant({ identity: 'agent-AJ_running', @@ -168,3 +260,66 @@ test('dispatch does not reuse disconnected agents', () => { assert.equal(findReusableAgentParticipant(participants, 'frontdesk-browser-agent'), null); }); + +test('post-Edge readiness polls exact tracks without creating another Agent dispatch', async () => { + const agent = participant({ + identity: 'agent-AJ_running', + kind: ParticipantInfo_Kind.AGENT, + attributes: { 'lk.agent.name': 'lexvoice-generic-agent' }, + }); + const ready = [ + agent, + participant({ + identity: 'room_audio_input', + tracks: [ + { name: 'room_audio', type: TrackType.AUDIO, muted: false }, + { name: 'room_video_raw', type: TrackType.VIDEO, muted: false }, + ], + }), + participant({ + identity: 'room_video_input', + tracks: [{ name: 'room_video', type: TrackType.VIDEO, muted: false }], + }), + ]; + const participantSnapshots = [[agent], ready]; + let sleeps = 0; + + const result = await waitForExistingRoomSessionReadiness( + { + roomName: 'room-1', + agentName: 'lexvoice-generic-agent', + }, + { + roomClient: { + listParticipants: async () => participantSnapshots.shift(), + }, + timeoutMs: 1_000, + pollMs: 1, + sleep: async () => { + sleeps += 1; + }, + } + ); + + assert.deepEqual(result, { identity: 'agent-AJ_running' }); + assert.equal(sleeps, 1); +}); + +test('post-Edge readiness observes cancellation while polling', async () => { + let cancelled = false; + await assert.rejects( + waitForExistingRoomSessionReadiness( + { roomName: 'room-1', agentName: 'lexvoice-generic-agent' }, + { + roomClient: { listParticipants: async () => [] }, + timeoutMs: 1_000, + pollMs: 1, + sleep: async () => { + cancelled = true; + }, + isCancelled: () => cancelled, + } + ), + /cancelled/ + ); +}); diff --git a/tests/session-start-dispatch.test.mjs b/tests/session-start-dispatch.test.mjs index dca58fa7e..471edadff 100644 --- a/tests/session-start-dispatch.test.mjs +++ b/tests/session-start-dispatch.test.mjs @@ -2,6 +2,10 @@ import assert from 'node:assert/strict'; import { readFile } from 'node:fs/promises'; import { test } from 'node:test'; +const { formatSessionDispatchError, runSessionDispatch } = await import( + '../app/api/session/generic-session-dispatch.ts' +); + test('connection details route does not dispatch agents while generating tokens', async () => { const routeSource = await readFile( new URL('../app/api/connection-details/route.ts', import.meta.url), @@ -66,6 +70,108 @@ test('session dispatch route retries explicit agent dispatch after the browser j ); }); +test('Generic dispatch is selected only by server config and performs Agent then dynamic Edge pairing', async () => { + const events = []; + const result = await runSessionDispatch( + { + roomName: 'voice_assistant_room_session-1', + sessionId: 'session-1', + agentName: 'lexvoice-generic-agent', + requireRoomVideoInputReady: false, + endpointUrl: 'http://attacker.invalid:9999/start', + disablePairing: true, + }, + { + environment: { + INPUT_SOURCE: 'generic', + LIVEKIT_URL: 'ws://localhost:7818', + LIVEKIT_BROWSER_URL: ' ws://10.2.77.108:7818 ', + }, + dispatchAgent: async () => { + events.push('agent'); + return { agentParticipant: { identity: 'agent-joined' } }; + }, + resolveTarget: async () => { + events.push('lease'); + return { + startUrl: 'http://10.2.2.199:8013/start', + stopUrl: 'http://10.2.2.199:8013/stop', + controlToken: 'secret', + deviceId: 'generic-orin', + address: '10.2.2.199', + }; + }, + pairEndpoint: async (request, target) => { + events.push(`edge:${target.address}:${request.controlSenderIdentity}:${request.roomUrl}`); + return { deviceId: target.deviceId, address: target.address }; + }, + } + ); + + assert.deepEqual(events, [ + 'agent', + 'lease', + 'edge:10.2.2.199:agent-joined:ws://10.2.77.108:7818', + ]); + assert.equal(result.edge.address, '10.2.2.199'); +}); + +test('Generic endpoint pairing falls back to the internal LiveKit URL', async () => { + let pairedRoomUrl; + await runSessionDispatch( + { + roomName: 'voice_assistant_room_session-fallback', + sessionId: 'session-fallback', + agentName: 'lexvoice-generic-agent', + }, + { + environment: { INPUT_SOURCE: 'generic', LIVEKIT_URL: ' ws://localhost:7818 ' }, + dispatchAgent: async () => ({ agentParticipant: { identity: 'agent-joined' } }), + resolveTarget: async () => ({ + startUrl: 'http://10.2.2.199:8013/start', + stopUrl: 'http://10.2.2.199:8013/stop', + controlToken: 'secret', + deviceId: 'generic-orin', + address: '10.2.2.199', + }), + pairEndpoint: async (request, target) => { + pairedRoomUrl = request.roomUrl; + return { deviceId: target.deviceId, address: target.address }; + }, + } + ); + + assert.equal(pairedRoomUrl, 'ws://localhost:7818'); +}); + +test('browser dispatch never resolves or controls a Generic endpoint', async () => { + const result = await runSessionDispatch( + { + roomName: 'voice_assistant_room_session-2', + sessionId: 'session-2', + agentName: 'lexvoice-browser-agent', + endpointUrl: 'http://10.2.2.199:8013/start', + }, + { + environment: { INPUT_SOURCE: 'browser' }, + dispatchAgent: async () => ({ agentParticipant: { identity: 'agent-browser' } }), + resolveTarget: async () => assert.fail('browser must not resolve an endpoint lease'), + pairEndpoint: async () => assert.fail('browser must not control an endpoint'), + } + ); + + assert.equal(result.agentParticipant.identity, 'agent-browser'); +}); + +test('Generic dispatch errors are fixed and do not expose server paths or credentials', () => { + const message = formatSessionDispatchError(new Error('/owner/lease/control-secret room-token'), { + INPUT_SOURCE: 'generic', + }); + + assert.equal(message, 'Generic session startup failed'); + assert.doesNotMatch(message, /owner|secret|token/); +}); + test('session dispatch retry backs off between repeated attempts', async () => { const serviceSource = await readFile( new URL('../app/api/session/session-dispatch-service.ts', import.meta.url), diff --git a/tests/session-stop.test.mjs b/tests/session-stop.test.mjs index 4ee4c06a5..d11c94169 100644 --- a/tests/session-stop.test.mjs +++ b/tests/session-stop.test.mjs @@ -1,8 +1,15 @@ import assert from 'node:assert/strict'; -import { readFile } from 'node:fs/promises'; +import { mkdtemp, readFile, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; import { test } from 'node:test'; +import { cleanupFailedGenericRoomSession } from '../app/api/session/generic-session-dispatch.ts'; import { POST as stopSession } from '../app/api/session/stop/route.ts'; import { readAgentWorkerStateFromLog } from '../lib/agent-worker-readiness.ts'; +import { + loadGenericEndpointLeaseConfig, + renewGenericEndpointLease, +} from '../lib/generic-endpoint-lease.ts'; import { executeRoomInputStopsSequentially, resolveLiveKitHttpUrl, @@ -155,6 +162,125 @@ test('session stop reports invalid split media configuration and continues room } }); +test('Generic stop uses the current lease target and ignores static EDGE_MEDIA_URL', async () => { + const previousEnv = { ...process.env }; + const previousFetch = globalThis.fetch; + const registryDir = await mkdtemp(path.join(tmpdir(), 'generic-stop-lease-')); + const calls = []; + Object.assign(process.env, { + INPUT_SOURCE: 'generic', + VIDEO_PROCESSOR_URL: 'http://127.0.0.1:8014/start', + EDGE_MEDIA_URL: 'http://attacker.invalid:9999/start', + GENERIC_EDGE_MEDIA_DEVICE_ID: 'generic-orin', + GENERIC_EDGE_MEDIA_ALLOWED_CIDRS: '10.2.0.0/16', + GENERIC_ENDPOINT_REGISTRY_DIR: registryDir, + EDGE_MEDIA_CONTROL_TOKEN: 'control-secret', + }); + delete process.env.LIVEKIT_URL; + delete process.env.LIVEKIT_API_KEY; + delete process.env.LIVEKIT_API_SECRET; + delete process.env.LEXVOICE_RUN_LOG_DIR; + await renewGenericEndpointLease( + { + deviceId: 'generic-orin', + instanceId: '11111111-2222-4333-8444-555555555555', + hostname: 'orin', + address: '10.2.2.199', + }, + loadGenericEndpointLeaseConfig(process.env) + ); + globalThis.fetch = async (url, init) => { + calls.push({ url: String(url), init }); + return new Response('{}', { status: 200 }); + }; + + try { + const response = await stopSession( + new Request('http://localhost/api/session/stop', { + method: 'POST', + body: JSON.stringify({ + sessionId: '00000000-0000-4000-8000-000000000021', + wait: true, + }), + }) + ); + const payload = await response.json(); + + assert.equal(response.status, 200); + assert.equal(payload.status, 'stopped'); + assert.deepEqual( + calls.map((call) => call.url), + ['http://127.0.0.1:8014/stop', 'http://10.2.2.199:8013/stop'] + ); + assert.equal(calls[1].init.headers['X-Lexvoice-Control-Token'], 'control-secret'); + assert.equal( + calls.some((call) => call.url.includes('attacker.invalid')), + false + ); + } finally { + globalThis.fetch = previousFetch; + restoreEnv(previousEnv); + await rm(registryDir, { recursive: true }); + } +}); + +test('failed Generic startup cleanup stops cloud input without re-resolving Edge', async () => { + const previousEnv = { ...process.env }; + const previousFetch = globalThis.fetch; + const calls = []; + Object.assign(process.env, { + INPUT_SOURCE: 'generic', + VIDEO_PROCESSOR_URL: 'http://127.0.0.1:8014/start', + }); + delete process.env.LIVEKIT_URL; + delete process.env.LIVEKIT_API_KEY; + delete process.env.LIVEKIT_API_SECRET; + delete process.env.LEXVOICE_RUN_LOG_DIR; + delete process.env.GENERIC_ENDPOINT_REGISTRY_DIR; + globalThis.fetch = async (url) => { + calls.push(String(url)); + return new Response('{}', { status: 200 }); + }; + + try { + await cleanupFailedGenericRoomSession( + 'voice_assistant_room_00000000-0000-4000-8000-000000000022', + '00000000-0000-4000-8000-000000000022' + ); + assert.deepEqual(calls, ['http://127.0.0.1:8014/stop']); + } finally { + globalThis.fetch = previousFetch; + restoreEnv(previousEnv); + } +}); + +test('failed Generic startup treats an unconfirmed cloud input stop as material', async () => { + const previousEnv = { ...process.env }; + const previousFetch = globalThis.fetch; + Object.assign(process.env, { + INPUT_SOURCE: 'generic', + VIDEO_PROCESSOR_URL: 'http://127.0.0.1:8014/start', + }); + delete process.env.LIVEKIT_URL; + delete process.env.LIVEKIT_API_KEY; + delete process.env.LIVEKIT_API_SECRET; + delete process.env.LEXVOICE_RUN_LOG_DIR; + globalThis.fetch = async () => new Response('{}', { status: 503 }); + + try { + await assert.rejects( + cleanupFailedGenericRoomSession( + 'voice_assistant_room_00000000-0000-4000-8000-000000000023', + '00000000-0000-4000-8000-000000000023' + ), + /cloud cleanup could not be confirmed/ + ); + } finally { + globalThis.fetch = previousFetch; + 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'; @@ -202,12 +328,13 @@ test('session stop route stops room input before deleting the room', async () => 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.match(stopUrlResolverSource, /isGenericEndpointPairingEnabled/); assert.ok(stopRoomInputSource, 'stopRoomInput should be defined'); assert.match(stopRoomInputSource, /executeRoomInputStopsSequentially\(stopUrls,/); + assert.match(stopRoomInputSource, /stopGenericEdgeMedia/); assert.match( cleanupSource, - /const roomInputResults = await stopRoomInput\(roomName, sessionId\);[\s\S]*const liveKitRoomResult = await deleteLiveKitRoom\(roomName\);/ + /const roomInputResults = await stopRoomInput\(roomName, sessionId, options\);[\s\S]*const liveKitRoomResult = await deleteLiveKitRoom\(roomName\);/ ); }); @@ -248,7 +375,7 @@ test('session stop route deletes the LiveKit room after the dispatch barrier', a ); assert.match(routeSource, /await waitForPendingDispatches\(roomName, sessionId\)/); - assert.match(routeSource, /await stopRoomInput\(roomName, sessionId\)/); + assert.match(routeSource, /await stopRoomInput\(roomName, sessionId, options\)/); assert.match(routeSource, /deleteLiveKitRoom\(roomName\)/); }); From 74a22c33b55cb32bd7c0877ba09da3fad7660273 Mon Sep 17 00:00:00 2001 From: why-tomato Date: Tue, 25 Aug 2026 21:04:09 +0800 Subject: [PATCH 2/2] fix: clean up partial Generic dispatches --- README.md | 2 +- app/api/session/generic-edge-media-pairing.ts | 2 +- tests/connection-details.test.mjs | 5 +--- tests/generic-edge-media.test.mjs | 27 +++++++++++++++++++ 4 files changed, 30 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 0eee36cd2..a2331a6b8 100644 --- a/README.md +++ b/README.md @@ -107,7 +107,7 @@ server directly: ```bash pnpm install -pnpm exec next dev --hostname 0.0.0.0 --port 3000 +pnpm dev ``` And open http://localhost:3000 in your browser. diff --git a/app/api/session/generic-edge-media-pairing.ts b/app/api/session/generic-edge-media-pairing.ts index b2d718092..a489f8378 100644 --- a/app/api/session/generic-edge-media-pairing.ts +++ b/app/api/session/generic-edge-media-pairing.ts @@ -222,8 +222,8 @@ export async function coordinateGenericRoomSession( request: GenericSessionCoordinatorRequest, dependencies: GenericSessionCoordinatorDependencies ) { - const dispatch = await dependencies.dispatchAgent(); try { + const dispatch = await dependencies.dispatchAgent(); const controlSenderIdentity = dispatch.agentParticipant?.identity?.trim(); if (!controlSenderIdentity) { throw new Error('Generic session Agent participant is unavailable'); diff --git a/tests/connection-details.test.mjs b/tests/connection-details.test.mjs index 7e6ef6f67..2c1ffb481 100644 --- a/tests/connection-details.test.mjs +++ b/tests/connection-details.test.mjs @@ -94,10 +94,7 @@ test('connection details advertise the explicit browser LiveKit URL without Host process.env.LIVEKIT_BROWSER_URL = 'ws://10.2.77.108:7818'; process.env.LIVEKIT_URL = 'ws://127.0.0.1:7818'; const { POST } = await import( - new URL( - '../app/api/connection-details/route.ts?browser-url-contract', - import.meta.url - ).href + new URL('../app/api/connection-details/route.ts?browser-url-contract', import.meta.url).href ); const response = await POST( diff --git a/tests/generic-edge-media.test.mjs b/tests/generic-edge-media.test.mjs index 74d38b84b..e24ab5329 100644 --- a/tests/generic-edge-media.test.mjs +++ b/tests/generic-edge-media.test.mjs @@ -381,6 +381,33 @@ test('production coordinator awaits cloud cleanup after a post-Agent pairing fai assert.deepEqual(events, ['agent', 'lease', 'edge', 'cloud-cleanup']); }); +test('production coordinator cleans the Agent Room when dispatch fails after a partial create', async () => { + const events = []; + await assert.rejects( + coordinateGenericRoomSession( + { + roomUrl: 'ws://livekit.test', + roomName: 'room-1', + sessionId: 'session-1', + agentName: 'lexvoice-generic-agent', + }, + { + dispatchAgent: async () => { + events.push('agent'); + throw new Error('dispatch response lost'); + }, + resolveTarget: async () => assert.fail('lease must not resolve after dispatch failure'), + pairEndpoint: async () => assert.fail('Edge must not start after dispatch failure'), + cleanupSession: async () => { + events.push('cloud-cleanup'); + }, + } + ), + /dispatch response lost/ + ); + assert.deepEqual(events, ['agent', 'cloud-cleanup']); +}); + test('production coordinator cleans the Agent Room when lease resolution fails closed', async () => { const events = []; await assert.rejects(