diff --git a/.changeset/catch-avatar-clear-buffer-rpc.md b/.changeset/catch-avatar-clear-buffer-rpc.md new file mode 100644 index 000000000..c2d536fa5 --- /dev/null +++ b/.changeset/catch-avatar-clear-buffer-rpc.md @@ -0,0 +1,6 @@ +--- +'@livekit/agents': patch +--- + +Prevent rejected avatar clear-buffer RPCs from emitting unhandled rejections or stranding +playout waiters. diff --git a/agents/src/voice/avatar/datastream_io.test.ts b/agents/src/voice/avatar/datastream_io.test.ts new file mode 100644 index 000000000..bb5e3dd97 --- /dev/null +++ b/agents/src/voice/avatar/datastream_io.test.ts @@ -0,0 +1,148 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { AudioFrame, ByteStreamWriter, Room } from '@livekit/rtc-node'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import { AudioOutput, type PlaybackFinishedEvent } from '../io.js'; +import { DataStreamAudioOutput } from './datastream_io.js'; + +const { logger } = vi.hoisted(() => ({ + logger: { + debug: vi.fn(), + warn: vi.fn(), + }, +})); + +vi.mock('../../log.js', () => ({ + log: () => logger, +})); + +function createByteStreamWriter(): ByteStreamWriter { + return new ByteStreamWriter(new WritableStream(), { + streamId: 'test-stream', + mimeType: 'application/octet-stream', + topic: 'lk.audio_stream', + timestamp: 0, + name: 'audio', + }); +} + +function createRoom(performRpcImpl: () => Promise) { + const avatar = { identity: 'avatar' }; + const room = new Room(); + const performRpc = vi.fn(performRpcImpl); + + Object.defineProperties(room, { + isConnected: { value: true }, + localParticipant: { + value: { + performRpc, + registerRpcMethod: vi.fn(), + streamBytes: vi.fn(async () => createByteStreamWriter()), + }, + }, + remoteParticipants: { value: new Map([[avatar.identity, avatar]]) }, + }); + + return { room, performRpc }; +} + +function createFrame(): AudioFrame { + return new AudioFrame(new Int16Array(160), 8000, 1, 160); +} + +function nextTick(): Promise { + return new Promise((resolve) => setImmediate(resolve)); +} + +describe('DataStreamAudioOutput.clearBuffer', () => { + beforeEach(() => { + vi.clearAllMocks(); + DataStreamAudioOutput._playbackFinishedRpcRegistered = false; + DataStreamAudioOutput._playbackFinishedHandlers = {}; + }); + + it('settles pending playout when the clear-buffer RPC rejects', async () => { + const error = new Error('Failed to send'); + const { room, performRpc } = createRoom(() => Promise.reject(error)); + const output = new DataStreamAudioOutput({ + room, + destinationIdentity: 'avatar', + }); + + await output.captureFrame(createFrame()); + output.flush(); + + const playout = output.waitForPlayout(); + output.clearBuffer(); + + await expect(playout).resolves.toEqual({ + playbackPosition: 0, + interrupted: true, + }); + expect(performRpc).toHaveBeenCalledExactlyOnceWith({ + destinationIdentity: 'avatar', + method: 'lk.clear_buffer', + payload: '', + }); + expect(logger.warn).toHaveBeenCalledExactlyOnceWith( + { error, destinationIdentity: 'avatar' }, + 'failed to perform clear buffer rpc', + ); + }); + + it('does not let a late rejection settle a newer segment', async () => { + let rejectRpc: (reason?: unknown) => void = () => {}; + const rpc = new Promise((_resolve, reject) => { + rejectRpc = reject; + }); + const { room } = createRoom(() => rpc); + const output = new DataStreamAudioOutput({ + room, + destinationIdentity: 'avatar', + }); + + await output.captureFrame(createFrame()); + output.flush(); + output.clearBuffer(); + + output.onPlaybackFinished({ playbackPosition: 0.01, interrupted: true }); + + await nextTick(); + await output.captureFrame(createFrame()); + output.flush(); + const secondPlayout = output.waitForPlayout(); + + rejectRpc(new Error('Failed to send')); + await vi.waitFor(() => expect(logger.warn).toHaveBeenCalledOnce()); + + const secondEvent = { playbackPosition: 0.02, interrupted: false }; + output.onPlaybackFinished(secondEvent); + await expect(secondPlayout).resolves.toEqual(secondEvent); + }); + + it('settles every segment pending when clearBuffer was called', async () => { + const { room } = createRoom(() => Promise.reject(new Error('Failed to send'))); + const output = new DataStreamAudioOutput({ + room, + destinationIdentity: 'avatar', + }); + const playbackEvents: PlaybackFinishedEvent[] = []; + output.on(AudioOutput.EVENT_PLAYBACK_FINISHED, (event) => playbackEvents.push(event)); + + await output.captureFrame(createFrame()); + output.flush(); + await nextTick(); + await output.captureFrame(createFrame()); + output.flush(); + + const playout = output.waitForPlayout(); + output.clearBuffer(); + await expect(playout).resolves.toEqual({ playbackPosition: 0, interrupted: true }); + + expect(playbackEvents).toEqual([ + { playbackPosition: 0, interrupted: true }, + { playbackPosition: 0, interrupted: true }, + ]); + }); +}); diff --git a/agents/src/voice/avatar/datastream_io.ts b/agents/src/voice/avatar/datastream_io.ts index a925474f6..f9c68be60 100644 --- a/agents/src/voice/avatar/datastream_io.ts +++ b/agents/src/voice/avatar/datastream_io.ts @@ -212,11 +212,26 @@ export class DataStreamAudioOutput extends AudioOutput { clearBuffer(): void { if (!this.started) return; - this.room.localParticipant!.performRpc({ - destinationIdentity: this.destinationIdentity, - method: RPC_CLEAR_BUFFER, - payload: '', - }); + // A delayed rejection must not finish audio captured after this call. + const playoutTarget = this.capturedPlayoutSegments; + + void this.room + .localParticipant!.performRpc({ + destinationIdentity: this.destinationIdentity, + method: RPC_CLEAR_BUFFER, + payload: '', + }) + .catch((error) => { + this.#logger.warn( + { error, destinationIdentity: this.destinationIdentity }, + 'failed to perform clear buffer rpc', + ); + + const finishedSegments = this.capturedPlayoutSegments - this.pendingPlayoutSegments; + for (let segment = finishedSegments; segment < playoutTarget; segment++) { + this.onPlaybackFinished({ playbackPosition: 0, interrupted: true }); + } + }); } private handlePlaybackFinished(data: RpcInvocationData): string {