From f2b8d4c3d0f524d56bd366bd9d3b40025a94164a Mon Sep 17 00:00:00 2001 From: guowei Date: Mon, 3 Aug 2026 14:54:10 +0800 Subject: [PATCH 1/4] feat: enforce browser audio media gate --- hooks/useBrowserSourceClient.ts | 527 ++++++----- lib/browser-audio-gate-device.ts | 153 ++++ lib/browser-source-runtime-lifecycle.ts | 113 +++ lib/livekit-media-gate.ts | 276 ++++++ lib/media-control-protocol.ts | 472 ++++++++++ lib/media-gate-executor.ts | 722 +++++++++++++++ tests/browser-audio-gate-device.test.mjs | 422 +++++++++ tests/browser-media-gate-wiring.test.mjs | 81 ++ .../browser-source-runtime-lifecycle.test.mjs | 102 +++ tests/livekit-media-gate.test.mjs | 512 +++++++++++ tests/media-control-protocol.test.mjs | 192 ++++ tests/media-gate-executor.test.mjs | 828 ++++++++++++++++++ 12 files changed, 4200 insertions(+), 200 deletions(-) create mode 100644 lib/browser-audio-gate-device.ts create mode 100644 lib/browser-source-runtime-lifecycle.ts create mode 100644 lib/livekit-media-gate.ts create mode 100644 lib/media-control-protocol.ts create mode 100644 lib/media-gate-executor.ts create mode 100644 tests/browser-audio-gate-device.test.mjs create mode 100644 tests/browser-media-gate-wiring.test.mjs create mode 100644 tests/browser-source-runtime-lifecycle.test.mjs create mode 100644 tests/livekit-media-gate.test.mjs create mode 100644 tests/media-control-protocol.test.mjs create mode 100644 tests/media-gate-executor.test.mjs diff --git a/hooks/useBrowserSourceClient.ts b/hooks/useBrowserSourceClient.ts index 778715168..34563cdae 100644 --- a/hooks/useBrowserSourceClient.ts +++ b/hooks/useBrowserSourceClient.ts @@ -11,7 +11,16 @@ import { createLocalVideoTrack, } from 'livekit-client'; import type { AppConfig } from '@/app-config'; +import { BrowserAudioGateDevice } from '@/lib/browser-audio-gate-device'; +import { + detachCurrentRuntime, + isCurrentRuntime, + replaceRuntimeAudioBinding, + runOwnedRuntimeStart, +} from '@/lib/browser-source-runtime-lifecycle'; import { startMediaTrackVadObserver } from '@/lib/frontend-vad-observer'; +import { LiveKitMediaGateAdapter } from '@/lib/livekit-media-gate'; +import { MediaGateExecutor } from '@/lib/media-gate-executor'; import { FRONTEND_EVENTS, OBSERVABILITY_ATTRS, @@ -23,6 +32,7 @@ const BROWSER_VIDEO_TRACK_NAME = 'browser_video_track'; const DEFAULT_BROWSER_MEDIA_STREAM_NAME = 'browser_input'; const BROWSER_VIDEO_DEFAULT_ENABLED = true; const BROWSER_VIDEO_STATS_INTERVAL_MS = 5000; +const BROWSER_MEDIA_GATE_MAX_OPEN_LEASE_MS = 3000; const BROWSER_AUDIO_CONSTRAINTS: MediaTrackConstraints = { echoCancellation: true, noiseSuppression: true, @@ -40,6 +50,15 @@ interface BrowserSourceRuntime { videoStatsTimer: number | null; previousVideoStats: BrowserVideoStatsSnapshot | null; audioObserverStop: (() => Promise) | null; + audioPublishPromise: Promise | null; + audioGate: BrowserAudioGateRuntime | null; + stopPromise: Promise | null; +} + +interface BrowserAudioGateRuntime { + device: BrowserAudioGateDevice; + executor: MediaGateExecutor; + adapter: LiveKitMediaGateAdapter; } interface BrowserVideoStatsSnapshot { @@ -115,148 +134,197 @@ export function useBrowserSourceClient( [appConfig.observabilityEnabled, room] ); - const ensureAudioPublished = useCallback(async () => { - const runtime = runtimeRef.current; - if (!audioConfigured || !runtime || runtime.audioTrack || !runtime.audioEnabled) { - return; - } + const ensureAudioPublished = useCallback( + async (runtime: BrowserSourceRuntime) => { + if ( + !audioConfigured || + !isCurrentRuntime(runtimeRef, runtime) || + runtime.audioTrack || + !runtime.audioEnabled + ) { + return; + } + if (runtime.audioPublishPromise) { + await runtime.audioPublishPromise; + return; + } - const vadAttributes: Record = { - [OBSERVABILITY_ATTRS.FRONTEND_AUDIO_DIRECTION]: 'input', - [OBSERVABILITY_ATTRS.FRONTEND_AUDIO_PROBE]: 'vad-web', - [OBSERVABILITY_ATTRS.TRACK_NAME]: BROWSER_AUDIO_TRACK_NAME, - [OBSERVABILITY_ATTRS.TRACK_SID]: null, - [OBSERVABILITY_ATTRS.TRACK_STREAM_NAME]: browserMediaStreamName, - }; - const audioTrack = await createLocalAudioTrack( - buildAudioCaptureOptions(audioDeviceIdRef.current) - ); - const captureTrack = audioTrack.mediaStreamTrack; - audioTrack.mediaStreamTrack.enabled = runtime.audioEnabled; - - try { - const publication = await room.localParticipant.publishTrack(audioTrack, { - name: BROWSER_AUDIO_TRACK_NAME, - source: Track.Source.Microphone, - stream: browserMediaStreamName, - }); - runtime.audioTrack = audioTrack; - runtime.audioPublication = publication; - runtime.audioObserverStop = null; - vadAttributes[OBSERVABILITY_ATTRS.TRACK_SID] = publication.trackSid || null; - recordFrontendObservability(FRONTEND_EVENTS.BROWSER_AUDIO_TRACK_PUBLISHED, { - [OBSERVABILITY_ATTRS.TRACK_NAME]: BROWSER_AUDIO_TRACK_NAME, - [OBSERVABILITY_ATTRS.TRACK_SID]: publication.trackSid || null, - [OBSERVABILITY_ATTRS.TRACK_STREAM_NAME]: browserMediaStreamName, - }); - if (appConfig.observabilityEnabled) { - void startMediaTrackVadObserver({ - mediaStreamTrack: captureTrack, - onSpeechStart: (event) => { - recordFrontendObservability( - FRONTEND_EVENTS.BROWSER_AUDIO_VAD_SPEECH_STARTED, - { - ...vadAttributes, - [OBSERVABILITY_ATTRS.VAD_PROVIDER]: event.provider, - [OBSERVABILITY_ATTRS.VAD_MODEL]: event.model, + const publishPromise = (async () => { + const vadAttributes: Record = { + [OBSERVABILITY_ATTRS.FRONTEND_AUDIO_DIRECTION]: 'input', + [OBSERVABILITY_ATTRS.FRONTEND_AUDIO_PROBE]: 'vad-web', + [OBSERVABILITY_ATTRS.TRACK_NAME]: BROWSER_AUDIO_TRACK_NAME, + [OBSERVABILITY_ATTRS.TRACK_SID]: null, + [OBSERVABILITY_ATTRS.TRACK_STREAM_NAME]: browserMediaStreamName, + }; + const audioTrack = await createLocalAudioTrack( + buildAudioCaptureOptions(audioDeviceIdRef.current) + ); + const captureTrack = audioTrack.mediaStreamTrack; + audioTrack.mediaStreamTrack.enabled = false; + + try { + await audioTrack.mute(); + if (runtimeRef.current !== runtime || !runtime.audioEnabled) { + audioTrack.stop(); + return; + } + const publication = await room.localParticipant.publishTrack(audioTrack, { + name: BROWSER_AUDIO_TRACK_NAME, + source: Track.Source.Microphone, + stream: browserMediaStreamName, + }); + if (runtimeRef.current !== runtime || !runtime.audioEnabled) { + await room.localParticipant.unpublishTrack(audioTrack, true).catch(() => undefined); + audioTrack.stop(); + return; + } + runtime.audioTrack = audioTrack; + runtime.audioPublication = publication; + runtime.audioObserverStop = null; + vadAttributes[OBSERVABILITY_ATTRS.TRACK_SID] = publication.trackSid || null; + recordFrontendObservability(FRONTEND_EVENTS.BROWSER_AUDIO_TRACK_PUBLISHED, { + [OBSERVABILITY_ATTRS.TRACK_NAME]: BROWSER_AUDIO_TRACK_NAME, + [OBSERVABILITY_ATTRS.TRACK_SID]: publication.trackSid || null, + [OBSERVABILITY_ATTRS.TRACK_STREAM_NAME]: browserMediaStreamName, + }); + if (appConfig.observabilityEnabled) { + void startMediaTrackVadObserver({ + mediaStreamTrack: captureTrack, + onSpeechStart: (event) => { + recordFrontendObservability( + FRONTEND_EVENTS.BROWSER_AUDIO_VAD_SPEECH_STARTED, + { + ...vadAttributes, + [OBSERVABILITY_ATTRS.VAD_PROVIDER]: event.provider, + [OBSERVABILITY_ATTRS.VAD_MODEL]: event.model, + }, + { wallTimeUnixMs: event.timestampMs } + ); }, - { wallTimeUnixMs: event.timestampMs } - ); - }, - onSpeechEnd: (event) => { - recordFrontendObservability( - FRONTEND_EVENTS.BROWSER_AUDIO_VAD_SPEECH_ENDED, - { - ...vadAttributes, - [OBSERVABILITY_ATTRS.VAD_PROVIDER]: event.provider, - [OBSERVABILITY_ATTRS.VAD_MODEL]: event.model, - [OBSERVABILITY_ATTRS.VAD_AUDIO_DURATION_MS]: event.audioDurationMs ?? null, + onSpeechEnd: (event) => { + recordFrontendObservability( + FRONTEND_EVENTS.BROWSER_AUDIO_VAD_SPEECH_ENDED, + { + ...vadAttributes, + [OBSERVABILITY_ATTRS.VAD_PROVIDER]: event.provider, + [OBSERVABILITY_ATTRS.VAD_MODEL]: event.model, + [OBSERVABILITY_ATTRS.VAD_AUDIO_DURATION_MS]: event.audioDurationMs ?? null, + }, + { wallTimeUnixMs: event.timestampMs } + ); }, - { wallTimeUnixMs: event.timestampMs } - ); - }, - }) - .then((observer) => { - if (runtime.audioTrack === audioTrack) { - runtime.audioObserverStop = observer.stop; - return; - } - observer.stop(); - }) - .catch((error) => { - if (runtime.audioTrack !== audioTrack) { - return; - } - console.warn('[browser-audio] VAD observer unavailable', error); - recordFrontendObservability(FRONTEND_EVENTS.BROWSER_AUDIO_VAD_PROBE_UNAVAILABLE, { - [OBSERVABILITY_ATTRS.TRACK_NAME]: BROWSER_AUDIO_TRACK_NAME, - [OBSERVABILITY_ATTRS.FRONTEND_AUDIO_PROBE]: 'vad-web', - [OBSERVABILITY_ATTRS.FRONTEND_AUDIO_ERROR]: - error instanceof Error ? error.message : String(error), - }); - }); + }) + .then((observer) => { + if (runtime.audioTrack === audioTrack) { + runtime.audioObserverStop = observer.stop; + return; + } + void observer.stop(); + }) + .catch((error) => { + if (runtime.audioTrack !== audioTrack) { + return; + } + console.warn('[browser-audio] VAD observer unavailable', error); + recordFrontendObservability(FRONTEND_EVENTS.BROWSER_AUDIO_VAD_PROBE_UNAVAILABLE, { + [OBSERVABILITY_ATTRS.TRACK_NAME]: BROWSER_AUDIO_TRACK_NAME, + [OBSERVABILITY_ATTRS.FRONTEND_AUDIO_PROBE]: 'vad-web', + [OBSERVABILITY_ATTRS.FRONTEND_AUDIO_ERROR]: + error instanceof Error ? error.message : String(error), + }); + }); + } + } catch (error) { + audioTrack.mediaStreamTrack.enabled = false; + void audioTrack.mute().catch(() => undefined); + audioTrack.stop(); + throw error; + } + })(); + runtime.audioPublishPromise = publishPromise; + try { + await publishPromise; + } finally { + if (runtime.audioPublishPromise === publishPromise) { + runtime.audioPublishPromise = null; + } } - } catch (error) { - audioTrack.stop(); - throw error; - } - }, [ - appConfig.observabilityEnabled, - audioConfigured, - browserMediaStreamName, - recordFrontendObservability, - room, - ]); + }, + [ + appConfig.observabilityEnabled, + audioConfigured, + browserMediaStreamName, + recordFrontendObservability, + room, + ] + ); - const ensureVideoPublished = useCallback(async () => { - const runtime = runtimeRef.current; - if (!videoConfigured || !runtime || runtime.videoTrack || !runtime.videoEnabled) { - return; - } + const ensureVideoPublished = useCallback( + async (runtime: BrowserSourceRuntime) => { + if ( + !videoConfigured || + !isCurrentRuntime(runtimeRef, runtime) || + runtime.videoTrack || + !runtime.videoEnabled + ) { + return; + } - const videoTrack = await createLocalVideoTrack({ - facingMode: 'user', - frameRate: { ideal: browserVideoFrameRate, max: browserVideoFrameRate }, - resolution: { - width: browserVideoWidth, - height: browserVideoHeight, - frameRate: browserVideoFrameRate, - }, - }); - videoTrack.mediaStreamTrack.enabled = runtime.videoEnabled; - - try { - const publication = await room.localParticipant.publishTrack(videoTrack, { - name: BROWSER_VIDEO_TRACK_NAME, - source: Track.Source.Camera, - stream: browserMediaStreamName, - simulcast: false, - degradationPreference: 'maintain-resolution', - videoEncoding: { - maxBitrate: browserVideoMaxBitrate, - maxFramerate: browserVideoFrameRate, + const videoTrack = await createLocalVideoTrack({ + facingMode: 'user', + frameRate: { ideal: browserVideoFrameRate, max: browserVideoFrameRate }, + resolution: { + width: browserVideoWidth, + height: browserVideoHeight, + frameRate: browserVideoFrameRate, }, }); - runtime.videoTrack = videoTrack; - runtime.videoPublication = publication; - setVideoTrackState(videoTrack); - if (browserVideoStatsEnabled) { - startBrowserVideoStatsLogging(runtime, videoTrack, publication, room); + videoTrack.mediaStreamTrack.enabled = runtime.videoEnabled; + if (!isCurrentRuntime(runtimeRef, runtime)) { + videoTrack.stop(); + return; } - } catch (error) { - videoTrack.stop(); - throw error; - } - }, [ - browserMediaStreamName, - browserVideoFrameRate, - browserVideoHeight, - browserVideoMaxBitrate, - browserVideoStatsEnabled, - browserVideoWidth, - room, - videoConfigured, - ]); + + try { + const publication = await room.localParticipant.publishTrack(videoTrack, { + name: BROWSER_VIDEO_TRACK_NAME, + source: Track.Source.Camera, + stream: browserMediaStreamName, + simulcast: false, + degradationPreference: 'maintain-resolution', + videoEncoding: { + maxBitrate: browserVideoMaxBitrate, + maxFramerate: browserVideoFrameRate, + }, + }); + if (!isCurrentRuntime(runtimeRef, runtime)) { + await room.localParticipant.unpublishTrack(videoTrack, true).catch(() => undefined); + videoTrack.stop(); + return; + } + runtime.videoTrack = videoTrack; + runtime.videoPublication = publication; + setVideoTrackState(videoTrack); + if (browserVideoStatsEnabled) { + startBrowserVideoStatsLogging(runtime, videoTrack, publication, room); + } + } catch (error) { + videoTrack.stop(); + throw error; + } + }, + [ + browserMediaStreamName, + browserVideoFrameRate, + browserVideoHeight, + browserVideoMaxBitrate, + browserVideoStatsEnabled, + browserVideoWidth, + room, + videoConfigured, + ] + ); const unpublishAudio = useCallback( async (runtime: BrowserSourceRuntime) => { @@ -290,7 +358,7 @@ export function useBrowserSourceClient( stopBrowserVideoStatsLogging(runtime); runtime.videoTrack = null; runtime.videoPublication = null; - setVideoTrackState(null); + setVideoTrackState((currentTrack) => (currentTrack === track ? null : currentTrack)); if (!track) return; await room.localParticipant.unpublishTrack(track, true).catch(() => undefined); @@ -299,20 +367,39 @@ export function useBrowserSourceClient( [room] ); + const stopRuntime = useCallback( + async (runtime: BrowserSourceRuntime) => { + if (!runtime.stopPromise) { + runtime.stopPromise = (async () => { + const audioGate = runtime.audioGate; + if (audioGate) { + const gateStop = audioGate.adapter.stop(); + const executorStop = audioGate.executor.stop(); + audioGate.device.close(); + await gateStop; + await executorStop; + runtime.audioGate = null; + } + await Promise.all([unpublishAudio(runtime), unpublishVideo(runtime)]); + })(); + } + await runtime.stopPromise; + }, + [unpublishAudio, unpublishVideo] + ); + const stop = useCallback(async () => { - const runtime = runtimeRef.current; - runtimeRef.current = null; + const runtime = detachCurrentRuntime(runtimeRef); if (!runtime) return; - - await Promise.all([unpublishAudio(runtime), unpublishVideo(runtime)]); - }, [unpublishAudio, unpublishVideo]); + await stopRuntime(runtime); + }, [stopRuntime]); const start = useCallback(async () => { if (!enabled || runtimeRef.current) { return; } - runtimeRef.current = { + const runtime: BrowserSourceRuntime = { audioTrack: null, videoTrack: null, audioPublication: null, @@ -323,31 +410,83 @@ export function useBrowserSourceClient( videoStatsTimer: null, previousVideoStats: null, audioObserverStop: null, + audioPublishPromise: null, + audioGate: null, + stopPromise: null, }; + runtimeRef.current = runtime; - try { - if (audioEnabledRef.current) { - await ensureAudioPublished(); + await runOwnedRuntimeStart(runtimeRef, runtime, stopRuntime, async (stage) => { + if (audioConfigured) { + const agentName = appConfig.agentName?.trim(); + if (!agentName) { + throw new Error('agentName is required for browser audio media control'); + } + const device = new BrowserAudioGateDevice({ + getBinding: () => + runtime.audioTrack && runtime.audioPublication + ? { track: runtime.audioTrack, publication: runtime.audioPublication } + : null, + ensurePublishedClosed: async (signal) => { + if (signal.aborted) throw new DOMException('browser audio open aborted', 'AbortError'); + await ensureAudioPublished(runtime); + if (signal.aborted) throw new DOMException('browser audio open aborted', 'AbortError'); + }, + }); + const adapter = new LiveKitMediaGateAdapter({ + room, + agentName, + allowAnonymousLiveKitAgentFallback: true, + onError: (error) => console.warn('[browser-audio] media gate event failed', error), + }); + const executor = new MediaGateExecutor({ + targetIdentity: room.localParticipant.identity, + device, + publishState: adapter.publishState, + uuid: () => window.crypto.randomUUID(), + nowUnixMs: () => Date.now(), + nowMonotonicMs: () => performance.now(), + scheduler: { + setTimeout: (callback, delayMs) => window.setTimeout(callback, delayMs), + clearTimeout: (handle) => window.clearTimeout(handle as number), + }, + maxOpenLeaseMs: BROWSER_MEDIA_GATE_MAX_OPEN_LEASE_MS, + }); + const audioGate = { device, executor, adapter }; + runtime.audioGate = audioGate; + + await stage(() => audioGate.executor.start()); + if (!audioEnabledRef.current) { + await stage(() => audioGate.executor.setUserMuted(true)); + } else { + await stage(() => ensureAudioPublished(runtime)); + } + await stage(() => audioGate.executor.reconcileDevice()); + await stage(() => audioGate.adapter.start(audioGate.executor)); } - } catch (error) { - await stop(); - throw error; - } - if (videoEnabledRef.current) { - try { - await ensureVideoPublished(); - } catch (error) { - videoEnabledRef.current = false; - setVideoEnabledState(false); - const runtime = runtimeRef.current; - if (runtime) { + if (videoEnabledRef.current) { + try { + await stage(() => ensureVideoPublished(runtime)); + } catch (error) { + if (!isCurrentRuntime(runtimeRef, runtime)) throw error; + videoEnabledRef.current = false; + setVideoEnabledState(false); runtime.videoEnabled = false; + onVideoError?.(error as Error); } - onVideoError?.(error as Error); } - } - }, [enabled, ensureAudioPublished, ensureVideoPublished, onVideoError, stop]); + }); + }, [ + appConfig.agentName, + audioConfigured, + enabled, + ensureAudioPublished, + ensureVideoPublished, + onVideoError, + room, + stopRuntime, + ]); const setAudioEnabled = useCallback( async (nextEnabled: boolean) => { @@ -355,55 +494,36 @@ export function useBrowserSourceClient( return; } setAudioPending(true); - const previousEnabled = audioEnabledRef.current; const runtime = runtimeRef.current; - const previousRuntimeEnabled = runtime?.audioEnabled; - const previousAudioTrack = runtime?.audioTrack ?? null; - try { - audioEnabledRef.current = nextEnabled; - setAudioEnabledState(nextEnabled); + audioEnabledRef.current = nextEnabled; + setAudioEnabledState(nextEnabled); + if (runtime) runtime.audioEnabled = nextEnabled; + try { if (!runtime) return; - - runtime.audioEnabled = nextEnabled; - if (nextEnabled) { - if (runtime.audioTrack) { - syncTrackEnabled(runtime.audioTrack, true); - await runtime.audioTrack.unmute(); - recordFrontendObservability(FRONTEND_EVENTS.BROWSER_AUDIO_TRACK_UNMUTED, { - [OBSERVABILITY_ATTRS.TRACK_NAME]: BROWSER_AUDIO_TRACK_NAME, - }); - } else { - await ensureAudioPublished(); - } - } else if (runtime.audioTrack) { - syncTrackEnabled(runtime.audioTrack, false); - await runtime.audioTrack.mute(); + const audioGate = runtime.audioGate; + const executor = audioGate?.executor; + if (!nextEnabled) { + audioGate?.device.close(); + const muted = executor?.setUserMuted(true) ?? Promise.resolve(); recordFrontendObservability(FRONTEND_EVENTS.BROWSER_AUDIO_TRACK_MUTED, { [OBSERVABILITY_ATTRS.TRACK_NAME]: BROWSER_AUDIO_TRACK_NAME, }); + await muted; + return; } - } catch (error) { - audioEnabledRef.current = previousEnabled; - setAudioEnabledState(previousEnabled); - if (runtime && previousRuntimeEnabled !== undefined) { - runtime.audioEnabled = previousRuntimeEnabled; - if ( - !previousRuntimeEnabled && - runtime.audioTrack && - runtime.audioTrack !== previousAudioTrack - ) { - await unpublishAudio(runtime); - } else { - syncTrackEnabled(runtime.audioTrack, previousRuntimeEnabled); - } + + await executor?.setUserMuted(false); + if (audioGate && !audioGate.device.snapshot().trackMuted) { + recordFrontendObservability(FRONTEND_EVENTS.BROWSER_AUDIO_TRACK_UNMUTED, { + [OBSERVABILITY_ATTRS.TRACK_NAME]: BROWSER_AUDIO_TRACK_NAME, + }); } - throw error; } finally { setAudioPending(false); } }, - [audioConfigured, ensureAudioPublished, recordFrontendObservability, unpublishAudio] + [audioConfigured, recordFrontendObservability] ); const setAudioDeviceId = useCallback( @@ -422,16 +542,23 @@ export function useBrowserSourceClient( audioDeviceIdRef.current = nextDeviceId; const runtime = runtimeRef.current; try { - if (runtime?.audioEnabled) { - await unpublishAudio(runtime); - await ensureAudioPublished(); + if (runtime) { + const audioGate = runtime.audioGate; + await replaceRuntimeAudioBinding(runtime, { + close: () => audioGate?.device.close(), + unpublish: () => unpublishAudio(runtime), + reconcile: () => audioGate?.executor.reconcileDevice() ?? Promise.resolve(), + ensurePublished: () => ensureAudioPublished(runtime), + hasBinding: () => runtime.audioTrack !== null, + }); } } catch (error) { audioDeviceIdRef.current = previousDeviceId; if (runtime?.audioEnabled && !runtime.audioTrack) { - await ensureAudioPublished().catch((restoreError) => { + await ensureAudioPublished(runtime).catch((restoreError) => { console.warn('[browser-audio] failed to restore previous input device', restoreError); }); + await runtime.audioGate?.executor.reconcileDevice().catch(() => undefined); } throw error; } finally { @@ -459,7 +586,7 @@ export function useBrowserSourceClient( runtime.videoEnabled = nextEnabled; if (nextEnabled) { - await ensureVideoPublished(); + await ensureVideoPublished(runtime); if (runtime.videoTrack) { runtime.videoTrack.mediaStreamTrack.enabled = true; await runtime.videoTrack.unmute(); diff --git a/lib/browser-audio-gate-device.ts b/lib/browser-audio-gate-device.ts new file mode 100644 index 000000000..2316d397d --- /dev/null +++ b/lib/browser-audio-gate-device.ts @@ -0,0 +1,153 @@ +import type { MediaGateDevice, MediaGateDeviceState } from './media-gate-executor'; + +export type BrowserAudioGateTrack = { + readonly mediaStreamTrack: { + enabled: boolean; + readonly readyState?: MediaStreamTrackState; + }; + readonly isMuted: boolean; + mute(): Promise; + unmute(): Promise; +}; + +export type BrowserAudioGateBinding = { + readonly track: BrowserAudioGateTrack; + readonly publication: object; +}; + +export type BrowserAudioGateDeviceOptions = { + readonly getBinding: () => BrowserAudioGateBinding | null; + readonly ensurePublishedClosed: (signal: AbortSignal) => Promise; +}; + +export class BrowserAudioGateDevice implements MediaGateDevice { + private readonly getBinding: BrowserAudioGateDeviceOptions['getBinding']; + private readonly ensurePublishedClosed: BrowserAudioGateDeviceOptions['ensurePublishedClosed']; + private readonly signalingTails = new WeakMap>(); + private generation = 0; + private forcedClosed = true; + + constructor(options: BrowserAudioGateDeviceOptions) { + this.getBinding = options.getBinding; + this.ensurePublishedClosed = options.ensurePublishedClosed; + } + + close(): void { + this.generation += 1; + this.forcedClosed = true; + const binding = this.getBinding(); + if (!binding) return; + disableCapture(binding); + this.queueMute(binding); + } + + async open(signal: AbortSignal): Promise { + const generation = this.generation; + throwIfAborted(signal); + await this.ensurePublishedClosed(signal); + throwIfAborted(signal); + + const binding = this.getBinding(); + if (!binding) throw new Error('browser audio track is not published'); + const { track } = binding; + track.mediaStreamTrack.enabled = false; + + try { + await this.enqueueSignaling(track, async () => { + this.requireCurrent(binding, generation, signal); + const unmute = track.unmute(); + // LiveKit may synchronously toggle the underlying MediaStreamTrack. Keep capture + // closed until both unmute and all cancellation checks have completed. + track.mediaStreamTrack.enabled = false; + await unmute; + track.mediaStreamTrack.enabled = false; + this.requireCurrent(binding, generation, signal); + }); + this.forcedClosed = false; + track.mediaStreamTrack.enabled = true; + this.requireCurrent(binding, generation, signal); + } catch (error) { + disableCapture(binding); + if (this.generation === generation && this.getBinding()?.track === track) { + this.forcedClosed = true; + this.queueMute(binding); + } + throw error; + } + } + + snapshot(): MediaGateDeviceState { + const binding = this.getBinding(); + if (!binding) { + return { + captureActive: false, + trackPublished: false, + trackMuted: true, + }; + } + + const { track } = binding; + const captureActive = + !this.forcedClosed && + track.mediaStreamTrack.enabled && + track.mediaStreamTrack.readyState !== 'ended'; + return { + captureActive, + trackPublished: true, + trackMuted: this.forcedClosed || track.isMuted || !track.mediaStreamTrack.enabled, + }; + } + + private requireCurrent( + binding: BrowserAudioGateBinding, + generation: number, + signal: AbortSignal + ): void { + throwIfAborted(signal); + if (this.generation !== generation || this.getBinding()?.track !== binding.track) { + throw new DOMException('browser audio gate operation was superseded', 'AbortError'); + } + } + + private queueMute(binding: BrowserAudioGateBinding): void { + void this.enqueueSignaling(binding.track, async () => { + disableCapture(binding); + try { + await binding.track.mute(); + } catch { + // Capture was disabled synchronously; LiveKit mute is best effort. + } finally { + disableCapture(binding); + } + }); + } + + private enqueueSignaling( + track: BrowserAudioGateTrack, + operation: () => Promise + ): Promise { + const previous = this.signalingTails.get(track) ?? Promise.resolve(); + const result = previous.then(operation, operation); + const settled = result.then( + () => undefined, + () => undefined + ); + this.signalingTails.set(track, settled); + void settled.then(() => { + if (this.signalingTails.get(track) === settled) { + this.signalingTails.delete(track); + } + }); + return result; + } +} + +function disableCapture(binding: BrowserAudioGateBinding): void { + binding.track.mediaStreamTrack.enabled = false; +} + +function throwIfAborted(signal: AbortSignal): void { + if (signal.aborted) { + throw new DOMException('browser audio gate operation was aborted', 'AbortError'); + } +} diff --git a/lib/browser-source-runtime-lifecycle.ts b/lib/browser-source-runtime-lifecycle.ts new file mode 100644 index 000000000..87cd32bf4 --- /dev/null +++ b/lib/browser-source-runtime-lifecycle.ts @@ -0,0 +1,113 @@ +export type RuntimeSlot = { + current: Runtime | null; +}; + +export type RuntimeStartStage = ( + operation: () => Value | PromiseLike +) => Promise; + +export type AudioBindingRuntime = { + readonly audioEnabled: boolean; + readonly audioPublishPromise: Promise | null; +}; + +export type AudioBindingReplacement = { + close(): void; + unpublish(): Promise; + reconcile(): Promise; + ensurePublished(): Promise; + hasBinding(): boolean; +}; + +export class RuntimeStartCancelledError extends Error { + override readonly name = 'AbortError'; + + constructor(cause?: unknown) { + super('browser source runtime start was superseded'); + this.cause = cause; + } +} + +export function isCurrentRuntime(slot: RuntimeSlot, runtime: Runtime): boolean { + return slot.current === runtime; +} + +export function detachCurrentRuntime(slot: RuntimeSlot): Runtime | null { + const runtime = slot.current; + slot.current = null; + return runtime; +} + +export async function stopOwnedRuntime( + slot: RuntimeSlot, + runtime: Runtime, + stopRuntime: (runtime: Runtime) => Promise +): Promise { + if (slot.current === runtime) { + slot.current = null; + } + await stopRuntime(runtime); +} + +export async function runOwnedRuntimeStart( + slot: RuntimeSlot, + runtime: Runtime, + stopRuntime: (runtime: Runtime) => Promise, + startRuntime: (stage: RuntimeStartStage) => Promise +): Promise { + const assertOwned = () => { + if (!isCurrentRuntime(slot, runtime)) { + throw new RuntimeStartCancelledError(); + } + }; + const stage: RuntimeStartStage = async (operation) => { + assertOwned(); + try { + const value = await operation(); + assertOwned(); + return value; + } catch (error) { + if (!isCurrentRuntime(slot, runtime)) { + throw new RuntimeStartCancelledError(error); + } + throw error; + } + }; + + try { + assertOwned(); + await startRuntime(stage); + assertOwned(); + } catch (error) { + try { + await stopOwnedRuntime(slot, runtime, stopRuntime); + } catch (stopError) { + if (isAbortError(error)) throw error; + throw stopError; + } + throw error; + } +} + +export async function replaceRuntimeAudioBinding( + runtime: AudioBindingRuntime, + replacement: AudioBindingReplacement +): Promise { + replacement.close(); + await runtime.audioPublishPromise?.catch(() => undefined); + await replacement.unpublish(); + if (!runtime.audioEnabled) return; + + await replacement.reconcile(); + if (replacement.hasBinding()) return; + + await replacement.ensurePublished(); + await replacement.reconcile(); +} + +function isAbortError(error: unknown): boolean { + return ( + error instanceof RuntimeStartCancelledError || + (typeof error === 'object' && error !== null && 'name' in error && error.name === 'AbortError') + ); +} diff --git a/lib/livekit-media-gate.ts b/lib/livekit-media-gate.ts new file mode 100644 index 000000000..0c2278861 --- /dev/null +++ b/lib/livekit-media-gate.ts @@ -0,0 +1,276 @@ +import { type RemoteParticipant, type Room, RoomEvent } from 'livekit-client'; +import { + MEDIA_CONTROL_TOPIC, + MEDIA_STATE_TOPIC, + type MediaControlCommand, + type MediaStateSnapshot, + decodeMediaControl, + encodeMediaState, +} from './media-control-protocol'; + +export type MediaGateExecutorPort = { + start(): Promise; + bindController(controllerIdentity: string): Promise; + handleCommand(controllerIdentity: string, command: MediaControlCommand): Promise; + handleMalformedControl(controllerIdentity: string, errorCode: string): Promise; + disconnectController(controllerIdentity: string): Promise; + stop(): Promise; +}; + +export type MediaStatePublisher = ( + controllerIdentity: string, + state: MediaStateSnapshot, + signal: AbortSignal +) => Promise; + +export type LiveKitMediaGateAdapterOptions = { + readonly room: Room; + readonly agentName: string; + readonly allowAnonymousLiveKitAgentFallback?: boolean; + readonly onError?: (error: unknown) => void; +}; + +export class LiveKitMediaGateAdapter { + private readonly room: Room; + private readonly agentName: string; + private readonly allowAnonymousLiveKitAgentFallback: boolean; + private readonly onError: ((error: unknown) => void) | undefined; + + private executor: MediaGateExecutorPort | null = null; + private controllerIdentity: string | null = null; + private started = false; + private stopped = false; + private eventTail: Promise = Promise.resolve(); + private readonly ownedTasks = new Set>(); + + constructor(options: LiveKitMediaGateAdapterOptions) { + if (!options.agentName) throw new Error('agentName must be non-empty'); + this.room = options.room; + this.agentName = options.agentName; + this.allowAnonymousLiveKitAgentFallback = options.allowAnonymousLiveKitAgentFallback ?? false; + this.onError = options.onError; + } + + async start(executor: MediaGateExecutorPort): Promise { + if (this.stopped) throw new Error('LiveKit media gate adapter cannot restart after stop'); + if (this.started) { + if (this.executor !== executor) { + throw new Error('LiveKit media gate adapter is already bound to an executor'); + } + return this.drain(); + } + + await executor.start(); + this.executor = executor; + this.started = true; + this.room.on(RoomEvent.DataReceived, this.handleDataReceived); + this.room.on(RoomEvent.ParticipantConnected, this.handleParticipantConnected); + this.room.on(RoomEvent.ParticipantDisconnected, this.handleParticipantDisconnected); + try { + await this.discoverAndBindController(); + } catch (startError) { + this.started = false; + this.stopped = true; + this.controllerIdentity = null; + this.removeListeners(); + try { + await Promise.all([executor.stop(), this.drain()]); + } catch (cleanupError) { + this.executor = null; + throw new AggregateError( + [startError, cleanupError], + 'LiveKit media gate adapter start and cleanup failed' + ); + } + this.executor = null; + throw startError; + } + } + + readonly publishState: MediaStatePublisher = async (controllerIdentity, state, signal) => { + throwIfAborted(signal); + if (!this.started || this.stopped || controllerIdentity !== this.controllerIdentity) return; + await publishLiveKitMediaState(this.room, controllerIdentity, state, signal); + }; + + async stop(): Promise { + if (this.stopped) return this.drain(); + this.stopped = true; + this.started = false; + this.removeListeners(); + + const executor = this.executor; + this.controllerIdentity = null; + const stopPromise = executor?.stop() ?? Promise.resolve(); + await Promise.all([stopPromise, this.drain()]); + this.executor = null; + } + + async drain(): Promise { + while (true) { + const eventTail = this.eventTail; + const tasks = [...this.ownedTasks]; + await Promise.all([eventTail, ...tasks]); + if (eventTail === this.eventTail && this.ownedTasks.size === 0) return; + } + } + + private readonly handleDataReceived = ( + payload: Uint8Array, + participant?: RemoteParticipant, + _kind?: unknown, + topic?: string + ): void => { + if (topic !== MEDIA_CONTROL_TOPIC) return; + const controllerIdentity = this.controllerIdentity; + if ( + !this.started || + this.stopped || + !participant?.isAgent || + !controllerIdentity || + participant.identity !== controllerIdentity + ) { + return; + } + + this.trackEvent(async () => { + const executor = this.executor; + if ( + !executor || + !this.started || + this.stopped || + this.controllerIdentity !== controllerIdentity + ) { + return; + } + let command: MediaControlCommand; + try { + command = decodeMediaControl(payload); + } catch (error) { + await executor.handleMalformedControl(controllerIdentity, mediaControlErrorCode(error)); + return; + } + await executor.handleCommand(controllerIdentity, command); + }); + }; + + private readonly handleParticipantConnected = (): void => { + this.trackEvent(() => this.discoverAndBindController()); + }; + + private readonly handleParticipantDisconnected = (participant: RemoteParticipant): void => { + if (participant.identity !== this.controllerIdentity) return; + this.controllerIdentity = null; + const executor = this.executor; + if (!executor) return; + this.trackOwned( + Promise.resolve().then(() => executor.disconnectController(participant.identity)) + ); + }; + + private async discoverAndBindController(): Promise { + if (!this.started || this.stopped || this.controllerIdentity) return; + const executor = this.executor; + if (!executor) return; + + const participant = findTrustedController( + [...this.room.remoteParticipants.values()], + this.agentName, + this.allowAnonymousLiveKitAgentFallback + ); + if (!participant) return; + + this.controllerIdentity = participant.identity; + try { + await executor.bindController(participant.identity); + } catch (error) { + if (this.controllerIdentity === participant.identity) this.controllerIdentity = null; + throw error; + } + } + + private trackEvent(action: () => Promise): void { + const next = this.eventTail.then(async () => { + if (!this.started || this.stopped) return; + await action(); + }); + this.eventTail = this.trackOwned(next); + } + + private trackOwned(task: Promise): Promise { + const observed = task.catch((error: unknown) => { + try { + this.onError?.(error); + } catch { + // The adapter owns event promises; a diagnostic callback must not create a rejection. + } + }); + this.ownedTasks.add(observed); + void observed.then(() => { + this.ownedTasks.delete(observed); + }); + return observed; + } + + private removeListeners(): void { + this.room.off(RoomEvent.DataReceived, this.handleDataReceived); + this.room.off(RoomEvent.ParticipantConnected, this.handleParticipantConnected); + this.room.off(RoomEvent.ParticipantDisconnected, this.handleParticipantDisconnected); + } +} + +export async function publishLiveKitMediaState( + room: Room, + controllerIdentity: string, + state: MediaStateSnapshot, + signal: AbortSignal +): Promise { + throwIfAborted(signal); + const payload = encodeMediaState(state); + throwIfAborted(signal); + await room.localParticipant.publishData(payload, { + reliable: true, + destinationIdentities: [controllerIdentity], + topic: MEDIA_STATE_TOPIC, + }); +} + +function mediaControlErrorCode(error: unknown): string { + if ( + error instanceof Error && + error.message.includes('schema_version') && + error.message.includes('not supported') + ) { + return 'unsupported_control_version'; + } + return 'malformed_control'; +} + +function findTrustedController( + participants: readonly RemoteParticipant[], + agentName: string, + allowAnonymousLiveKitAgentFallback: boolean +): RemoteParticipant | null { + const expected = participants.find( + (participant) => participant.isAgent && readAgentName(participant) === agentName + ); + if (expected) return expected; + if (!allowAnonymousLiveKitAgentFallback) return null; + + const anonymousAgents = participants.filter( + (participant) => + participant.isAgent && + participant.identity.startsWith('agent-') && + !readAgentName(participant) + ); + return anonymousAgents.length === 1 ? anonymousAgents[0] : null; +} + +function readAgentName(participant: RemoteParticipant): string { + const attributes = participant.attributes; + return attributes['lk.agent.name'] || attributes['lk.agent_name'] || attributes.lkAgentName || ''; +} + +function throwIfAborted(signal: AbortSignal): void { + if (signal.aborted) throw new DOMException('media state publish aborted', 'AbortError'); +} diff --git a/lib/media-control-protocol.ts b/lib/media-control-protocol.ts new file mode 100644 index 000000000..e022d9c95 --- /dev/null +++ b/lib/media-control-protocol.ts @@ -0,0 +1,472 @@ +export const MEDIA_CONTROL_TOPIC = 'lk.media.control'; +export const MEDIA_STATE_TOPIC = 'lk.media.state'; + +const SCHEMA_VERSION = 1; +const MAX_PACKET_BYTES = 16 * 1024; +const MAX_SAFE_INTEGER_BIGINT = BigInt(Number.MAX_SAFE_INTEGER); + +export type MediaControlCommand = { + readonly schema_version: 1; + readonly type: typeof MEDIA_CONTROL_TOPIC; + readonly command_id: string; + readonly policy_epoch: string; + readonly sequence: number; + readonly target_identity: string; + readonly desired_listening: 'open' | 'closed'; + readonly issued_at_unix_ms: number; + readonly expires_at_unix_ms: number; + readonly reason: string | null; +}; + +export type MediaStateSnapshot = { + readonly schema_version: 1; + readonly type: typeof MEDIA_STATE_TOPIC; + readonly target_identity: string; + readonly state_epoch: string; + readonly state_sequence: number; + readonly observed_at_unix_ms: number; + readonly capture_active: boolean; + readonly track_published: boolean; + readonly track_muted: boolean; + readonly user_muted: boolean; + readonly blocked_by: readonly string[]; + readonly command_id: string | null; + readonly policy_epoch: string | null; + readonly command_sequence: number | null; + readonly command_status: 'applied' | 'rejected' | 'expired' | 'unsupported' | null; + readonly error_code: string | null; +}; + +type JsonObject = Record; + +type ParsedValue = { + value: unknown; + integerToken: boolean; +}; + +class StrictJsonParser { + private index = 0; + private readonly rootIntegerFields = new Set(); + + constructor(private readonly text: string) {} + + parse(): { value: unknown; rootIntegerFields: ReadonlySet } { + this.skipWhitespace(); + const { value } = this.parseValue(0); + this.skipWhitespace(); + if (this.index !== this.text.length) { + invalid('payload is not valid JSON'); + } + return { value, rootIntegerFields: this.rootIntegerFields }; + } + + private parseValue(depth: number): ParsedValue { + const character = this.text[this.index]; + if (character === '{') return { value: this.parseObject(depth), integerToken: false }; + if (character === '[') return { value: this.parseArray(depth), integerToken: false }; + if (character === '"') return { value: this.parseString(), integerToken: false }; + if (character === 't') return { value: this.parseLiteral('true', true), integerToken: false }; + if (character === 'f') return { value: this.parseLiteral('false', false), integerToken: false }; + if (character === 'n') return { value: this.parseLiteral('null', null), integerToken: false }; + if (character === '-' || isDigit(character)) return this.parseNumber(); + return invalid('payload is not valid JSON'); + } + + private parseObject(depth: number): JsonObject { + this.index += 1; + this.skipWhitespace(); + const result: JsonObject = Object.create(null) as JsonObject; + const keys = new Set(); + if (this.consume('}')) return result; + + while (true) { + if (this.text[this.index] !== '"') invalid('payload is not valid JSON'); + const key = this.parseString(); + if (keys.has(key)) invalid(`duplicate JSON field ${JSON.stringify(key)}`); + keys.add(key); + this.skipWhitespace(); + this.expect(':'); + this.skipWhitespace(); + const parsed = this.parseValue(depth + 1); + if (depth === 0 && parsed.integerToken) this.rootIntegerFields.add(key); + Object.defineProperty(result, key, { + value: parsed.value, + enumerable: true, + configurable: true, + writable: true, + }); + this.skipWhitespace(); + if (this.consume('}')) return result; + this.expect(','); + this.skipWhitespace(); + } + } + + private parseArray(depth: number): unknown[] { + this.index += 1; + this.skipWhitespace(); + const result: unknown[] = []; + if (this.consume(']')) return result; + + while (true) { + result.push(this.parseValue(depth + 1).value); + this.skipWhitespace(); + if (this.consume(']')) return result; + this.expect(','); + this.skipWhitespace(); + } + } + + private parseString(): string { + const start = this.index; + this.index += 1; + while (this.index < this.text.length) { + const character = this.text[this.index]; + this.index += 1; + if (character === '"') { + try { + return JSON.parse(this.text.slice(start, this.index)) as string; + } catch { + return invalid('payload is not valid JSON'); + } + } + if (character === '\\') { + const escape = this.text[this.index]; + this.index += 1; + if (escape === 'u') { + const digits = this.text.slice(this.index, this.index + 4); + if (!/^[0-9a-fA-F]{4}$/.test(digits)) invalid('payload is not valid JSON'); + this.index += 4; + } else if (!'"\\/bfnrt'.includes(escape ?? '')) { + invalid('payload is not valid JSON'); + } + } else if (character.charCodeAt(0) < 0x20) { + invalid('payload is not valid JSON'); + } + } + return invalid('payload is not valid JSON'); + } + + private parseNumber(): ParsedValue { + const start = this.index; + if (this.consume('-') && this.index === this.text.length) { + return invalid('payload is not valid JSON'); + } + if (this.consume('0')) { + if (isDigit(this.text[this.index])) return invalid('payload is not valid JSON'); + } else { + if (!isNonzeroDigit(this.text[this.index])) return invalid('payload is not valid JSON'); + while (isDigit(this.text[this.index])) this.index += 1; + } + + let integerToken = true; + if (this.consume('.')) { + integerToken = false; + if (!isDigit(this.text[this.index])) return invalid('payload is not valid JSON'); + while (isDigit(this.text[this.index])) this.index += 1; + } + if (this.text[this.index] === 'e' || this.text[this.index] === 'E') { + integerToken = false; + this.index += 1; + if (this.text[this.index] === '+' || this.text[this.index] === '-') this.index += 1; + if (!isDigit(this.text[this.index])) return invalid('payload is not valid JSON'); + while (isDigit(this.text[this.index])) this.index += 1; + } + + const token = this.text.slice(start, this.index); + if (integerToken) { + let value: bigint; + try { + value = BigInt(token); + } catch { + return invalid('payload is not valid JSON'); + } + if (value < -MAX_SAFE_INTEGER_BIGINT || value > MAX_SAFE_INTEGER_BIGINT) { + return invalid('JSON integer exceeds the protocol safe-integer bound'); + } + } + const value = Number(token); + return { value: Object.is(value, -0) ? 0 : value, integerToken }; + } + + private parseLiteral(token: string, value: T): T { + if (this.text.slice(this.index, this.index + token.length) !== token) { + return invalid('payload is not valid JSON'); + } + this.index += token.length; + return value; + } + + private skipWhitespace(): void { + while (' \n\r\t'.includes(this.text[this.index] ?? '\0')) this.index += 1; + } + + private consume(character: string): boolean { + if (this.text[this.index] !== character) return false; + this.index += 1; + return true; + } + + private expect(character: string): void { + if (!this.consume(character)) invalid('payload is not valid JSON'); + } +} + +export function decodeMediaControl(payload: Uint8Array | string): MediaControlCommand { + const text = decodePayload(payload); + let parsed: { value: unknown; rootIntegerFields: ReadonlySet }; + try { + parsed = new StrictJsonParser(text).parse(); + } catch (error) { + if (error instanceof Error) throw error; + return invalid('payload is not valid JSON'); + } + const values = requireObject(parsed.value); + validateEnvelope(values, parsed.rootIntegerFields, MEDIA_CONTROL_TOPIC); + + const command: MediaControlCommand = { + schema_version: SCHEMA_VERSION, + type: MEDIA_CONTROL_TOPIC, + command_id: requireNonemptyString(values, 'command_id'), + policy_epoch: requireNonemptyString(values, 'policy_epoch'), + sequence: requirePositiveInteger(values, parsed.rootIntegerFields, 'sequence'), + target_identity: requireNonemptyString(values, 'target_identity'), + desired_listening: requireListening(values, 'desired_listening'), + issued_at_unix_ms: requireNonnegativeInteger( + values, + parsed.rootIntegerFields, + 'issued_at_unix_ms' + ), + expires_at_unix_ms: requireInteger(values, parsed.rootIntegerFields, 'expires_at_unix_ms'), + reason: Object.hasOwn(values, 'reason') ? requireNonemptyString(values, 'reason') : null, + }; + if (command.expires_at_unix_ms <= command.issued_at_unix_ms) { + invalid('expires_at_unix_ms must be greater than issued_at_unix_ms'); + } + return Object.freeze(command); +} + +export function encodeMediaState(message: MediaStateSnapshot): Uint8Array { + const values = requireObject(message); + validateMediaState(values); + const ordered = Object.fromEntries( + Object.entries({ + schema_version: values.schema_version, + type: values.type, + target_identity: values.target_identity, + state_epoch: values.state_epoch, + state_sequence: values.state_sequence, + observed_at_unix_ms: values.observed_at_unix_ms, + capture_active: values.capture_active, + track_published: values.track_published, + track_muted: values.track_muted, + user_muted: values.user_muted, + blocked_by: values.blocked_by, + command_id: values.command_id, + policy_epoch: values.policy_epoch, + command_sequence: values.command_sequence, + command_status: values.command_status, + error_code: values.error_code, + }).sort(([left], [right]) => (left < right ? -1 : left > right ? 1 : 0)) + ); + const encoded = new TextEncoder().encode(JSON.stringify(ordered)); + enforcePacketSize(encoded.byteLength); + return encoded; +} + +function decodePayload(payload: Uint8Array | string): string { + if (typeof payload === 'string') { + if (payload.length > MAX_PACKET_BYTES) enforcePacketSize(payload.length); + requireWellFormed(payload, 'payload'); + enforcePacketSize(new TextEncoder().encode(payload).byteLength); + return payload; + } + if (!(payload instanceof Uint8Array)) invalid('payload must be bytes or text'); + enforcePacketSize(payload.byteLength); + try { + return new TextDecoder('utf-8', { fatal: true, ignoreBOM: true }).decode(payload); + } catch { + return invalid('payload is not valid UTF-8'); + } +} + +function validateEnvelope( + values: JsonObject, + integerFields: ReadonlySet, + expectedType: string +): void { + const version = requireInteger(values, integerFields, 'schema_version'); + if (version !== SCHEMA_VERSION) invalid(`schema_version ${version} is not supported`); + if (required(values, 'type') !== expectedType) invalid(`expected message type ${expectedType}`); +} + +function validateMediaState(values: JsonObject): void { + if (required(values, 'schema_version') !== SCHEMA_VERSION) invalid('unsupported schema_version'); + if (required(values, 'type') !== MEDIA_STATE_TOPIC) invalid('wrong media state type'); + requireNonemptyString(values, 'target_identity'); + requireNonemptyString(values, 'state_epoch'); + requireRuntimePositiveInteger(values, 'state_sequence'); + requireRuntimeNonnegativeInteger(values, 'observed_at_unix_ms'); + requireBoolean(values, 'capture_active'); + requireBoolean(values, 'track_published'); + requireBoolean(values, 'track_muted'); + requireBoolean(values, 'user_muted'); + + const blockers = required(values, 'blocked_by'); + if (!Array.isArray(blockers)) invalid('blocked_by must be a list'); + const uniqueBlockers = new Set(); + for (const blocker of blockers) { + if (typeof blocker !== 'string' || blocker.length === 0) { + invalid('blocked_by items must be non-empty strings'); + } + requireWellFormed(blocker, 'blocked_by item'); + if (uniqueBlockers.has(blocker)) invalid('blocked_by must not contain duplicates'); + uniqueBlockers.add(blocker); + } + + const correlation = [ + required(values, 'command_id'), + required(values, 'policy_epoch'), + required(values, 'command_sequence'), + required(values, 'command_status'), + ]; + const presentCount = correlation.filter((value) => value !== null).length; + if (presentCount !== 0 && presentCount !== correlation.length) { + invalid('command correlation must be entirely null or entirely non-null'); + } + if (presentCount > 0) { + requireNonemptyString(values, 'command_id'); + requireNonemptyString(values, 'policy_epoch'); + requireRuntimePositiveInteger(values, 'command_sequence'); + if ( + !['applied', 'rejected', 'expired', 'unsupported'].includes(values.command_status as string) + ) { + invalid('command_status is not supported'); + } + } + + const errorCode = required(values, 'error_code'); + if (errorCode !== null) requireNonemptyString(values, 'error_code'); + for (const field of [ + 'target_identity', + 'state_epoch', + 'command_id', + 'policy_epoch', + 'error_code', + ]) { + const value = values[field]; + if (typeof value === 'string') requireWellFormed(value, field); + } +} + +function requireObject(value: unknown): JsonObject { + if (typeof value !== 'object' || value === null || Array.isArray(value)) { + return invalid('payload must contain a JSON object'); + } + return value as JsonObject; +} + +function required(values: JsonObject, field: string): unknown { + if (!Object.hasOwn(values, field)) invalid(`required field ${field} is missing`); + return values[field]; +} + +function requireNonemptyString(values: JsonObject, field: string): string { + const value = required(values, field); + if (typeof value !== 'string' || value.length === 0) { + return invalid(`${field} must be a non-empty string`); + } + requireWellFormed(value, field); + return value; +} + +function requireInteger( + values: JsonObject, + integerFields: ReadonlySet, + field: string +): number { + const value = required(values, field); + if (!integerFields.has(field) || typeof value !== 'number' || !Number.isSafeInteger(value)) { + return invalid(`${field} must be a safe integer token`); + } + return value; +} + +function requirePositiveInteger( + values: JsonObject, + integerFields: ReadonlySet, + field: string +): number { + const value = requireInteger(values, integerFields, field); + if (value <= 0) return invalid(`${field} must be positive`); + return value; +} + +function requireNonnegativeInteger( + values: JsonObject, + integerFields: ReadonlySet, + field: string +): number { + const value = requireInteger(values, integerFields, field); + if (value < 0) return invalid(`${field} must be non-negative`); + return value; +} + +function requireRuntimePositiveInteger(values: JsonObject, field: string): number { + const value = required(values, field); + if (typeof value !== 'number' || !Number.isSafeInteger(value) || value <= 0) { + return invalid(`${field} must be a positive safe integer`); + } + return value; +} + +function requireRuntimeNonnegativeInteger(values: JsonObject, field: string): number { + const value = required(values, field); + if (typeof value !== 'number' || !Number.isSafeInteger(value) || value < 0) { + return invalid(`${field} must be a non-negative safe integer`); + } + return value; +} + +function requireListening(values: JsonObject, field: string): 'open' | 'closed' { + const value = required(values, field); + if (value !== 'open' && value !== 'closed') { + return invalid(`${field} must be 'open' or 'closed'`); + } + return value; +} + +function requireBoolean(values: JsonObject, field: string): boolean { + const value = required(values, field); + if (typeof value !== 'boolean') return invalid(`${field} must be a boolean`); + return value; +} + +function requireWellFormed(value: string, field: string): void { + for (let index = 0; index < value.length; index += 1) { + const code = value.charCodeAt(index); + if (code >= 0xd800 && code <= 0xdbff) { + const next = value.charCodeAt(index + 1); + if (!(next >= 0xdc00 && next <= 0xdfff)) invalid(`${field} is not valid UTF-8 text`); + index += 1; + } else if (code >= 0xdc00 && code <= 0xdfff) { + invalid(`${field} is not valid UTF-8 text`); + } + } +} + +function enforcePacketSize(byteLength: number): void { + if (byteLength > MAX_PACKET_BYTES) invalid('payload exceeds the 16 KiB packet limit'); +} + +function isDigit(character: string | undefined): boolean { + return character !== undefined && character >= '0' && character <= '9'; +} + +function isNonzeroDigit(character: string | undefined): boolean { + return character !== undefined && character >= '1' && character <= '9'; +} + +function invalid(message: string): never { + throw new Error(message); +} diff --git a/lib/media-gate-executor.ts b/lib/media-gate-executor.ts new file mode 100644 index 000000000..a8a7570c2 --- /dev/null +++ b/lib/media-gate-executor.ts @@ -0,0 +1,722 @@ +import { + MEDIA_STATE_TOPIC, + type MediaControlCommand, + type MediaStateSnapshot, +} from './media-control-protocol'; + +export type MediaGateDeviceState = { + readonly captureActive: boolean; + readonly trackPublished: boolean; + readonly trackMuted: boolean; +}; + +export type MediaGateDevice = { + close(): void; + open(signal: AbortSignal): Promise; + snapshot(): MediaGateDeviceState; +}; + +export type MediaGateScheduler = { + setTimeout(callback: () => void, delayMs: number): unknown; + clearTimeout(handle: unknown): void; +}; + +export type MediaGateExecutorOptions = { + readonly targetIdentity: string; + readonly device: MediaGateDevice; + readonly publishState: ( + controllerIdentity: string, + state: MediaStateSnapshot, + signal: AbortSignal + ) => Promise; + readonly uuid: () => string; + readonly nowUnixMs: () => number; + readonly nowMonotonicMs: () => number; + readonly scheduler: MediaGateScheduler; + readonly maxOpenLeaseMs: number; +}; + +type CommandStatus = NonNullable; + +type CommandResult = { + readonly command: MediaControlCommand; + status: CommandStatus | null; + errorCode: string | null; +}; + +type OpenLease = { + readonly id: number; + readonly deadlineMonotonicMs: number; + readonly command: MediaControlCommand; + timer: unknown; +}; + +const NULL_CORRELATION = { + command_id: null, + policy_epoch: null, + command_sequence: null, + command_status: null, +} as const; +const MAX_RETIRED_POLICY_EPOCHS = 16; + +export class MediaGateExecutor { + private readonly targetIdentity: string; + private readonly device: MediaGateDevice; + private readonly publishStateSink: MediaGateExecutorOptions['publishState']; + private readonly nowUnixMs: () => number; + private readonly nowMonotonicMs: () => number; + private readonly scheduler: MediaGateScheduler; + private readonly maxOpenLeaseMs: number; + private readonly stateEpoch: string; + + private started = false; + private stopped = false; + private controllerIdentity: string | null = null; + private userMuted = false; + private desiredListening: 'open' | 'closed' = 'closed'; + private closedBlocker = 'controller_disconnected'; + private stateSequence = 0; + private operationVersion = 0; + private appliedOpenVersion: number | null = null; + private operationAbortController = new AbortController(); + private nextLeaseId = 1; + private lease: OpenLease | null = null; + private activePolicyEpoch: string | null = null; + private readonly retiredPolicyEpochs = new Set(); + private readonly retiredPolicyEpochOrder: string[] = []; + private lastOrderedCommand: MediaControlCommand | null = null; + private lastCommand: CommandResult | null = null; + private tail: Promise = Promise.resolve(); + + constructor(options: MediaGateExecutorOptions) { + requireNonempty(options.targetIdentity, 'targetIdentity'); + if (!Number.isSafeInteger(options.maxOpenLeaseMs) || options.maxOpenLeaseMs <= 0) { + throw new Error('maxOpenLeaseMs must be a positive safe integer'); + } + const stateEpoch = options.uuid(); + requireNonempty(stateEpoch, 'state epoch'); + + this.targetIdentity = options.targetIdentity; + this.device = options.device; + this.publishStateSink = options.publishState; + this.nowUnixMs = options.nowUnixMs; + this.nowMonotonicMs = options.nowMonotonicMs; + this.scheduler = options.scheduler; + this.maxOpenLeaseMs = options.maxOpenLeaseMs; + this.stateEpoch = stateEpoch; + } + + start(): Promise { + if (this.stopped) throw new Error('media gate executor cannot restart after stop'); + if (this.started) return this.drain(); + this.started = true; + this.invalidateAndClose('controller_disconnected', true); + return Promise.resolve(); + } + + bindController(controllerIdentity: string): Promise { + this.requireStarted(); + requireNonempty(controllerIdentity, 'controller identity'); + if (this.controllerIdentity && this.controllerIdentity !== controllerIdentity) { + this.invalidateAndClose('controller_conflict', true); + return Promise.reject(new Error('a different media controller is already bound')); + } + + this.controllerIdentity = controllerIdentity; + this.closedBlocker = 'lease_missing'; + this.device.close(); + return this.enqueue(() => this.publishSnapshot(controllerIdentity, null, null)); + } + + handleCommand(controllerIdentity: string, command: MediaControlCommand): Promise { + if ( + !this.started || + this.stopped || + controllerIdentity !== this.controllerIdentity || + command.target_identity !== this.targetIdentity + ) { + return Promise.resolve(); + } + + const ordering = this.classifyCommand(command); + if (ordering === 'duplicate') { + const result = this.lastCommand; + if (!result) return Promise.resolve(); + return this.enqueue(() => this.publishCommandResult(controllerIdentity, result)); + } + if (ordering === 'rejected') { + const result = { + command, + status: 'rejected' as const, + errorCode: 'command_rejected', + }; + this.transitionActiveOpenResult('rejected', 'superseded'); + this.invalidateAndClose('command_rejected', true); + return this.enqueue(async () => { + await this.publishCommandResult(controllerIdentity, result); + await this.publishSnapshot(controllerIdentity, null, null); + }); + } + + this.rememberAcceptedCommand(command); + if (command.expires_at_unix_ms <= this.nowUnixMs()) { + const result = { command, status: 'expired' as const, errorCode: null }; + this.lastCommand = result; + this.invalidateAndClose('lease_expired', true); + return this.enqueue(() => this.publishCommandResult(controllerIdentity, result)); + } + + if (command.desired_listening === 'closed') { + const result = { command, status: 'applied' as const, errorCode: null }; + this.lastCommand = result; + this.invalidateAndClose(command.reason ?? 'automatic_gate_closed', true); + return this.enqueue(() => this.publishCommandResult(controllerIdentity, result)); + } + + return this.acceptOpen(controllerIdentity, command); + } + + setUserMuted(userMuted: boolean): Promise { + this.requireStarted(); + if (this.stopped || this.userMuted === userMuted) return this.drain(); + + this.userMuted = userMuted; + this.advanceOperation(); + if (userMuted) { + this.transitionActiveOpenResult('rejected', 'user_muted'); + this.device.close(); + return this.publishCurrentWhenBound(); + } + + if (!this.hasFreshOpenLease()) { + this.device.close(); + return this.publishCurrentWhenBound(); + } + const controllerIdentity = this.controllerIdentity; + if (!controllerIdentity) return Promise.resolve(); + const result = this.activeOpenResult(); + const operationVersion = this.operationVersion; + return this.enqueue(() => + this.applyOpen(controllerIdentity, operationVersion, result, false, false) + ); + } + + disconnectController(controllerIdentity: string): Promise { + if (controllerIdentity !== this.controllerIdentity) return this.drain(); + this.controllerIdentity = null; + this.resetCommandOrdering(); + this.invalidateAndClose('controller_disconnected', true); + return this.drain(); + } + + handleMalformedControl(controllerIdentity: string, errorCode: string): Promise { + if (controllerIdentity !== this.controllerIdentity || this.stopped) { + return Promise.resolve(); + } + requireNonempty(errorCode, 'error code'); + this.transitionActiveOpenResult('rejected', 'superseded'); + this.invalidateAndClose('malformed_control', true); + return this.enqueue(() => this.publishSnapshot(controllerIdentity, null, errorCode)); + } + + reconcileDevice(): Promise { + this.requireStarted(); + if (this.stopped) return this.drain(); + + this.advanceOperation(); + this.device.close(); + const controllerIdentity = this.controllerIdentity; + if (!controllerIdentity) return this.drain(); + const result = this.activeOpenResult(); + if (!this.userMuted && this.hasFreshOpenLease() && result) { + const operationVersion = this.operationVersion; + return this.enqueue(() => + this.applyOpen(controllerIdentity, operationVersion, result, false, false) + ); + } + return this.enqueue(() => this.publishSnapshot(controllerIdentity, null, null)); + } + + stop(): Promise { + if (this.stopped) return this.drain(); + this.stopped = true; + this.started = false; + this.controllerIdentity = null; + this.invalidateAndClose('stopped', true); + return this.drain(); + } + + drain(): Promise { + return this.tail; + } + + private acceptOpen(controllerIdentity: string, command: MediaControlCommand): Promise { + this.clearLease(); + this.advanceOperation(); + this.desiredListening = 'open'; + this.closedBlocker = 'lease_missing'; + const result: CommandResult = { + command, + status: this.userMuted ? 'rejected' : null, + errorCode: this.userMuted ? 'user_muted' : null, + }; + this.lastCommand = result; + const operationVersion = this.operationVersion; + const remainingWallMs = command.expires_at_unix_ms - this.nowUnixMs(); + const durationMs = Math.min(remainingWallMs, this.maxOpenLeaseMs); + const lease: OpenLease = { + id: this.nextLeaseId, + deadlineMonotonicMs: this.nowMonotonicMs() + durationMs, + command, + timer: 0, + }; + this.nextLeaseId += 1; + lease.timer = this.scheduler.setTimeout(() => this.expireLease(lease.id), durationMs); + this.lease = lease; + + if (this.userMuted) { + this.device.close(); + return this.enqueue(() => this.publishCommandResult(controllerIdentity, result)); + } + return this.enqueue(() => + this.applyOpen(controllerIdentity, operationVersion, result, true, true) + ); + } + + private async applyOpen( + controllerIdentity: string, + operationVersion: number, + result: CommandResult | null, + correlateResult: boolean, + throwDeviceError: boolean + ): Promise { + if (!this.canOpen(controllerIdentity, operationVersion)) { + this.device.close(); + await this.publishInvalidatedOpen(controllerIdentity, result, correlateResult); + return; + } + const signal = this.operationAbortController.signal; + try { + await this.openDevice(signal, operationVersion); + } catch (error) { + const operationAborted = signal.aborted || isOperationAborted(error); + if (!operationAborted && this.isOperationCurrent(controllerIdentity, operationVersion)) { + this.invalidateAndClose('device_apply_failed', true); + await this.publishOpenFailure( + controllerIdentity, + result, + 'device_apply_failed', + correlateResult + ); + } else { + this.device.close(); + await this.publishInvalidatedOpen(controllerIdentity, result, correlateResult); + } + if (throwDeviceError && !operationAborted) throw error; + return; + } + + if (!this.canOpen(controllerIdentity, operationVersion)) { + this.device.close(); + await this.publishInvalidatedOpen(controllerIdentity, result, correlateResult); + return; + } + let actual: MediaGateDeviceState; + try { + actual = this.device.snapshot(); + } catch (error) { + this.invalidateAndClose('device_state_unavailable', true); + if (result) { + result.status = 'rejected'; + result.errorCode = 'device_state_unavailable'; + } + throw error; + } + if (!actual.captureActive || !actual.trackPublished || actual.trackMuted) { + this.invalidateAndClose('device_not_ready', true); + await this.publishOpenFailure( + controllerIdentity, + result, + 'device_not_ready', + correlateResult + ); + return; + } + this.appliedOpenVersion = operationVersion; + if (result) { + result.status = 'applied'; + result.errorCode = null; + } + if (result && correlateResult) { + try { + await this.publishCommandResult(controllerIdentity, result); + } catch (error) { + if (isOperationAborted(error)) { + await this.publishInvalidatedOpen(controllerIdentity, result, true); + return; + } + result.status = 'rejected'; + result.errorCode = 'state_publish_failed'; + throw error; + } + } else { + try { + await this.publishSnapshot(controllerIdentity, null, null); + } catch (error) { + if (isOperationAborted(error)) { + await this.publishInvalidatedOpen(controllerIdentity, result, false); + return; + } + if (result) { + result.status = 'rejected'; + result.errorCode = 'state_publish_failed'; + } + throw error; + } + } + } + + private async openDevice(signal: AbortSignal, operationVersion: number): Promise { + const openPromise = Promise.resolve().then(async () => { + if (signal.aborted) throw new OperationAbortedError(); + await this.device.open(signal); + }); + void openPromise.then( + () => { + if (signal.aborted && !this.hasNewerAppliedOpen(operationVersion)) { + this.device.close(); + } + }, + () => undefined + ); + await raceWithAbort(openPromise, signal); + } + + private expireLease(leaseId: number): void { + const lease = this.lease; + if (!lease || lease.id !== leaseId) return; + this.transitionActiveOpenResult('expired', 'lease_expired'); + this.lease = null; + this.desiredListening = 'closed'; + this.closedBlocker = 'lease_expired'; + this.advanceOperation(); + this.device.close(); + const controllerIdentity = this.controllerIdentity; + if (controllerIdentity) { + void this.enqueue(() => this.publishSnapshot(controllerIdentity, null, null)); + } + } + + private classifyCommand(command: MediaControlCommand): 'accepted' | 'duplicate' | 'rejected' { + if (this.retiredPolicyEpochs.has(command.policy_epoch)) return 'rejected'; + if (this.activePolicyEpoch === null) return 'accepted'; + if (this.activePolicyEpoch !== command.policy_epoch) return 'accepted'; + if (!this.lastOrderedCommand) return 'accepted'; + if (command.sequence > this.lastOrderedCommand.sequence) return 'accepted'; + if ( + command.sequence === this.lastOrderedCommand.sequence && + commandsEqual(command, this.lastOrderedCommand) + ) { + return 'duplicate'; + } + return 'rejected'; + } + + private rememberAcceptedCommand(command: MediaControlCommand): void { + if (this.activePolicyEpoch && this.activePolicyEpoch !== command.policy_epoch) { + this.retirePolicyEpoch(this.activePolicyEpoch); + this.lastCommand = null; + this.lastOrderedCommand = null; + } + this.activePolicyEpoch = command.policy_epoch; + this.lastOrderedCommand = command; + } + + private resetCommandOrdering(): void { + this.activePolicyEpoch = null; + this.retiredPolicyEpochs.clear(); + this.retiredPolicyEpochOrder.length = 0; + this.lastOrderedCommand = null; + this.lastCommand = null; + } + + private retirePolicyEpoch(policyEpoch: string): void { + if (this.retiredPolicyEpochOrder.length === MAX_RETIRED_POLICY_EPOCHS) { + const expired = this.retiredPolicyEpochOrder.shift(); + if (expired) this.retiredPolicyEpochs.delete(expired); + } + this.retiredPolicyEpochOrder.push(policyEpoch); + this.retiredPolicyEpochs.add(policyEpoch); + } + + private hasFreshOpenLease(): boolean { + const lease = this.lease; + if (!lease || this.desiredListening !== 'open') return false; + if (this.nowMonotonicMs() < lease.deadlineMonotonicMs) return true; + this.expireLease(lease.id); + return false; + } + + private canOpen(controllerIdentity: string, operationVersion: number): boolean { + return ( + this.isOperationCurrent(controllerIdentity, operationVersion) && + !this.userMuted && + this.hasFreshOpenLease() + ); + } + + private isOperationCurrent(controllerIdentity: string, operationVersion: number): boolean { + return ( + this.started && + !this.stopped && + this.controllerIdentity === controllerIdentity && + this.operationVersion === operationVersion + ); + } + + private invalidateAndClose(blocker: string, clearLease: boolean): void { + this.advanceOperation(); + this.desiredListening = 'closed'; + this.closedBlocker = blocker; + if (clearLease) this.clearLease(); + this.device.close(); + } + + private advanceOperation(): void { + this.appliedOpenVersion = null; + this.operationAbortController.abort(); + this.operationAbortController = new AbortController(); + this.operationVersion += 1; + } + + private hasNewerAppliedOpen(abortedOperationVersion: number): boolean { + if ( + this.operationVersion === abortedOperationVersion || + this.appliedOpenVersion !== this.operationVersion || + !this.started || + this.stopped || + !this.controllerIdentity || + this.userMuted || + this.desiredListening !== 'open' + ) { + return false; + } + const lease = this.lease; + if (!lease) return false; + try { + return this.nowMonotonicMs() < lease.deadlineMonotonicMs; + } catch { + return false; + } + } + + private clearLease(): void { + if (!this.lease) return; + this.scheduler.clearTimeout(this.lease.timer); + this.lease = null; + } + + private publishCurrentWhenBound(): Promise { + const controllerIdentity = this.controllerIdentity; + if (!controllerIdentity) return this.drain(); + return this.enqueue(() => this.publishSnapshot(controllerIdentity, null, null)); + } + + private publishCommandResult(controllerIdentity: string, result: CommandResult): Promise { + if (result.status === null) { + throw new Error('media command result is not settled'); + } + return this.publishSnapshot(controllerIdentity, result, result.errorCode); + } + + private publishInvalidatedOpen( + controllerIdentity: string, + result: CommandResult | null, + correlateResult: boolean + ): Promise { + const { status, errorCode } = this.openInvalidationResult(); + return this.publishOpenFailure(controllerIdentity, result, errorCode, correlateResult, status); + } + + private publishOpenFailure( + controllerIdentity: string, + result: CommandResult | null, + errorCode: string, + correlateResult: boolean, + status: CommandStatus = 'rejected' + ): Promise { + if (this.controllerIdentity !== controllerIdentity || this.stopped) { + if (result) { + result.status = status; + result.errorCode = errorCode; + } + return Promise.resolve(); + } + if (result) { + result.status = status; + result.errorCode = errorCode; + } + if (result && correlateResult) { + return this.publishCommandResult(controllerIdentity, result); + } + return this.publishSnapshot(controllerIdentity, null, errorCode); + } + + private activeOpenResult(): CommandResult | null { + const lease = this.lease; + const result = this.lastCommand; + if (!lease || !result || !commandsEqual(lease.command, result.command)) { + return null; + } + return result; + } + + private transitionActiveOpenResult(status: CommandStatus, errorCode: string): void { + const result = this.activeOpenResult(); + if (!result) return; + result.status = status; + result.errorCode = errorCode; + } + + private openInvalidationResult(): { + status: CommandStatus; + errorCode: string; + } { + if (this.userMuted) return { status: 'rejected', errorCode: 'user_muted' }; + if (this.closedBlocker === 'lease_expired') { + return { status: 'expired', errorCode: 'lease_expired' }; + } + return { status: 'rejected', errorCode: 'superseded' }; + } + + private async publishSnapshot( + controllerIdentity: string, + result: CommandResult | null, + errorCode: string | null + ): Promise { + if (this.controllerIdentity !== controllerIdentity || this.stopped) return; + const signal = this.operationAbortController.signal; + try { + const actual = this.device.snapshot(); + const correlation = result + ? { + command_id: result.command.command_id, + policy_epoch: result.command.policy_epoch, + command_sequence: result.command.sequence, + command_status: result.status, + } + : NULL_CORRELATION; + const state: MediaStateSnapshot = { + schema_version: 1, + type: MEDIA_STATE_TOPIC, + target_identity: this.targetIdentity, + state_epoch: this.stateEpoch, + state_sequence: this.nextStateSequence(), + observed_at_unix_ms: this.safeNowUnixMs(), + capture_active: actual.captureActive, + track_published: actual.trackPublished, + track_muted: actual.trackMuted, + user_muted: this.userMuted, + blocked_by: this.blockedBy(actual), + ...correlation, + error_code: errorCode, + }; + const publishPromise = Promise.resolve().then(() => { + if (signal.aborted) throw new OperationAbortedError(); + return this.publishStateSink(controllerIdentity, state, signal); + }); + await raceWithAbort(publishPromise, signal); + } catch (error) { + if (signal.aborted || isOperationAborted(error)) { + throw new OperationAbortedError(); + } + this.invalidateAndClose('state_publish_failed', true); + throw error; + } + } + + private blockedBy(actual: MediaGateDeviceState): readonly string[] { + const blockers: string[] = []; + if (this.userMuted) blockers.push('user_muted'); + if (!this.started || this.stopped) blockers.push('stopped'); + if (!this.controllerIdentity) blockers.push('controller_disconnected'); + if (this.desiredListening !== 'open') blockers.push(this.closedBlocker); + if (!actual.captureActive) blockers.push('capture_inactive'); + if (!actual.trackPublished) blockers.push('track_unpublished'); + return [...new Set(blockers)]; + } + + private nextStateSequence(): number { + if (this.stateSequence >= Number.MAX_SAFE_INTEGER) { + throw new Error('media state sequence exhausted'); + } + this.stateSequence += 1; + return this.stateSequence; + } + + private safeNowUnixMs(): number { + const value = this.nowUnixMs(); + if (!Number.isSafeInteger(value) || value < 0) { + throw new Error('wall clock must return a non-negative safe integer'); + } + return value; + } + + private enqueue(action: () => Promise): Promise { + const next = this.tail.catch(() => undefined).then(action); + this.tail = next.catch(() => undefined); + return next; + } + + private requireStarted(): void { + if (!this.started || this.stopped) throw new Error('media gate executor is not running'); + } +} + +function commandsEqual(left: MediaControlCommand, right: MediaControlCommand): boolean { + return ( + left.command_id === right.command_id && + left.policy_epoch === right.policy_epoch && + left.sequence === right.sequence && + left.target_identity === right.target_identity && + left.desired_listening === right.desired_listening && + left.issued_at_unix_ms === right.issued_at_unix_ms && + left.expires_at_unix_ms === right.expires_at_unix_ms && + left.reason === right.reason + ); +} + +function requireNonempty(value: string, label: string): void { + if (!value) throw new Error(`${label} must be non-empty`); +} + +class OperationAbortedError extends Error { + constructor() { + super('media gate operation aborted'); + this.name = 'AbortError'; + } +} + +function isOperationAborted(error: unknown): boolean { + return error instanceof OperationAbortedError; +} + +function raceWithAbort(promise: Promise, signal: AbortSignal): Promise { + if (signal.aborted) return Promise.reject(new OperationAbortedError()); + return new Promise((resolve, reject) => { + const onAbort = () => { + reject(new OperationAbortedError()); + }; + signal.addEventListener('abort', onAbort, { once: true }); + promise.then( + (value) => { + signal.removeEventListener('abort', onAbort); + resolve(value); + }, + (error: unknown) => { + signal.removeEventListener('abort', onAbort); + reject(error); + } + ); + }); +} diff --git a/tests/browser-audio-gate-device.test.mjs b/tests/browser-audio-gate-device.test.mjs new file mode 100644 index 000000000..7d462315c --- /dev/null +++ b/tests/browser-audio-gate-device.test.mjs @@ -0,0 +1,422 @@ +import assert from 'node:assert/strict'; +import { test } from 'node:test'; + +const { BrowserAudioGateDevice } = await import('../lib/browser-audio-gate-device.ts'); +const { replaceRuntimeAudioBinding } = await import('../lib/browser-source-runtime-lifecycle.ts'); +const { MediaGateExecutor } = await import('../lib/media-gate-executor.ts'); + +function deferred() { + let resolve; + const promise = new Promise((resolvePromise) => { + resolve = resolvePromise; + }); + return { promise, resolve }; +} + +class FakeClock { + wall = 1_000; + monotonic = 10; + nextTimer = 1; + timers = new Map(); + + setTimeout(callback, delayMs) { + const handle = this.nextTimer++; + this.timers.set(handle, { callback, deadline: this.monotonic + delayMs }); + return handle; + } + + clearTimeout(handle) { + this.timers.delete(handle); + } + + advance(ms) { + this.wall += ms; + this.monotonic += ms; + for (;;) { + const next = [...this.timers.entries()] + .filter(([, timer]) => timer.deadline <= this.monotonic) + .sort((left, right) => left[1].deadline - right[1].deadline)[0]; + if (!next) return; + this.timers.delete(next[0]); + next[1].callback(); + } + } +} + +function fakeTrack(events, { unmuteBarrier = null } = {}) { + return { + isMuted: true, + mediaStreamTrack: { enabled: false, readyState: 'live' }, + async mute() { + events.push('mute'); + this.isMuted = true; + this.mediaStreamTrack.enabled = false; + }, + async unmute() { + events.push('unmute:start'); + this.isMuted = false; + this.mediaStreamTrack.enabled = true; + if (unmuteBarrier) await unmuteBarrier.promise; + events.push('unmute:done'); + }, + }; +} + +function command(overrides = {}) { + return { + schema_version: 1, + type: 'lk.media.control', + command_id: 'command-1', + policy_epoch: 'policy-1', + sequence: 1, + target_identity: 'browser-1', + desired_listening: 'open', + issued_at_unix_ms: 1_000, + expires_at_unix_ms: 2_000, + reason: 'face_present', + ...overrides, + }; +} + +function harness({ trackOptions } = {}) { + const events = []; + const clock = new FakeClock(); + let binding = null; + let publishes = 0; + const track = fakeTrack(events, trackOptions); + const device = new BrowserAudioGateDevice({ + getBinding: () => binding, + ensurePublishedClosed: async (signal) => { + if (binding) return; + events.push('publish:start'); + assert.equal(signal.aborted, false); + track.mediaStreamTrack.enabled = false; + track.isMuted = true; + binding = { track, publication: { isMuted: true } }; + publishes += 1; + events.push('publish:closed'); + }, + }); + const states = []; + const executor = new MediaGateExecutor({ + targetIdentity: 'browser-1', + device, + publishState: async (_controllerIdentity, state) => states.push(structuredClone(state)), + uuid: () => 'state-epoch-1', + nowUnixMs: () => clock.wall, + nowMonotonicMs: () => clock.monotonic, + scheduler: clock, + maxOpenLeaseMs: 3_000, + }); + return { + clock, + device, + events, + executor, + getBinding: () => binding, + getPublishes: () => publishes, + replaceBinding(next) { + binding = next; + }, + states, + track, + }; +} + +test('publishes a missing track closed and only opens it through a valid executor command', async () => { + const { device, events, executor, getPublishes, states, track } = harness(); + + await executor.start(); + await executor.bindController('agent-1'); + assert.deepEqual(device.snapshot(), { + captureActive: false, + trackPublished: false, + trackMuted: true, + }); + + await executor.handleCommand('agent-1', command()); + + assert.equal(getPublishes(), 1); + assert.equal(track.mediaStreamTrack.enabled, true); + assert.equal(track.isMuted, false); + assert.deepEqual(events, ['publish:start', 'publish:closed', 'unmute:start', 'unmute:done']); + assert.equal(states.at(-1).command_status, 'applied'); + assert.equal(states.at(-1).track_muted, false); +}); + +test('close disables capture synchronously while LiveKit mute remains best effort', async () => { + const { device, events, replaceBinding, track } = harness(); + track.isMuted = false; + track.mediaStreamTrack.enabled = true; + replaceBinding({ track, publication: { isMuted: false } }); + + device.close(); + + assert.equal(track.mediaStreamTrack.enabled, false); + assert.deepEqual(device.snapshot(), { + captureActive: false, + trackPublished: true, + trackMuted: true, + }); + await Promise.resolve(); + assert.deepEqual(events, ['mute']); +}); + +test('close remains fail-safe when the LiveKit mute call throws synchronously', () => { + const track = fakeTrack([]); + track.mediaStreamTrack.enabled = true; + track.isMuted = false; + track.mute = () => { + throw new Error('mute failed'); + }; + const device = new BrowserAudioGateDevice({ + getBinding: () => ({ track, publication: {} }), + ensurePublishedClosed: async () => {}, + }); + + assert.doesNotThrow(() => device.close()); + assert.equal(track.mediaStreamTrack.enabled, false); + assert.equal(device.snapshot().trackMuted, true); +}); + +test('a later open waits for an in-flight LiveKit mute before unmuting', async () => { + const muteBarrier = deferred(); + const events = []; + const track = { + isMuted: false, + mediaStreamTrack: { enabled: true, readyState: 'live' }, + async mute() { + events.push('mute:start'); + await muteBarrier.promise; + this.isMuted = true; + this.mediaStreamTrack.enabled = false; + events.push('mute:done'); + }, + async unmute() { + events.push('unmute'); + this.isMuted = false; + this.mediaStreamTrack.enabled = true; + }, + }; + const binding = { track, publication: {} }; + const device = new BrowserAudioGateDevice({ + getBinding: () => binding, + ensurePublishedClosed: async () => {}, + }); + + device.close(); + let opened = false; + const opening = device.open(new AbortController().signal).then(() => { + opened = true; + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + assert.equal(opened, false); + assert.deepEqual(events, ['mute:start']); + + muteBarrier.resolve(); + await opening; + + assert.deepEqual(events, ['mute:start', 'mute:done', 'unmute']); + assert.equal(track.mediaStreamTrack.enabled, true); + assert.equal(track.isMuted, false); + assert.equal(device.snapshot().trackMuted, false); +}); + +test('user mute wins over an in-flight open and a later unmute cannot revive it', async () => { + const barrier = deferred(); + const { device, executor, states, track } = harness({ + trackOptions: { unmuteBarrier: barrier }, + }); + await executor.start(); + await executor.bindController('agent-1'); + + const opening = executor.handleCommand('agent-1', command()); + await new Promise((resolve) => setTimeout(resolve, 0)); + const muting = executor.setUserMuted(true); + assert.equal(track.mediaStreamTrack.enabled, false); + barrier.resolve(); + await Promise.all([opening, muting]); + + assert.equal(track.mediaStreamTrack.enabled, false); + assert.equal(device.snapshot().trackMuted, true); + assert.equal(states.at(-1).user_muted, true); + assert.notEqual(states.at(-1).command_status, 'applied'); +}); + +test('removing user mute only reopens while the same open lease is still fresh', async () => { + const { clock, device, executor, track } = harness(); + await executor.start(); + await executor.bindController('agent-1'); + await executor.setUserMuted(true); + await executor.handleCommand('agent-1', command({ expires_at_unix_ms: 5_000 })); + + await executor.setUserMuted(false); + assert.equal(device.snapshot().trackMuted, false); + + await executor.setUserMuted(true); + clock.advance(3_000); + await executor.setUserMuted(false); + await executor.drain(); + + assert.equal(track.mediaStreamTrack.enabled, false); + assert.equal(device.snapshot().trackMuted, true); +}); + +test('a track replacement remains closed until reconcile reapplies a fresh lease', async () => { + const { device, events, executor, replaceBinding } = harness(); + await executor.start(); + await executor.bindController('agent-1'); + await executor.handleCommand('agent-1', command()); + + device.close(); + const replacement = fakeTrack(events); + replacement.mediaStreamTrack.enabled = false; + replacement.isMuted = true; + replaceBinding({ track: replacement, publication: { isMuted: true } }); + assert.equal(device.snapshot().trackMuted, true); + + await executor.reconcileDevice(); + + assert.equal(replacement.mediaStreamTrack.enabled, true); + assert.equal(replacement.isMuted, false); +}); + +test('an old asynchronous open cannot leak after the bound track is replaced', async () => { + const barrier = deferred(); + const { device, replaceBinding, track } = harness({ + trackOptions: { unmuteBarrier: barrier }, + }); + const signal = new AbortController().signal; + const opening = device.open(signal); + await new Promise((resolve) => setTimeout(resolve, 0)); + + device.close(); + const replacement = fakeTrack([]); + replaceBinding({ track: replacement, publication: { isMuted: true } }); + barrier.resolve(); + + await assert.rejects(opening, { name: 'AbortError' }); + assert.equal(track.mediaStreamTrack.enabled, false); + assert.equal(replacement.mediaStreamTrack.enabled, false); + assert.equal(device.snapshot().trackMuted, true); +}); + +test('a late old open failure cannot overwrite a newer successful replacement open', async () => { + const barrier = deferred(); + const oldTrack = fakeTrack([], { unmuteBarrier: barrier }); + let binding = { track: oldTrack, publication: {} }; + const device = new BrowserAudioGateDevice({ + getBinding: () => binding, + ensurePublishedClosed: async () => {}, + }); + const oldOpen = device.open(new AbortController().signal); + await new Promise((resolve) => setTimeout(resolve, 0)); + + device.close(); + const replacement = fakeTrack([]); + binding = { track: replacement, publication: {} }; + await device.open(new AbortController().signal); + barrier.resolve(); + await assert.rejects(oldOpen, { name: 'AbortError' }); + + assert.equal(oldTrack.mediaStreamTrack.enabled, false); + assert.equal(replacement.mediaStreamTrack.enabled, true); + assert.deepEqual(device.snapshot(), { + captureActive: true, + trackPublished: true, + trackMuted: false, + }); +}); + +test('switching devices while user-muted removes A without publishing and the next open creates B', async () => { + const events = []; + let selectedDeviceId = 'A'; + const oldTrack = fakeTrack(events); + let binding = { track: oldTrack, publication: { deviceId: 'A' } }; + const createdDeviceIds = []; + const device = new BrowserAudioGateDevice({ + getBinding: () => binding, + ensurePublishedClosed: async () => { + const deviceId = selectedDeviceId; + const track = fakeTrack(events); + createdDeviceIds.push(deviceId); + binding = { track, publication: { deviceId } }; + }, + }); + const runtime = { audioEnabled: false, audioPublishPromise: null }; + + selectedDeviceId = 'B'; + await replaceRuntimeAudioBinding(runtime, { + close: () => device.close(), + unpublish: async () => { + events.push(`unpublish:${binding?.publication.deviceId}`); + binding = null; + }, + reconcile: async () => assert.fail('muted replacement must not reconcile open'), + ensurePublished: async () => assert.fail('muted replacement must not publish B'), + hasBinding: () => binding !== null, + }); + + assert.equal(binding, null); + assert.deepEqual(createdDeviceIds, []); + assert.equal(oldTrack.mediaStreamTrack.enabled, false); + + runtime.audioEnabled = true; + await device.open(new AbortController().signal); + + assert.deepEqual(createdDeviceIds, ['B']); + assert.equal(binding.publication.deviceId, 'B'); + assert.equal(binding.track.mediaStreamTrack.enabled, true); + assert.equal(binding.track.isMuted, false); +}); + +test('switching devices while listening reconciles a fresh lease onto B without restoring A', async () => { + const events = []; + const clock = new FakeClock(); + let selectedDeviceId = 'A'; + let binding = null; + const createdDeviceIds = []; + const device = new BrowserAudioGateDevice({ + getBinding: () => binding, + ensurePublishedClosed: async () => { + const deviceId = selectedDeviceId; + const track = fakeTrack(events); + createdDeviceIds.push(deviceId); + binding = { track, publication: { deviceId } }; + }, + }); + const executor = new MediaGateExecutor({ + targetIdentity: 'browser-1', + device, + publishState: async () => {}, + uuid: () => 'state-epoch-1', + nowUnixMs: () => clock.wall, + nowMonotonicMs: () => clock.monotonic, + scheduler: clock, + maxOpenLeaseMs: 3_000, + }); + const runtime = { audioEnabled: true, audioPublishPromise: null }; + await executor.start(); + await executor.bindController('agent-1'); + await executor.handleCommand('agent-1', command({ expires_at_unix_ms: 4_000 })); + const oldTrack = binding.track; + + selectedDeviceId = 'B'; + await replaceRuntimeAudioBinding(runtime, { + close: () => device.close(), + unpublish: async () => { + events.push(`unpublish:${binding?.publication.deviceId}`); + binding = null; + }, + reconcile: () => executor.reconcileDevice(), + ensurePublished: async () => assert.fail('fresh lease reconcile must create B itself'), + hasBinding: () => binding !== null, + }); + + assert.deepEqual(createdDeviceIds, ['A', 'B']); + assert.equal(oldTrack.mediaStreamTrack.enabled, false); + assert.equal(binding.publication.deviceId, 'B'); + assert.equal(binding.track.mediaStreamTrack.enabled, true); + assert.equal(binding.track.isMuted, false); +}); diff --git a/tests/browser-media-gate-wiring.test.mjs b/tests/browser-media-gate-wiring.test.mjs new file mode 100644 index 000000000..c96e44064 --- /dev/null +++ b/tests/browser-media-gate-wiring.test.mjs @@ -0,0 +1,81 @@ +import assert from 'node:assert/strict'; +import { readFile } from 'node:fs/promises'; +import { test } from 'node:test'; + +test('browser raw audio wires one media gate onto the existing room and local identity', async () => { + const source = await readFile('hooks/useBrowserSourceClient.ts', 'utf8'); + + assert.match(source, /new LiveKitMediaGateAdapter\(\{[\s\S]*room,[\s\S]*agentName/); + assert.match(source, /allowAnonymousLiveKitAgentFallback:\s*true/); + assert.match(source, /targetIdentity:\s*room\.localParticipant\.identity/); + assert.doesNotMatch(source, /new Room\(/); + assert.doesNotMatch(source, /room\.connect\(/); +}); + +test('browser audio starts closed before publish and starts control after device reconciliation', async () => { + const source = await readFile('hooks/useBrowserSourceClient.ts', 'utf8'); + const ensureAudio = source.match( + /const ensureAudioPublished = useCallback\([\s\S]*?(?=\n const ensureVideoPublished)/ + )?.[0]; + const start = source.match(/const start = useCallback\(async \(\) => \{[\s\S]*?\n \}, \[/)?.[0]; + + assert.ok(ensureAudio); + assert.ok(start); + assert.match( + ensureAudio, + /audioTrack\.mediaStreamTrack\.enabled = false[\s\S]*await audioTrack\.mute\(\)[\s\S]*publishTrack/ + ); + assert.match( + start, + /await stage\(\(\) => audioGate\.executor\.start\(\)\)[\s\S]*await stage\(\(\) => audioGate\.executor\.reconcileDevice\(\)\)[\s\S]*await stage\(\(\) => audioGate\.adapter\.start\(audioGate\.executor\)\)/ + ); +}); + +test('privacy mute, device replacement, and stop all close synchronously before async work', async () => { + const source = await readFile('hooks/useBrowserSourceClient.ts', 'utf8'); + const setAudioEnabled = source.match( + /const setAudioEnabled = useCallback\([\s\S]*?(?=\n const setAudioDeviceId)/ + )?.[0]; + const setAudioDeviceId = source.match( + /const setAudioDeviceId = useCallback\([\s\S]*?(?=\n const setVideoEnabled)/ + )?.[0]; + const stopRuntime = source.match( + /const stopRuntime = useCallback\([\s\S]*?(?=\n const stop = useCallback)/ + )?.[0]; + const stop = source.match(/const stop = useCallback\(async \(\) => \{[\s\S]*?\n \}, \[/)?.[0]; + + assert.ok(setAudioEnabled); + assert.ok(setAudioDeviceId); + assert.ok(stopRuntime); + assert.ok(stop); + assert.match( + setAudioEnabled, + /if \(!nextEnabled\) \{[\s\S]*audioGate\?\.device\.close\(\)[\s\S]*executor\?\.setUserMuted\(true\)/ + ); + assert.doesNotMatch(setAudioEnabled, /audioEnabledRef\.current = previousEnabled/); + assert.match(setAudioDeviceId, /replaceRuntimeAudioBinding\(runtime/); + assert.match( + stopRuntime, + /const gateStop = audioGate\.adapter\.stop\(\)[\s\S]*audioGate\.device\.close\(\)[\s\S]*await gateStop[\s\S]*unpublishAudio/ + ); + assert.match(stop, /detachCurrentRuntime\(runtimeRef\)[\s\S]*stopRuntime\(runtime\)/); +}); + +test('late start cleanup and media publication remain owned by their original runtime', async () => { + const source = await readFile('hooks/useBrowserSourceClient.ts', 'utf8'); + + assert.match(source, /runOwnedRuntimeStart\(runtimeRef, runtime, stopRuntime/); + assert.match( + source, + /ensurePublishedClosed:[\s\S]*ensureAudioPublished\(runtime\)[\s\S]*stage\([\s\S]*adapter\.start/ + ); + assert.match(source, /!isCurrentRuntime\(runtimeRef, runtime\)/); + assert.doesNotMatch(source, /catch \(error\) \{\s*await stop\(\)/); +}); + +test('the browser control lease cap is the backend contract value', async () => { + const source = await readFile('hooks/useBrowserSourceClient.ts', 'utf8'); + + assert.match(source, /const BROWSER_MEDIA_GATE_MAX_OPEN_LEASE_MS = 3000/); + assert.match(source, /maxOpenLeaseMs:\s*BROWSER_MEDIA_GATE_MAX_OPEN_LEASE_MS/); +}); diff --git a/tests/browser-source-runtime-lifecycle.test.mjs b/tests/browser-source-runtime-lifecycle.test.mjs new file mode 100644 index 000000000..e42581010 --- /dev/null +++ b/tests/browser-source-runtime-lifecycle.test.mjs @@ -0,0 +1,102 @@ +import assert from 'node:assert/strict'; +import { test } from 'node:test'; + +const { detachCurrentRuntime, isCurrentRuntime, runOwnedRuntimeStart, stopOwnedRuntime } = + await import('../lib/browser-source-runtime-lifecycle.ts'); + +function deferred() { + let resolve; + let reject; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, reject, resolve }; +} + +for (const settlement of ['resolve', 'reject']) { + test(`stopped runtime A cannot continue its start pipeline when a pending stage ${settlement}s after runtime B starts`, async () => { + const slot = { current: null }; + const pendingStage = deferred(); + const events = []; + const runtimeA = { id: 'A', stopPromise: null }; + const runtimeB = { id: 'B', stopPromise: null }; + const stopRuntime = (runtime) => { + if (!runtime.stopPromise) { + runtime.stopPromise = Promise.resolve().then(() => events.push(`stop:${runtime.id}`)); + } + return runtime.stopPromise; + }; + + slot.current = runtimeA; + const startA = runOwnedRuntimeStart(slot, runtimeA, stopRuntime, async (stage) => { + await stage(async () => { + events.push('create-or-publish:A'); + return pendingStage.promise; + }); + events.push('reconcile:A'); + await stage(async () => events.push('adapter:A')); + }); + await Promise.resolve(); + + const detachedA = detachCurrentRuntime(slot); + assert.equal(detachedA, runtimeA); + await stopRuntime(detachedA); + slot.current = runtimeB; + if (settlement === 'resolve') { + pendingStage.resolve(); + } else { + pendingStage.reject(new Error('late publish failure')); + } + + await assert.rejects(startA, (error) => error?.name === 'AbortError'); + assert.equal(slot.current, runtimeB); + assert.equal(runtimeB.stopPromise, null); + assert.deepEqual(events, ['create-or-publish:A', 'stop:A']); + }); +} + +test('a late failure from stopped runtime A cannot detach or stop replacement runtime B', async () => { + const slot = { current: null }; + const stopped = []; + const lateFailure = deferred(); + const runtimeA = { id: 'A', stopped: false }; + const runtimeB = { id: 'B', stopped: false }; + const stopRuntime = async (runtime) => { + if (runtime.stopped) return; + runtime.stopped = true; + stopped.push(runtime.id); + }; + + slot.current = runtimeA; + const startA = (async () => { + try { + await lateFailure.promise; + } catch (error) { + await stopOwnedRuntime(slot, runtimeA, stopRuntime); + throw error; + } + })(); + + const detachedA = detachCurrentRuntime(slot); + assert.equal(detachedA, runtimeA); + await stopRuntime(detachedA); + slot.current = runtimeB; + lateFailure.reject(new Error('A start failed late')); + await assert.rejects(startA, /A start failed late/); + + assert.equal(slot.current, runtimeB); + assert.equal(isCurrentRuntime(slot, runtimeA), false); + assert.equal(isCurrentRuntime(slot, runtimeB), true); + assert.equal(runtimeB.stopped, false); + assert.deepEqual(stopped, ['A']); +}); + +test('public stop atomically detaches the current runtime before asynchronous cleanup', async () => { + const slot = { current: { id: 'A' } }; + + const detached = detachCurrentRuntime(slot); + + assert.deepEqual(detached, { id: 'A' }); + assert.equal(slot.current, null); +}); diff --git a/tests/livekit-media-gate.test.mjs b/tests/livekit-media-gate.test.mjs new file mode 100644 index 000000000..e11cfb1b8 --- /dev/null +++ b/tests/livekit-media-gate.test.mjs @@ -0,0 +1,512 @@ +import assert from 'node:assert/strict'; +import { EventEmitter } from 'node:events'; +import { test } from 'node:test'; + +const { RoomEvent } = await import('livekit-client'); +const { LiveKitMediaGateAdapter } = await import('../lib/livekit-media-gate.ts'); +const { MEDIA_CONTROL_TOPIC, MEDIA_STATE_TOPIC } = await import('../lib/media-control-protocol.ts'); + +const textEncoder = new TextEncoder(); +const textDecoder = new TextDecoder(); + +class FakeRoom extends EventEmitter { + constructor(participants = []) { + super(); + this.remoteParticipants = new Map( + participants.map((participant) => [participant.identity, participant]) + ); + this.published = []; + this.publishGate = null; + this.localParticipant = { + publishData: async (payload, options) => { + this.published.push({ payload, options }); + await this.publishGate?.promise; + }, + }; + } + + connectParticipant(participant) { + this.remoteParticipants.set(participant.identity, participant); + this.emit(RoomEvent.ParticipantConnected, participant); + } + + disconnectParticipant(participant) { + this.remoteParticipants.delete(participant.identity); + this.emit(RoomEvent.ParticipantDisconnected, participant); + } + + sendData(payload, participant, topic) { + this.emit(RoomEvent.DataReceived, payload, participant, 0, topic); + } +} + +class FakeExecutor { + constructor({ targetIdentity = 'browser-edge', commandGate = null } = {}) { + this.targetIdentity = targetIdentity; + this.commandGate = commandGate; + this.calls = []; + this.appliedCommands = []; + } + + async start() { + this.calls.push(['start']); + } + + async bindController(identity) { + this.calls.push(['bindController', identity]); + } + + async handleCommand(identity, command) { + this.calls.push(['handleCommand', identity, command]); + await this.commandGate?.promise; + if (command.target_identity === this.targetIdentity) { + this.appliedCommands.push(command); + } + } + + async handleMalformedControl(identity, errorCode) { + this.calls.push(['handleMalformedControl', identity, errorCode]); + } + + async disconnectController(identity) { + this.calls.push(['disconnectController', identity]); + } + + async stop() { + this.calls.push(['stop']); + } +} + +function participant({ identity, agentName, isAgent = true }) { + const attributes = agentName === undefined ? {} : { 'lk.agent.name': agentName }; + return { identity, isAgent, attributes }; +} + +function command(overrides = {}) { + return { + schema_version: 1, + type: MEDIA_CONTROL_TOPIC, + command_id: 'command-1', + policy_epoch: 'policy-1', + sequence: 1, + target_identity: 'browser-edge', + desired_listening: 'open', + issued_at_unix_ms: 1_000, + expires_at_unix_ms: 2_000, + reason: 'face_present', + ...overrides, + }; +} + +function encodedCommand(overrides = {}) { + return textEncoder.encode(JSON.stringify(command(overrides))); +} + +function state(overrides = {}) { + return { + schema_version: 1, + type: MEDIA_STATE_TOPIC, + target_identity: 'browser-edge', + state_epoch: 'state-1', + state_sequence: 1, + observed_at_unix_ms: 1_100, + capture_active: false, + track_published: true, + track_muted: true, + user_muted: false, + blocked_by: ['lease_missing'], + command_id: null, + policy_epoch: null, + command_sequence: null, + command_status: null, + error_code: null, + ...overrides, + }; +} + +function deferred() { + let resolve; + let reject; + const promise = new Promise((promiseResolve, promiseReject) => { + resolve = promiseResolve; + reject = promiseReject; + }); + return { promise, resolve, reject }; +} + +function namedAgent(identity = 'agent-frontdesk') { + return participant({ identity, agentName: 'frontdesk-browser-agent' }); +} + +test('start discovers an existing named Agent and binds its exact identity', async () => { + const controller = namedAgent(); + const room = new FakeRoom([controller]); + const executor = new FakeExecutor(); + const adapter = new LiveKitMediaGateAdapter({ + room, + agentName: 'frontdesk-browser-agent', + }); + + await adapter.start(executor); + + assert.deepEqual(executor.calls, [['start'], ['bindController', controller.identity]]); +}); + +test('start rolls back listeners and stops permanently when initial controller binding fails', async () => { + const controller = namedAgent(); + const room = new FakeRoom([controller]); + const executor = new FakeExecutor(); + executor.bindController = async (identity) => { + executor.calls.push(['bindController', identity]); + throw new Error('initial bind failed'); + }; + const adapter = new LiveKitMediaGateAdapter({ + room, + agentName: 'frontdesk-browser-agent', + }); + + await assert.rejects(adapter.start(executor), /initial bind failed/); + + assert.equal(room.listenerCount(RoomEvent.DataReceived), 0); + assert.equal(room.listenerCount(RoomEvent.ParticipantConnected), 0); + assert.equal(room.listenerCount(RoomEvent.ParticipantDisconnected), 0); + assert.equal( + executor.calls.some(([name]) => name === 'stop'), + true + ); + await assert.rejects(adapter.start(executor), /cannot restart after stop/); +}); + +test('participant discovery requires Agent role and matching agent-name attributes', async () => { + const room = new FakeRoom(); + const executor = new FakeExecutor(); + const adapter = new LiveKitMediaGateAdapter({ + room, + agentName: 'frontdesk-browser-agent', + }); + await adapter.start(executor); + + room.connectParticipant( + participant({ identity: 'standard', agentName: 'frontdesk-browser-agent', isAgent: false }) + ); + room.connectParticipant(participant({ identity: 'wrong-agent', agentName: 'other-agent' })); + room.connectParticipant({ + identity: 'expected-agent', + isAgent: true, + attributes: { lkAgentName: 'frontdesk-browser-agent' }, + }); + await adapter.drain(); + + assert.deepEqual(executor.calls, [['start'], ['bindController', 'expected-agent']]); +}); + +test('a delayed participant-connected task cannot bind an Agent that already left the room', async () => { + const firstBindGate = deferred(); + const first = namedAgent('agent-first'); + const departed = namedAgent('agent-departed'); + const room = new FakeRoom(); + const executor = new FakeExecutor(); + executor.bindController = async (identity) => { + executor.calls.push(['bindController', identity]); + if (identity === first.identity) await firstBindGate.promise; + }; + const adapter = new LiveKitMediaGateAdapter({ + room, + agentName: 'frontdesk-browser-agent', + }); + await adapter.start(executor); + + room.connectParticipant(first); + await new Promise((resolve) => setImmediate(resolve)); + room.disconnectParticipant(first); + room.connectParticipant(departed); + room.disconnectParticipant(departed); + + firstBindGate.resolve(); + await adapter.drain(); + + assert.equal( + executor.calls.some( + ([name, identity]) => name === 'bindController' && identity === departed.identity + ), + false + ); +}); + +test('anonymous fallback binds only when exactly one unnamed Agent exists', async () => { + const soleAnonymous = participant({ identity: 'agent-anonymous' }); + const soleRoom = new FakeRoom([soleAnonymous]); + const soleExecutor = new FakeExecutor(); + const soleAdapter = new LiveKitMediaGateAdapter({ + room: soleRoom, + agentName: 'frontdesk-browser-agent', + allowAnonymousLiveKitAgentFallback: true, + }); + await soleAdapter.start(soleExecutor); + + assert.deepEqual(soleExecutor.calls, [['start'], ['bindController', soleAnonymous.identity]]); + + const ambiguousRoom = new FakeRoom([ + participant({ identity: 'agent-anonymous-1' }), + participant({ identity: 'agent-anonymous-2' }), + ]); + const ambiguousExecutor = new FakeExecutor(); + const ambiguousAdapter = new LiveKitMediaGateAdapter({ + room: ambiguousRoom, + agentName: 'frontdesk-browser-agent', + allowAnonymousLiveKitAgentFallback: true, + }); + await ambiguousAdapter.start(ambiguousExecutor); + + assert.deepEqual(ambiguousExecutor.calls, [['start']]); +}); + +test('anonymous fallback rejects an unnamed Agent outside the agent identity namespace', async () => { + const room = new FakeRoom([participant({ identity: 'worker-anonymous' })]); + const executor = new FakeExecutor(); + const adapter = new LiveKitMediaGateAdapter({ + room, + agentName: 'frontdesk-browser-agent', + allowAnonymousLiveKitAgentFallback: true, + }); + + await adapter.start(executor); + + assert.deepEqual(executor.calls, [['start']]); +}); + +test('exact topic is checked before decode and trusted malformed control fails closed', async () => { + const controller = namedAgent(); + const room = new FakeRoom([controller]); + const executor = new FakeExecutor(); + const adapter = new LiveKitMediaGateAdapter({ + room, + agentName: 'frontdesk-browser-agent', + }); + await adapter.start(executor); + + const malformed = textEncoder.encode('{'); + room.sendData(malformed, controller, 'other.topic'); + room.sendData(malformed, controller, MEDIA_CONTROL_TOPIC); + await adapter.drain(); + + assert.deepEqual(executor.calls.slice(2), [ + ['handleMalformedControl', controller.identity, 'malformed_control'], + ]); +}); + +test('unsupported control versions use a stable fail-closed error code', async () => { + const controller = namedAgent(); + const room = new FakeRoom([controller]); + const executor = new FakeExecutor(); + const adapter = new LiveKitMediaGateAdapter({ + room, + agentName: 'frontdesk-browser-agent', + }); + await adapter.start(executor); + + room.sendData(encodedCommand({ schema_version: 2 }), controller, MEDIA_CONTROL_TOPIC); + await adapter.drain(); + + assert.deepEqual(executor.calls.at(-1), [ + 'handleMalformedControl', + controller.identity, + 'unsupported_control_version', + ]); +}); + +test('controller identity is pinned and valid commands from every other sender are ignored', async () => { + const controller = namedAgent('agent-controller'); + const impostor = namedAgent('agent-impostor'); + const room = new FakeRoom([controller, impostor]); + const executor = new FakeExecutor(); + const adapter = new LiveKitMediaGateAdapter({ + room, + agentName: 'frontdesk-browser-agent', + }); + await adapter.start(executor); + + room.sendData(encodedCommand(), impostor, MEDIA_CONTROL_TOPIC); + room.sendData(encodedCommand(), controller, MEDIA_CONTROL_TOPIC); + await adapter.drain(); + + assert.equal(executor.calls.filter(([name]) => name === 'handleCommand').length, 1); + assert.deepEqual(executor.calls.at(-1).slice(0, 2), ['handleCommand', controller.identity]); +}); + +test('target validation stays in the executor boundary', async () => { + const controller = namedAgent(); + const room = new FakeRoom([controller]); + const executor = new FakeExecutor(); + const adapter = new LiveKitMediaGateAdapter({ + room, + agentName: 'frontdesk-browser-agent', + }); + await adapter.start(executor); + + room.sendData( + encodedCommand({ target_identity: 'different-edge' }), + controller, + MEDIA_CONTROL_TOPIC + ); + await adapter.drain(); + + assert.equal(executor.calls.filter(([name]) => name === 'handleCommand').length, 1); + assert.equal(executor.appliedCommands.length, 0); +}); + +test('state publishing is reliable, exact-topic, encoded, and directed only to the pinned controller', async () => { + const controller = namedAgent(); + const room = new FakeRoom([controller]); + const executor = new FakeExecutor(); + const adapter = new LiveKitMediaGateAdapter({ + room, + agentName: 'frontdesk-browser-agent', + }); + await adapter.start(executor); + + await adapter.publishState(controller.identity, state(), new AbortController().signal); + + assert.equal(room.published.length, 1); + assert.deepEqual(room.published[0].options, { + reliable: true, + destinationIdentities: [controller.identity], + topic: MEDIA_STATE_TOPIC, + }); + assert.deepEqual(JSON.parse(textDecoder.decode(room.published[0].payload)), state()); + + await adapter.publishState( + 'agent-impostor', + state({ state_sequence: 2 }), + new AbortController().signal + ); + assert.equal(room.published.length, 1); +}); + +test('an aborted state sink does not publish data', async () => { + const controller = namedAgent(); + const room = new FakeRoom([controller]); + const executor = new FakeExecutor(); + const adapter = new LiveKitMediaGateAdapter({ + room, + agentName: 'frontdesk-browser-agent', + }); + await adapter.start(executor); + const abortController = new AbortController(); + abortController.abort(); + + await assert.rejects( + adapter.publishState(controller.identity, state(), abortController.signal), + (error) => error?.name === 'AbortError' + ); + assert.equal(room.published.length, 0); +}); + +test('controller disconnect closes through executor, clears the pin, and permits a later rebind', async () => { + const first = namedAgent('agent-first'); + const room = new FakeRoom([first]); + const executor = new FakeExecutor(); + const adapter = new LiveKitMediaGateAdapter({ + room, + agentName: 'frontdesk-browser-agent', + }); + await adapter.start(executor); + + room.disconnectParticipant(first); + await adapter.drain(); + const second = namedAgent('agent-second'); + room.connectParticipant(second); + await adapter.drain(); + + assert.deepEqual(executor.calls, [ + ['start'], + ['bindController', first.identity], + ['disconnectController', first.identity], + ['bindController', second.identity], + ]); +}); + +test('controller disconnect is not blocked behind stuck command work', async () => { + const commandGate = deferred(); + const controller = namedAgent(); + const room = new FakeRoom([controller]); + const executor = new FakeExecutor({ commandGate }); + const adapter = new LiveKitMediaGateAdapter({ + room, + agentName: 'frontdesk-browser-agent', + }); + await adapter.start(executor); + room.sendData(encodedCommand(), controller, MEDIA_CONTROL_TOPIC); + await new Promise((resolve) => setImmediate(resolve)); + + room.disconnectParticipant(controller); + await new Promise((resolve) => setImmediate(resolve)); + + assert.equal( + executor.calls.some( + ([name, identity]) => name === 'disconnectController' && identity === controller.identity + ), + true + ); + + commandGate.resolve(); + await adapter.drain(); +}); + +test('stop removes all listeners, stops executor, and drains owned command work', async () => { + const commandGate = deferred(); + const controller = namedAgent(); + const room = new FakeRoom([controller]); + const executor = new FakeExecutor({ commandGate }); + const adapter = new LiveKitMediaGateAdapter({ + room, + agentName: 'frontdesk-browser-agent', + }); + await adapter.start(executor); + room.sendData(encodedCommand(), controller, MEDIA_CONTROL_TOPIC); + await new Promise((resolve) => setImmediate(resolve)); + + let stopped = false; + const stopPromise = adapter.stop().then(() => { + stopped = true; + }); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(stopped, false); + + commandGate.resolve(); + await stopPromise; + assert.equal(stopped, true); + assert.equal(room.listenerCount(RoomEvent.DataReceived), 0); + assert.equal(room.listenerCount(RoomEvent.ParticipantConnected), 0); + assert.equal(room.listenerCount(RoomEvent.ParticipantDisconnected), 0); + assert.equal( + executor.calls.some(([name]) => name === 'stop'), + true + ); + + const callsAfterStop = executor.calls.length; + room.sendData(encodedCommand(), controller, MEDIA_CONTROL_TOPIC); + assert.equal(executor.calls.length, callsAfterStop); +}); + +test('event promise failures are observed without unhandled rejection', async () => { + const errors = []; + const controller = namedAgent(); + const room = new FakeRoom([controller]); + const executor = new FakeExecutor(); + executor.handleCommand = async () => { + throw new Error('command failed'); + }; + const adapter = new LiveKitMediaGateAdapter({ + room, + agentName: 'frontdesk-browser-agent', + onError: (error) => errors.push(error), + }); + await adapter.start(executor); + + room.sendData(encodedCommand(), controller, MEDIA_CONTROL_TOPIC); + await adapter.drain(); + + assert.equal(errors.length, 1); + assert.match(errors[0].message, /command failed/); +}); diff --git a/tests/media-control-protocol.test.mjs b/tests/media-control-protocol.test.mjs new file mode 100644 index 000000000..199687103 --- /dev/null +++ b/tests/media-control-protocol.test.mjs @@ -0,0 +1,192 @@ +import assert from 'node:assert/strict'; +import { test } from 'node:test'; + +const protocol = await import('../lib/media-control-protocol.ts'); + +const MAX_SAFE_INTEGER = Number.MAX_SAFE_INTEGER; + +function mediaControl(overrides = {}) { + return { + schema_version: 1, + type: 'lk.media.control', + command_id: 'command-1', + policy_epoch: 'policy-1', + sequence: 3, + target_identity: 'browser-1', + desired_listening: 'closed', + issued_at_unix_ms: 2_000, + expires_at_unix_ms: 2_750, + reason: 'face_absent', + ...overrides, + }; +} + +function mediaState(overrides = {}) { + return { + schema_version: 1, + type: 'lk.media.state', + target_identity: 'browser-1', + state_epoch: 'state-1', + state_sequence: 11, + observed_at_unix_ms: 2_100, + capture_active: true, + track_published: true, + track_muted: true, + user_muted: false, + blocked_by: ['face_absent'], + command_id: 'command-1', + policy_epoch: 'policy-1', + command_sequence: 3, + command_status: 'applied', + error_code: null, + ...overrides, + }; +} + +function json(value) { + return JSON.stringify(value); +} + +test('exports the locked media topics and decodes deterministic v1 control fields', () => { + assert.equal(protocol.MEDIA_CONTROL_TOPIC, 'lk.media.control'); + assert.equal(protocol.MEDIA_STATE_TOPIC, 'lk.media.state'); + + const expected = mediaControl(); + assert.deepEqual( + protocol.decodeMediaControl(json({ ...expected, future_extension: true })), + expected + ); + assert.deepEqual(protocol.decodeMediaControl(new TextEncoder().encode(json(expected))), expected); +}); + +test('maps an absent optional reason to null and rejects explicit null', () => { + const withoutReason = mediaControl(); + delete withoutReason.reason; + + assert.deepEqual(protocol.decodeMediaControl(json(withoutReason)), { + ...withoutReason, + reason: null, + }); + assert.throws(() => protocol.decodeMediaControl(json(mediaControl({ reason: null })))); +}); + +test('rejects malformed JSON, duplicate fields, non-objects, and invalid UTF-8', () => { + assert.throws(() => protocol.decodeMediaControl('{')); + assert.throws(() => protocol.decodeMediaControl('[]')); + assert.throws(() => + protocol.decodeMediaControl('{"schema_version":1,"schema_version":1,"type":"lk.media.control"}') + ); + assert.throws(() => protocol.decodeMediaControl(Uint8Array.of(0xff))); +}); + +test('rejects UTF-8 bytes prefixed with a byte-order mark', () => { + const jsonBytes = new TextEncoder().encode(json(mediaControl())); + const payload = new Uint8Array(3 + jsonBytes.byteLength); + payload.set([0xef, 0xbb, 0xbf]); + payload.set(jsonBytes, 3); + + assert.throws(() => protocol.decodeMediaControl(payload), /valid JSON/); +}); + +test('enforces the 16 KiB packet cap by UTF-8 bytes before decoding', () => { + assert.throws(() => protocol.decodeMediaControl(new Uint8Array(16 * 1024 + 1)), /16 KiB/); + assert.throws(() => protocol.decodeMediaControl('é'.repeat(8_193)), /16 KiB/); +}); + +test('rejects missing, unsupported, and wrongly typed envelopes', () => { + const missingVersion = mediaControl(); + delete missingVersion.schema_version; + + assert.throws(() => protocol.decodeMediaControl(json(missingVersion))); + assert.throws(() => protocol.decodeMediaControl(json(mediaControl({ schema_version: 2 })))); + assert.throws(() => protocol.decodeMediaControl(json(mediaControl({ schema_version: true })))); + assert.throws(() => protocol.decodeMediaControl(json(mediaControl({ type: 'lk.media.state' })))); +}); + +test('accepts the JS-safe integer boundary and rejects unsafe integer tokens', () => { + assert.equal( + protocol.decodeMediaControl(json(mediaControl({ sequence: MAX_SAFE_INTEGER }))).sequence, + MAX_SAFE_INTEGER + ); + + const payload = json(mediaControl()).replace( + '"sequence":3', + `"sequence":${MAX_SAFE_INTEGER + 1}` + ); + assert.throws(() => protocol.decodeMediaControl(payload)); + assert.throws(() => + protocol.decodeMediaControl(`${json(mediaControl()).slice(0, -1)},"future":9007199254740992}`) + ); +}); + +test('requires integer tokens, exact field types, and a future expiry', () => { + assert.throws(() => + protocol.decodeMediaControl(json(mediaControl()).replace('"sequence":3', '"sequence":3.0')) + ); + assert.throws(() => protocol.decodeMediaControl(json(mediaControl({ sequence: true })))); + assert.throws(() => protocol.decodeMediaControl(json(mediaControl({ command_id: '' })))); + assert.throws(() => + protocol.decodeMediaControl(json(mediaControl({ desired_listening: 'paused' }))) + ); + assert.throws(() => protocol.decodeMediaControl(json(mediaControl({ issued_at_unix_ms: -1 })))); + assert.throws(() => + protocol.decodeMediaControl(json(mediaControl({ expires_at_unix_ms: 2_000 }))) + ); +}); + +for (const field of ['command_id', 'policy_epoch', 'target_identity', 'reason']) { + test(`rejects an escaped unpaired surrogate in command field ${field}`, () => { + const payload = json(mediaControl({ [field]: '\ud800' })); + + assert.match(payload, /\\ud800/); + assert.throws(() => protocol.decodeMediaControl(payload), /valid UTF-8 text/); + }); +} + +test('encodes media state as Python-compatible sorted compact UTF-8 JSON', () => { + const encoded = protocol.encodeMediaState(mediaState()); + + assert.ok(encoded instanceof Uint8Array); + assert.equal( + new TextDecoder().decode(encoded), + '{"blocked_by":["face_absent"],"capture_active":true,"command_id":"command-1","command_sequence":3,"command_status":"applied","error_code":null,"observed_at_unix_ms":2100,"policy_epoch":"policy-1","schema_version":1,"state_epoch":"state-1","state_sequence":11,"target_identity":"browser-1","track_muted":true,"track_published":true,"type":"lk.media.state","user_muted":false}' + ); +}); + +test('validates media state types, safe integers, blockers, and output size', () => { + assert.throws(() => protocol.encodeMediaState(mediaState({ state_sequence: 0 }))); + assert.throws(() => + protocol.encodeMediaState(mediaState({ state_sequence: MAX_SAFE_INTEGER + 1 })) + ); + assert.throws(() => protocol.encodeMediaState(mediaState({ capture_active: 1 }))); + assert.throws(() => protocol.encodeMediaState(mediaState({ blocked_by: 'face_absent' }))); + assert.throws(() => + protocol.encodeMediaState(mediaState({ blocked_by: ['face_absent', 'face_absent'] })) + ); + assert.throws(() => protocol.encodeMediaState(mediaState({ blocked_by: [''] }))); + assert.throws( + () => protocol.encodeMediaState(mediaState({ error_code: 'x'.repeat(17_000) })), + /16 KiB/ + ); +}); + +test('requires command correlation to be fully null or fully present', () => { + assert.doesNotThrow(() => + protocol.encodeMediaState( + mediaState({ + command_id: null, + policy_epoch: null, + command_sequence: null, + command_status: null, + error_code: 'capture_unavailable', + }) + ) + ); + assert.throws(() => protocol.encodeMediaState(mediaState({ command_id: null }))); + assert.throws(() => protocol.encodeMediaState(mediaState({ command_status: 'ignored' }))); +}); + +test('rejects text and state strings that cannot be represented as valid UTF-8', () => { + assert.throws(() => protocol.decodeMediaControl('\ud800')); + assert.throws(() => protocol.encodeMediaState(mediaState({ error_code: '\ud800' }))); +}); diff --git a/tests/media-gate-executor.test.mjs b/tests/media-gate-executor.test.mjs new file mode 100644 index 000000000..d07373d00 --- /dev/null +++ b/tests/media-gate-executor.test.mjs @@ -0,0 +1,828 @@ +import assert from 'node:assert/strict'; +import { test } from 'node:test'; + +const { MediaGateExecutor } = await import('../lib/media-gate-executor.ts'); + +function deferred() { + let resolve; + let reject; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, resolve, reject }; +} + +async function settlesWithin(promise, timeoutMs = 50) { + let timeout; + try { + return await Promise.race([ + promise, + new Promise((_, reject) => { + timeout = setTimeout(() => reject(new Error('operation did not settle')), timeoutMs); + }), + ]); + } finally { + clearTimeout(timeout); + } +} + +class FakeClock { + wall = 1_000; + monotonic = 10; + nextTimer = 1; + timers = new Map(); + + setTimeout(callback, delayMs) { + const handle = this.nextTimer++; + this.timers.set(handle, { + callback, + deadline: this.monotonic + delayMs, + }); + return handle; + } + + clearTimeout(handle) { + this.timers.delete(handle); + } + + advance(ms) { + this.wall += ms; + this.monotonic += ms; + for (;;) { + const next = [...this.timers.entries()] + .filter(([, timer]) => timer.deadline <= this.monotonic) + .sort((left, right) => left[1].deadline - right[1].deadline)[0]; + if (!next) return; + this.timers.delete(next[0]); + next[1].callback(); + } + } +} + +class FakeDevice { + events = []; + captureActive = true; + trackPublished = true; + trackMuted = false; + nextOpen = null; + + close() { + this.events.push('close'); + this.trackMuted = true; + } + + async open() { + this.events.push('open:start'); + if (this.nextOpen) await this.nextOpen.promise; + this.trackMuted = false; + this.events.push('open:done'); + } + + snapshot() { + return { + captureActive: this.captureActive, + trackPublished: this.trackPublished, + trackMuted: this.trackMuted, + }; + } +} + +function command(overrides = {}) { + return { + schema_version: 1, + type: 'lk.media.control', + command_id: 'command-1', + policy_epoch: 'policy-1', + sequence: 1, + target_identity: 'browser-1', + desired_listening: 'open', + issued_at_unix_ms: 1_000, + expires_at_unix_ms: 2_000, + reason: 'face_present', + ...overrides, + }; +} + +function harness(overrides = {}) { + const clock = new FakeClock(); + const device = new FakeDevice(); + const states = []; + const destinations = []; + const publishState = async (controllerIdentity, state) => { + destinations.push(controllerIdentity); + states.push(structuredClone(state)); + }; + const executor = new MediaGateExecutor({ + targetIdentity: 'browser-1', + device, + publishState, + uuid: () => 'state-epoch-1', + nowUnixMs: () => clock.wall, + nowMonotonicMs: () => clock.monotonic, + scheduler: clock, + maxOpenLeaseMs: 500, + ...overrides, + }); + return { clock, destinations, device, executor, states }; +} + +test('starts closed and only publishes after binding an exact controller', async () => { + const { destinations, device, executor, states } = harness(); + + const started = executor.start(); + assert.equal(device.trackMuted, true); + await started; + assert.deepEqual(states, []); + + await executor.bindController('agent-1'); + + assert.equal(states.length, 1); + assert.deepEqual(destinations, ['agent-1']); + assert.deepEqual(states[0], { + schema_version: 1, + type: 'lk.media.state', + target_identity: 'browser-1', + state_epoch: 'state-epoch-1', + state_sequence: 1, + observed_at_unix_ms: 1_000, + capture_active: true, + track_published: true, + track_muted: true, + user_muted: false, + blocked_by: ['lease_missing'], + command_id: null, + policy_epoch: null, + command_sequence: null, + command_status: null, + error_code: null, + }); +}); + +test('binds one exact controller and applies targeted open and close commands', async () => { + const { device, executor, states } = harness(); + await executor.start(); + await executor.bindController('agent-1'); + device.events.length = 0; + states.length = 0; + + await executor.handleCommand('agent-1', command()); + await executor.handleCommand( + 'agent-1', + command({ + command_id: 'command-2', + sequence: 2, + desired_listening: 'closed', + reason: 'face_absent', + }) + ); + + assert.deepEqual(device.events, ['open:start', 'open:done', 'close']); + assert.deepEqual( + states.map((state) => [ + state.state_sequence, + state.command_id, + state.command_status, + state.track_muted, + ]), + [ + [2, 'command-1', 'applied', false], + [3, 'command-2', 'applied', true], + ] + ); +}); + +test('drops commands from another controller or for another target without side effects or ACKs', async () => { + const { device, executor, states } = harness(); + await executor.start(); + await executor.bindController('agent-1'); + device.events.length = 0; + states.length = 0; + + await executor.handleCommand('agent-2', command()); + await executor.handleCommand('agent-1', command({ target_identity: 'browser-2' })); + + assert.deepEqual(device.events, []); + assert.deepEqual(states, []); + assert.equal(device.trackMuted, true); +}); + +test('ACKs an exact duplicate without repeating the device side effect', async () => { + const { device, executor, states } = harness(); + await executor.start(); + await executor.bindController('agent-1'); + device.events.length = 0; + states.length = 0; + + await executor.handleCommand('agent-1', command()); + await executor.handleCommand('agent-1', command()); + + assert.deepEqual(device.events, ['open:start', 'open:done']); + assert.deepEqual( + states.map((state) => [state.state_sequence, state.command_status]), + [ + [2, 'applied'], + [3, 'applied'], + ] + ); +}); + +test('rejects stale sequences and retired epochs while closing fail-safe', async () => { + const { device, executor, states } = harness(); + await executor.start(); + await executor.bindController('agent-1'); + await executor.handleCommand('agent-1', command({ sequence: 5 })); + states.length = 0; + device.events.length = 0; + + await executor.handleCommand( + 'agent-1', + command({ command_id: 'stale', sequence: 4, desired_listening: 'closed' }) + ); + const eventsAfterStale = device.events.length; + await executor.handleCommand('agent-1', command({ sequence: 5 })); + assert.equal(device.events.length, eventsAfterStale); + await executor.handleCommand( + 'agent-1', + command({ command_id: 'new-policy', policy_epoch: 'policy-2', sequence: 1 }) + ); + await executor.handleCommand( + 'agent-1', + command({ command_id: 'retired', policy_epoch: 'policy-1', sequence: 6 }) + ); + + assert.equal(device.trackMuted, true); + assert.equal(states[0].command_status, 'rejected'); + assert.equal( + states.filter((state) => state.command_id !== null).at(-1).command_status, + 'rejected' + ); + assert.equal(states.at(-1).command_status, null); + assert.ok(device.events.includes('close')); +}); + +test('expires a command by wall time without opening', async () => { + const { clock, device, executor, states } = harness(); + await executor.start(); + await executor.bindController('agent-1'); + device.events.length = 0; + states.length = 0; + clock.wall = 2_001; + + await executor.handleCommand('agent-1', command()); + + assert.deepEqual(device.events, ['close']); + assert.equal(states[0].command_status, 'expired'); + assert.equal(states[0].track_muted, true); +}); + +test('bounds an open lease on the monotonic clock and closes when it expires', async () => { + const { clock, device, executor, states } = harness(); + await executor.start(); + await executor.bindController('agent-1'); + await executor.handleCommand('agent-1', command({ expires_at_unix_ms: 10_000 })); + states.length = 0; + + clock.wall = 100; + clock.advance(499); + await executor.drain(); + assert.equal(device.trackMuted, false); + clock.advance(1); + assert.equal(device.trackMuted, true); + await executor.drain(); + + assert.deepEqual( + [ + states.at(-1).command_id, + states.at(-1).policy_epoch, + states.at(-1).command_sequence, + states.at(-1).command_status, + ], + [null, null, null, null] + ); + assert.ok(states.at(-1).blocked_by.includes('lease_expired')); +}); + +test('a renewal cancels the previous lease timer without a close race', async () => { + const { clock, device, executor } = harness(); + await executor.start(); + await executor.bindController('agent-1'); + await executor.handleCommand('agent-1', command({ expires_at_unix_ms: 1_400 })); + clock.advance(300); + device.events.length = 0; + + await executor.handleCommand( + 'agent-1', + command({ command_id: 'renew', sequence: 2, expires_at_unix_ms: 1_800 }) + ); + clock.advance(100); + await executor.drain(); + + assert.equal(device.trackMuted, false); + assert.ok(!device.events.includes('close')); +}); + +test('privacy mute closes synchronously and wins an in-flight open', async () => { + const { device, executor, states } = harness(); + device.nextOpen = deferred(); + await executor.start(); + await executor.bindController('agent-1'); + states.length = 0; + device.events.length = 0; + const opening = executor.handleCommand('agent-1', command()); + await new Promise((resolve) => setImmediate(resolve)); + + const privacy = executor.setUserMuted(true); + assert.equal(device.trackMuted, true); + assert.deepEqual(device.events, ['open:start', 'close']); + device.nextOpen.resolve(); + await opening; + await privacy; + + assert.equal(device.trackMuted, true); + assert.equal( + states.some((state) => state.track_muted === false), + false + ); + assert.equal(states[0].command_id, 'command-1'); + assert.equal(states[0].command_status, 'rejected'); + assert.equal(states[0].error_code, 'user_muted'); + assert.deepEqual( + [ + states.at(-1).command_id, + states.at(-1).policy_epoch, + states.at(-1).command_sequence, + states.at(-1).command_status, + ], + [null, null, null, null] + ); + assert.equal(states.at(-1).user_muted, true); + assert.ok(states.at(-1).blocked_by.includes('user_muted')); +}); + +test('remote open cannot clear privacy and unmute only follows a fresh automatic lease', async () => { + const { clock, device, executor, states } = harness(); + await executor.start(); + await executor.bindController('agent-1'); + await executor.setUserMuted(true); + device.events.length = 0; + + await executor.handleCommand('agent-1', command({ expires_at_unix_ms: 1_400 })); + assert.equal(device.trackMuted, true); + assert.equal(states.at(-1).user_muted, true); + assert.equal(states.at(-1).command_status, 'rejected'); + assert.equal(states.at(-1).error_code, 'user_muted'); + await executor.setUserMuted(false); + assert.equal(device.trackMuted, false); + assert.deepEqual( + [ + states.at(-1).command_id, + states.at(-1).policy_epoch, + states.at(-1).command_sequence, + states.at(-1).command_status, + ], + [null, null, null, null] + ); + const eventsAfterUnmute = device.events.length; + await executor.handleCommand('agent-1', command({ expires_at_unix_ms: 1_400 })); + assert.equal(device.events.length, eventsAfterUnmute); + assert.equal(states.at(-1).command_status, 'applied'); + assert.equal(states.at(-1).error_code, null); + + clock.advance(400); + await executor.drain(); + await executor.setUserMuted(true); + device.events.length = 0; + await executor.setUserMuted(false); + assert.equal(device.trackMuted, true); + assert.equal(device.events.includes('open:start'), false); +}); + +test('controller disconnect and stop close synchronously and defeat late open completion', async () => { + for (const action of ['disconnect', 'stop']) { + const { device, executor, states } = harness(); + device.nextOpen = deferred(); + await executor.start(); + await executor.bindController('agent-1'); + states.length = 0; + const opening = executor.handleCommand('agent-1', command()); + await new Promise((resolve) => setImmediate(resolve)); + + const closing = + action === 'disconnect' ? executor.disconnectController('agent-1') : executor.stop(); + assert.equal(device.trackMuted, true); + device.nextOpen.resolve(); + await opening; + await closing; + + assert.equal(device.trackMuted, true, action); + assert.equal( + states.some((state) => state.track_muted === false), + false, + action + ); + } +}); + +test('a newer close command defeats late completion from an older open', async () => { + const { device, executor, states } = harness(); + device.nextOpen = deferred(); + await executor.start(); + await executor.bindController('agent-1'); + states.length = 0; + const opening = executor.handleCommand('agent-1', command()); + await new Promise((resolve) => setImmediate(resolve)); + + const closing = executor.handleCommand( + 'agent-1', + command({ command_id: 'close', sequence: 2, desired_listening: 'closed' }) + ); + assert.equal(device.trackMuted, true); + device.nextOpen.resolve(); + await opening; + await closing; + + assert.equal(device.trackMuted, true); + assert.equal( + states.some((state) => state.track_muted === false), + false + ); + assert.equal(states.at(-1).command_id, 'close'); + assert.equal(states.at(-1).command_status, 'applied'); + assert.deepEqual( + states.map((state) => [state.command_id, state.command_status]), + [ + ['command-1', 'rejected'], + ['close', 'applied'], + ] + ); +}); + +test('an in-flight duplicate waits for the exact command final result', async () => { + const { device, executor, states } = harness(); + device.nextOpen = deferred(); + await executor.start(); + await executor.bindController('agent-1'); + states.length = 0; + device.events.length = 0; + const opening = executor.handleCommand('agent-1', command()); + await new Promise((resolve) => setImmediate(resolve)); + + const duplicate = executor.handleCommand('agent-1', command()); + device.nextOpen.reject(new Error('device failed')); + await assert.rejects(opening, /device failed/); + await duplicate; + + assert.deepEqual(device.events, ['open:start', 'close']); + assert.deepEqual( + states.map((state) => [state.command_id, state.command_status, state.error_code]), + [ + ['command-1', 'rejected', 'device_apply_failed'], + ['command-1', 'rejected', 'device_apply_failed'], + ] + ); +}); + +for (const transition of [ + { + name: 'local privacy mute', + status: 'rejected', + errorCode: 'user_muted', + apply: async ({ executor }) => executor.setUserMuted(true), + }, + { + name: 'local lease expiry', + status: 'expired', + errorCode: 'lease_expired', + apply: async ({ clock, executor }) => { + clock.advance(500); + await executor.drain(); + }, + }, + { + name: 'stale command fail-close', + status: 'rejected', + errorCode: 'superseded', + apply: async ({ executor }) => + executor.handleCommand( + 'agent-1', + command({ command_id: 'stale', sequence: 4, desired_listening: 'closed' }) + ), + }, +]) { + test(`duplicate reflects final result after ${transition.name}`, async () => { + const context = harness(); + const { device, executor, states } = context; + await executor.start(); + await executor.bindController('agent-1'); + const original = command({ sequence: 5, expires_at_unix_ms: 10_000 }); + await executor.handleCommand('agent-1', original); + states.length = 0; + device.events.length = 0; + + await transition.apply(context); + assert.ok( + states.some( + (state) => + state.command_id === null && + state.policy_epoch === null && + state.command_sequence === null && + state.command_status === null + ) + ); + const eventsAfterTransition = device.events.length; + await executor.handleCommand('agent-1', original); + + assert.equal(device.events.length, eventsAfterTransition); + assert.equal(states.at(-1).command_id, original.command_id); + assert.equal(states.at(-1).command_status, transition.status); + assert.equal(states.at(-1).error_code, transition.errorCode); + }); +} + +test('state publication failure while opening immediately fails closed', async () => { + let publishCount = 0; + const { device, executor } = harness({ + publishState: async () => { + publishCount += 1; + if (publishCount === 2) throw new Error('transport down'); + }, + }); + await executor.start(); + await executor.bindController('agent-1'); + + await assert.rejects(() => executor.handleCommand('agent-1', command()), /transport down/); + + assert.equal(device.trackMuted, true); + assert.equal(device.events.at(-1), 'close'); +}); + +test('invalid state construction after device open immediately fails closed', async () => { + let clockReads = 0; + const { device, executor } = harness({ + nowUnixMs: () => { + clockReads += 1; + return clockReads <= 3 ? 1_000 : Number.NaN; + }, + }); + await executor.start(); + await executor.bindController('agent-1'); + + await assert.rejects(() => executor.handleCommand('agent-1', command()), /wall clock/); + + assert.equal(device.trackMuted, true); + assert.equal(device.events.at(-1), 'close'); +}); + +test('device snapshot failure after open immediately fails closed', async () => { + const { device, executor } = harness(); + const snapshot = device.snapshot.bind(device); + let snapshotReads = 0; + device.snapshot = () => { + snapshotReads += 1; + if (snapshotReads === 2) throw new Error('snapshot failed'); + return snapshot(); + }; + await executor.start(); + await executor.bindController('agent-1'); + + await assert.rejects(() => executor.handleCommand('agent-1', command()), /snapshot failed/); + + assert.equal(device.trackMuted, true); + assert.equal(device.events.at(-1), 'close'); +}); + +test('privacy aborts an in-flight applied state publish and supersedes it with closed state', async () => { + const blockedPublish = deferred(); + const accepted = []; + let attemptedApplied = null; + const { device, executor } = harness({ + publishState: async (_controllerIdentity, state, signal) => { + if (state.command_status === 'applied' && state.command_id === 'command-1') { + attemptedApplied = { signal, state: structuredClone(state) }; + await blockedPublish.promise; + } + if (signal?.aborted) throw new Error('publish aborted'); + accepted.push(structuredClone(state)); + }, + }); + await executor.start(); + await executor.bindController('agent-1'); + accepted.length = 0; + const opening = executor.handleCommand('agent-1', command()); + await new Promise((resolve) => setImmediate(resolve)); + + const privacy = executor.setUserMuted(true); + assert.ok(attemptedApplied?.signal instanceof AbortSignal); + assert.equal(attemptedApplied.signal.aborted, true); + await settlesWithin(opening); + await settlesWithin(privacy); + + assert.equal(device.trackMuted, true); + assert.equal( + accepted.some((state) => state.track_muted === false), + false + ); + assert.deepEqual( + accepted.map((state) => [state.command_id, state.command_status]), + [ + ['command-1', 'rejected'], + [null, null], + ] + ); + assert.ok(accepted[0].state_sequence > attemptedApplied.state.state_sequence); +}); + +test('stop and drain abort a state sink that never settles', async () => { + const blockedPublish = deferred(); + let appliedSignal = null; + const { device, executor } = harness({ + publishState: async (_controllerIdentity, state, signal) => { + if (state.command_status !== 'applied') return; + appliedSignal = signal; + await blockedPublish.promise; + }, + }); + await executor.start(); + await executor.bindController('agent-1'); + const opening = executor.handleCommand('agent-1', command()); + await new Promise((resolve) => setImmediate(resolve)); + + const stopping = executor.stop(); + assert.ok(appliedSignal instanceof AbortSignal); + assert.equal(appliedSignal.aborted, true); + await settlesWithin(opening); + await settlesWithin(stopping); + await settlesWithin(executor.drain()); + + assert.equal(device.trackMuted, true); +}); + +test('lease expiry aborts a stuck open and preserves expired final result for duplicate', async () => { + const { clock, device, executor, states } = harness(); + device.nextOpen = deferred(); + await executor.start(); + await executor.bindController('agent-1'); + states.length = 0; + const original = command({ expires_at_unix_ms: 10_000 }); + const opening = executor.handleCommand('agent-1', original); + await new Promise((resolve) => setImmediate(resolve)); + + clock.advance(500); + await settlesWithin(opening); + await settlesWithin(executor.drain()); + await executor.handleCommand('agent-1', original); + + assert.equal(device.trackMuted, true); + assert.equal(states[0].command_status, 'expired'); + assert.equal(states[0].error_code, 'lease_expired'); + assert.equal(states.at(-1).command_status, 'expired'); + assert.equal(states.at(-1).error_code, 'lease_expired'); + + device.nextOpen.resolve(); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(device.trackMuted, true); + assert.equal(device.events.at(-1), 'close'); +}); + +test('newer close and stop are not blocked by a device open that never settles', async () => { + for (const action of ['close', 'stop']) { + const { device, executor, states } = harness(); + device.nextOpen = deferred(); + await executor.start(); + await executor.bindController('agent-1'); + states.length = 0; + const opening = executor.handleCommand('agent-1', command()); + await new Promise((resolve) => setImmediate(resolve)); + + const closing = + action === 'close' + ? executor.handleCommand( + 'agent-1', + command({ command_id: 'close', sequence: 2, desired_listening: 'closed' }) + ) + : executor.stop(); + await settlesWithin(opening); + await settlesWithin(closing); + await settlesWithin(executor.drain()); + + assert.equal(device.trackMuted, true, action); + if (action === 'close') { + assert.deepEqual( + states.map((state) => [state.command_id, state.command_status]), + [ + ['command-1', 'rejected'], + ['close', 'applied'], + ] + ); + } + } +}); + +test('late completion from an aborted open does not close a newer successful open', async () => { + const { device, executor, states } = harness(); + const oldOpen = deferred(); + device.nextOpen = oldOpen; + await executor.start(); + await executor.bindController('agent-1'); + states.length = 0; + const firstOpen = executor.handleCommand('agent-1', command()); + await new Promise((resolve) => setImmediate(resolve)); + + device.nextOpen = null; + const close = executor.handleCommand( + 'agent-1', + command({ command_id: 'close', sequence: 2, desired_listening: 'closed' }) + ); + await settlesWithin(firstOpen); + await settlesWithin(close); + await executor.handleCommand('agent-1', command({ command_id: 'new-open', sequence: 3 })); + assert.equal(device.trackMuted, false); + + oldOpen.resolve(); + await new Promise((resolve) => setImmediate(resolve)); + + assert.equal(device.trackMuted, false); + assert.equal(states.at(-1).command_id, 'new-open'); + assert.equal(states.at(-1).command_status, 'applied'); +}); + +test('a silent no-op open is rejected when actual device state is not ready', async () => { + const { device, executor, states } = harness(); + device.open = async () => { + device.events.push('open:no-op'); + }; + await executor.start(); + await executor.bindController('agent-1'); + states.length = 0; + + await executor.handleCommand('agent-1', command()); + + assert.equal(device.trackMuted, true); + assert.equal(states.at(-1).command_status, 'rejected'); + assert.equal(states.at(-1).error_code, 'device_not_ready'); +}); + +test('retired policy epoch replay history is bounded to sixteen entries', async () => { + const { executor, states } = harness(); + await executor.start(); + await executor.bindController('agent-1'); + for (let index = 1; index <= 18; index += 1) { + await executor.handleCommand( + 'agent-1', + command({ + command_id: `epoch-${index}`, + policy_epoch: `policy-${index}`, + desired_listening: 'closed', + }) + ); + } + states.length = 0; + + await executor.handleCommand( + 'agent-1', + command({ command_id: 'oldest-reused', policy_epoch: 'policy-1' }) + ); + + assert.equal(states.at(-1).command_status, 'applied'); +}); + +test('malformed trusted control fails closed', async () => { + const { device, executor, states } = harness(); + await executor.start(); + await executor.bindController('agent-1'); + await executor.handleCommand('agent-1', command()); + states.length = 0; + + const failed = executor.handleMalformedControl('agent-1', 'malformed_control'); + assert.equal(device.trackMuted, true); + await failed; + + assert.equal(states.at(-1).error_code, 'malformed_control'); +}); + +test('republished device state is reconciled closed before serialized work', async () => { + const { device, executor, states } = harness(); + await executor.start(); + await executor.bindController('agent-1'); + await executor.setUserMuted(true); + states.length = 0; + device.events.length = 0; + device.trackMuted = false; + + const reconciled = executor.reconcileDevice(); + assert.equal(device.trackMuted, true); + await reconciled; + + assert.deepEqual(device.events, ['close']); + assert.equal(states.at(-1).track_muted, true); + assert.deepEqual( + [ + states.at(-1).command_id, + states.at(-1).policy_epoch, + states.at(-1).command_sequence, + states.at(-1).command_status, + ], + [null, null, null, null] + ); +}); From d36d3fb5cab47ef4d35d9ed915a092123cfa8d36 Mon Sep 17 00:00:00 2001 From: guowei Date: Mon, 3 Aug 2026 15:30:51 +0800 Subject: [PATCH 2/4] fix: preserve browser gate controller identity --- hooks/useBrowserSourceClient.ts | 3 +++ lib/utils.ts | 4 ++++ tests/local-dispatch-config.test.mjs | 11 +++++++++++ 3 files changed, 18 insertions(+) diff --git a/hooks/useBrowserSourceClient.ts b/hooks/useBrowserSourceClient.ts index ff877452e..6f1dffc8d 100644 --- a/hooks/useBrowserSourceClient.ts +++ b/hooks/useBrowserSourceClient.ts @@ -394,6 +394,7 @@ export function useBrowserSourceClient( if (audioGate) { const gateStop = audioGate.adapter.stop(); const executorStop = audioGate.executor.stop(); + // Privacy closure is synchronous; listener teardown may finish afterward. audioGate.device.close(); await gateStop; await executorStop; @@ -463,6 +464,8 @@ export function useBrowserSourceClient( const adapter = new LiveKitMediaGateAdapter({ room, agentName, + // Existing LexVoice agents may omit lk.agent.name. The adapter accepts this + // only when exactly one anonymous agent exists, then pins that controller. allowAnonymousLiveKitAgentFallback: true, onError: (error) => console.warn('[browser-audio] media gate event failed', error), }); diff --git a/lib/utils.ts b/lib/utils.ts index d8a15b629..beb2f368b 100644 --- a/lib/utils.ts +++ b/lib/utils.ts @@ -344,6 +344,10 @@ export const getAppConfig = cache(async (headers: Headers): Promise = } } + // A blank sandbox override must not remove the controller identity required by + // Browser audio gating and agent dispatch. Re-derive it from the final input role. + config.agentName = resolveAgentNameForInputSource(config.inputSource, config.agentName); + return config; } catch (error) { console.error('ERROR: getAppConfig() - lib/utils.ts', error); diff --git a/tests/local-dispatch-config.test.mjs b/tests/local-dispatch-config.test.mjs index 029c4dc2b..3e5164204 100644 --- a/tests/local-dispatch-config.test.mjs +++ b/tests/local-dispatch-config.test.mjs @@ -37,6 +37,17 @@ test('frontend keeps explicit AGENT_NAME as an override', async () => { assert.equal(resolveAgentNameForInputSource('generic', 'custom-agent'), 'custom-agent'); }); +test('frontend restores a derived agent name after a blank sandbox override', async () => { + const { resolveAgentNameForInputSource } = await loadAppConfigModule(); + const utilsSource = await readFile('lib/utils.ts', 'utf8'); + + assert.equal(resolveAgentNameForInputSource('browser', ' '), 'lexvoice-browser-agent'); + assert.match( + utilsSource, + /config\.agentName = resolveAgentNameForInputSource\(config\.inputSource, config\.agentName\)/ + ); +}); + test('frontend exposes the server-owned voice session id to dispatch callers', async () => { const previousEnv = { ...process.env }; From b87e75eede13fe6ecc4f4ff742dd71a5308be32f Mon Sep 17 00:00:00 2001 From: guowei Date: Mon, 3 Aug 2026 16:07:11 +0800 Subject: [PATCH 3/4] test: cover blank sandbox agent override --- tests/local-dispatch-config.test.mjs | 36 ++++++++++++++++++++++------ 1 file changed, 29 insertions(+), 7 deletions(-) diff --git a/tests/local-dispatch-config.test.mjs b/tests/local-dispatch-config.test.mjs index 3e5164204..15c9a703d 100644 --- a/tests/local-dispatch-config.test.mjs +++ b/tests/local-dispatch-config.test.mjs @@ -38,14 +38,36 @@ test('frontend keeps explicit AGENT_NAME as an override', async () => { }); test('frontend restores a derived agent name after a blank sandbox override', async () => { - const { resolveAgentNameForInputSource } = await loadAppConfigModule(); - const utilsSource = await readFile('lib/utils.ts', 'utf8'); + const previousEnv = { ...process.env }; + const originalFetch = globalThis.fetch; - assert.equal(resolveAgentNameForInputSource('browser', ' '), 'lexvoice-browser-agent'); - assert.match( - utilsSource, - /config\.agentName = resolveAgentNameForInputSource\(config\.inputSource, config\.agentName\)/ - ); + try { + process.env.APP_CONFIG_ENDPOINT = 'https://config.example.test/app-config'; + process.env.SANDBOX_ID = 'sandbox-blank-agent'; + process.env.INPUT_SOURCE = 'browser'; + delete process.env.AGENT_NAME; + delete process.env.NEXT_PUBLIC_AGENT_NAME; + delete process.env.NEXT_PUBLIC_LEXVOICE_AGENT_NAME; + + globalThis.fetch = async (input, init) => { + assert.equal(String(input), process.env.APP_CONFIG_ENDPOINT); + assert.equal(new Headers(init?.headers).get('X-Sandbox-ID'), process.env.SANDBOX_ID); + return new Response( + JSON.stringify({ + agentName: { type: 'string', value: ' ' }, + }), + { headers: { 'content-type': 'application/json' } } + ); + }; + + const { getAppConfig } = await import(`../lib/utils.ts?blank-agent-override=${Date.now()}`); + const config = await getAppConfig(new Headers()); + + assert.equal(config.agentName, 'lexvoice-browser-agent'); + } finally { + globalThis.fetch = originalFetch; + restoreEnv(previousEnv); + } }); test('frontend exposes the server-owned voice session id to dispatch callers', async () => { From 381382bf283b5b4fbdf33492ee647646edade766 Mon Sep 17 00:00:00 2001 From: guowei Date: Mon, 3 Aug 2026 17:13:51 +0800 Subject: [PATCH 4/4] fix: roll back failed media controller bind --- lib/media-gate-executor.ts | 13 +++++++- tests/livekit-media-gate.test.mjs | 54 +++++++++++++++++++++++++++++++ 2 files changed, 66 insertions(+), 1 deletion(-) diff --git a/lib/media-gate-executor.ts b/lib/media-gate-executor.ts index a8a7570c2..6c60cf456 100644 --- a/lib/media-gate-executor.ts +++ b/lib/media-gate-executor.ts @@ -125,7 +125,18 @@ export class MediaGateExecutor { this.controllerIdentity = controllerIdentity; this.closedBlocker = 'lease_missing'; this.device.close(); - return this.enqueue(() => this.publishSnapshot(controllerIdentity, null, null)); + return this.enqueue(async () => { + try { + await this.publishSnapshot(controllerIdentity, null, null); + } catch (error) { + if (this.controllerIdentity === controllerIdentity) { + this.controllerIdentity = null; + this.resetCommandOrdering(); + this.invalidateAndClose('controller_disconnected', true); + } + throw error; + } + }); } handleCommand(controllerIdentity: string, command: MediaControlCommand): Promise { diff --git a/tests/livekit-media-gate.test.mjs b/tests/livekit-media-gate.test.mjs index e11cfb1b8..9b83de0ec 100644 --- a/tests/livekit-media-gate.test.mjs +++ b/tests/livekit-media-gate.test.mjs @@ -4,6 +4,7 @@ import { test } from 'node:test'; const { RoomEvent } = await import('livekit-client'); const { LiveKitMediaGateAdapter } = await import('../lib/livekit-media-gate.ts'); +const { MediaGateExecutor } = await import('../lib/media-gate-executor.ts'); const { MEDIA_CONTROL_TOPIC, MEDIA_STATE_TOPIC } = await import('../lib/media-control-protocol.ts'); const textEncoder = new TextEncoder(); @@ -426,6 +427,59 @@ test('controller disconnect closes through executor, clears the pin, and permits ]); }); +test('a failed late controller bind rolls back and permits a replacement controller', async () => { + const room = new FakeRoom(); + const errors = []; + const adapter = new LiveKitMediaGateAdapter({ + room, + agentName: 'frontdesk-browser-agent', + onError: (error) => errors.push(error), + }); + const executor = new MediaGateExecutor({ + targetIdentity: 'browser-edge', + device: { + close() {}, + async open() {}, + snapshot() { + return { + captureActive: true, + trackPublished: true, + trackMuted: true, + }; + }, + }, + publishState: adapter.publishState, + uuid: () => 'state-epoch-1', + nowUnixMs: () => 1_000, + nowMonotonicMs: () => 1_000, + scheduler: { + setTimeout: () => 1, + clearTimeout() {}, + }, + maxOpenLeaseMs: 1_000, + }); + await adapter.start(executor); + + const failedPublish = deferred(); + room.publishGate = failedPublish; + const first = namedAgent('agent-first'); + room.connectParticipant(first); + failedPublish.reject(new Error('initial state publish failed')); + await adapter.drain(); + + room.publishGate = null; + room.disconnectParticipant(first); + const second = namedAgent('agent-second'); + room.connectParticipant(second); + await adapter.drain(); + + assert.match(errors[0]?.message ?? '', /initial state publish failed/); + assert.deepEqual( + room.published.map(({ options }) => options.destinationIdentities), + [[first.identity], [second.identity]] + ); +}); + test('controller disconnect is not blocked behind stuck command work', async () => { const commandGate = deferred(); const controller = namedAgent();