From 518cc6a3cad7285fc0cfcd0f13bfe85b69d630b3 Mon Sep 17 00:00:00 2001 From: Sora Morimoto Date: Thu, 16 Jul 2026 17:38:46 +0900 Subject: [PATCH 1/4] fix(avatar): handle clear buffer RPC failures --- .changeset/catch-avatar-clear-buffer-rpc.md | 5 ++ agents/src/voice/avatar/datastream_io.test.ts | 48 +++++++++++++++++++ agents/src/voice/avatar/datastream_io.ts | 17 +++++-- 3 files changed, 65 insertions(+), 5 deletions(-) create mode 100644 .changeset/catch-avatar-clear-buffer-rpc.md create mode 100644 agents/src/voice/avatar/datastream_io.test.ts diff --git a/.changeset/catch-avatar-clear-buffer-rpc.md b/.changeset/catch-avatar-clear-buffer-rpc.md new file mode 100644 index 000000000..1459f808a --- /dev/null +++ b/.changeset/catch-avatar-clear-buffer-rpc.md @@ -0,0 +1,5 @@ +--- +'@livekit/agents': patch +--- + +Handle rejected avatar clear-buffer RPCs without emitting unhandled promise rejections. 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..167ddefd6 --- /dev/null +++ b/agents/src/voice/avatar/datastream_io.test.ts @@ -0,0 +1,48 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import type { Room } from '@livekit/rtc-node'; +import { describe, expect, it, vi } from 'vitest'; +import { DataStreamAudioOutput } from './datastream_io.js'; + +const { logger } = vi.hoisted(() => ({ + logger: { + debug: vi.fn(), + warn: vi.fn(), + }, +})); + +vi.mock('../../log.js', () => ({ + log: () => logger, +})); + +function createRoom(performRpc: () => Promise) { + const avatar = { identity: 'avatar' }; + + return { + isConnected: true, + localParticipant: { + performRpc: vi.fn(performRpc), + registerRpcMethod: vi.fn(), + }, + remoteParticipants: new Map([[avatar.identity, avatar]]), + on: vi.fn(), + off: vi.fn(), + }; +} + +describe('DataStreamAudioOutput.clearBuffer', () => { + it('handles a rejected clear-buffer RPC', async () => { + const room = createRoom(() => Promise.reject(new Error('Failed to send'))); + const output = new DataStreamAudioOutput({ + room: room as unknown as Room, + destinationIdentity: 'avatar', + }); + + await vi.waitFor(() => { + output.clearBuffer(); + expect(room.localParticipant.performRpc).toHaveBeenCalledOnce(); + }); + await vi.waitFor(() => expect(logger.warn).toHaveBeenCalledOnce()); + }); +}); diff --git a/agents/src/voice/avatar/datastream_io.ts b/agents/src/voice/avatar/datastream_io.ts index a925474f6..2a6e79612 100644 --- a/agents/src/voice/avatar/datastream_io.ts +++ b/agents/src/voice/avatar/datastream_io.ts @@ -212,11 +212,18 @@ export class DataStreamAudioOutput extends AudioOutput { clearBuffer(): void { if (!this.started) return; - this.room.localParticipant!.performRpc({ - destinationIdentity: this.destinationIdentity, - method: RPC_CLEAR_BUFFER, - payload: '', - }); + 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', + ); + }); } private handlePlaybackFinished(data: RpcInvocationData): string { From 804786c69119d922d8de774a88234cd445abe934 Mon Sep 17 00:00:00 2001 From: Sora Morimoto Date: Fri, 17 Jul 2026 11:38:34 +0900 Subject: [PATCH 2/4] test(avatar): use typed Room fixture for RPC failure behaviour --- agents/src/voice/avatar/datastream_io.test.ts | 25 +++++++++++-------- 1 file changed, 14 insertions(+), 11 deletions(-) diff --git a/agents/src/voice/avatar/datastream_io.test.ts b/agents/src/voice/avatar/datastream_io.test.ts index 167ddefd6..ae2f40681 100644 --- a/agents/src/voice/avatar/datastream_io.test.ts +++ b/agents/src/voice/avatar/datastream_io.test.ts @@ -1,7 +1,7 @@ // SPDX-FileCopyrightText: 2026 LiveKit, Inc. // // SPDX-License-Identifier: Apache-2.0 -import type { Room } from '@livekit/rtc-node'; +import { Room } from '@livekit/rtc-node'; import { describe, expect, it, vi } from 'vitest'; import { DataStreamAudioOutput } from './datastream_io.js'; @@ -16,26 +16,29 @@ vi.mock('../../log.js', () => ({ log: () => logger, })); -function createRoom(performRpc: () => Promise) { +function createRoom(performRpc: () => Promise): Room { const avatar = { identity: 'avatar' }; + const room = new Room(); - return { - isConnected: true, + Object.defineProperties(room, { + isConnected: { value: true }, localParticipant: { - performRpc: vi.fn(performRpc), - registerRpcMethod: vi.fn(), + value: { + performRpc: vi.fn(performRpc), + registerRpcMethod: vi.fn(), + }, }, - remoteParticipants: new Map([[avatar.identity, avatar]]), - on: vi.fn(), - off: vi.fn(), - }; + remoteParticipants: { value: new Map([[avatar.identity, avatar]]) }, + }); + + return room; } describe('DataStreamAudioOutput.clearBuffer', () => { it('handles a rejected clear-buffer RPC', async () => { const room = createRoom(() => Promise.reject(new Error('Failed to send'))); const output = new DataStreamAudioOutput({ - room: room as unknown as Room, + room, destinationIdentity: 'avatar', }); From 9c913adfea946edcdaa32da50ce5c99d5d80eb26 Mon Sep 17 00:00:00 2001 From: Sora Morimoto Date: Fri, 17 Jul 2026 17:43:34 +0900 Subject: [PATCH 3/4] fix(avatar): settle playout after clear-buffer RPC failure Preserve the clear call's segment boundary so delayed failures do not finish newer audio. Cover the failure behaviour with typed Room tests. --- .changeset/catch-avatar-clear-buffer-rpc.md | 3 +- agents/src/voice/avatar/datastream_io.test.ts | 135 ++++++++++++++++-- agents/src/voice/avatar/datastream_io.ts | 15 ++ 3 files changed, 142 insertions(+), 11 deletions(-) diff --git a/.changeset/catch-avatar-clear-buffer-rpc.md b/.changeset/catch-avatar-clear-buffer-rpc.md index 1459f808a..c2d536fa5 100644 --- a/.changeset/catch-avatar-clear-buffer-rpc.md +++ b/.changeset/catch-avatar-clear-buffer-rpc.md @@ -2,4 +2,5 @@ '@livekit/agents': patch --- -Handle rejected avatar clear-buffer RPCs without emitting unhandled promise rejections. +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 index ae2f40681..89e58bbc9 100644 --- a/agents/src/voice/avatar/datastream_io.test.ts +++ b/agents/src/voice/avatar/datastream_io.test.ts @@ -1,8 +1,9 @@ // SPDX-FileCopyrightText: 2026 LiveKit, Inc. // // SPDX-License-Identifier: Apache-2.0 -import { Room } from '@livekit/rtc-node'; -import { describe, expect, it, vi } from 'vitest'; +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(() => ({ @@ -16,36 +17,150 @@ vi.mock('../../log.js', () => ({ log: () => logger, })); -function createRoom(performRpc: () => Promise): Room { +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); + const streamBytes = vi.fn(async () => createByteStreamWriter()); Object.defineProperties(room, { isConnected: { value: true }, localParticipant: { value: { - performRpc: vi.fn(performRpc), + performRpc, registerRpcMethod: vi.fn(), + streamBytes, }, }, remoteParticipants: { value: new Map([[avatar.identity, avatar]]) }, }); - return room; + 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', () => { - it('handles a rejected clear-buffer RPC', async () => { - const room = createRoom(() => Promise.reject(new Error('Failed to send'))); + beforeEach(() => { + vi.clearAllMocks(); + DataStreamAudioOutput._playbackFinishedRpcRegistered = false; + DataStreamAudioOutput._playbackFinishedHandlers = {}; + DataStreamAudioOutput._playbackStartedRpcRegistered = false; + DataStreamAudioOutput._playbackStartedHandlers = {}; + }); + + 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', + ); + expect(output.pendingPlayoutSegments).toBe(0); + }); + + 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', }); + const playbackEvents: PlaybackFinishedEvent[] = []; + output.on(AudioOutput.EVENT_PLAYBACK_FINISHED, (event) => playbackEvents.push(event)); + + await output.captureFrame(createFrame()); + output.flush(); + const firstPlayout = output.waitForPlayout(); + output.clearBuffer(); - await vi.waitFor(() => { - output.clearBuffer(); - expect(room.localParticipant.performRpc).toHaveBeenCalledOnce(); + const firstEvent = { playbackPosition: 0.01, interrupted: true }; + output.onPlaybackFinished(firstEvent); + await expect(firstPlayout).resolves.toEqual(firstEvent); + + await nextTick(); + await output.captureFrame(createFrame()); + output.flush(); + const secondPlayout = output.waitForPlayout(); + let secondSettled = false; + void secondPlayout.then(() => { + secondSettled = true; }); + + rejectRpc(new Error('Failed to send')); await vi.waitFor(() => expect(logger.warn).toHaveBeenCalledOnce()); + + expect(secondSettled).toBe(false); + expect(output.pendingPlayoutSegments).toBe(1); + expect(playbackEvents).toEqual([firstEvent]); + + 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 }, + ]); + expect(output.pendingPlayoutSegments).toBe(0); }); }); diff --git a/agents/src/voice/avatar/datastream_io.ts b/agents/src/voice/avatar/datastream_io.ts index 2a6e79612..342f60875 100644 --- a/agents/src/voice/avatar/datastream_io.ts +++ b/agents/src/voice/avatar/datastream_io.ts @@ -212,6 +212,11 @@ export class DataStreamAudioOutput extends AudioOutput { clearBuffer(): void { if (!this.started) return; + // Capture the current segment high-water mark before starting the RPC. The rejection may + // arrive after a real playback-finished event or after a new segment has started, so draining + // the live pending count in the catch handler could incorrectly finish newer audio. + const playoutTarget = this.capturedPlayoutSegments; + void this.room .localParticipant!.performRpc({ destinationIdentity: this.destinationIdentity, @@ -223,9 +228,19 @@ export class DataStreamAudioOutput extends AudioOutput { { error, destinationIdentity: this.destinationIdentity }, 'failed to perform clear buffer rpc', ); + + this.settlePlayoutThrough(playoutTarget); }); } + private settlePlayoutThrough(playoutTarget: number): void { + const finishedSegments = this.capturedPlayoutSegments - this.pendingPlayoutSegments; + + for (let segment = finishedSegments; segment < playoutTarget; segment++) { + this.onPlaybackFinished({ playbackPosition: 0, interrupted: true }); + } + } + private handlePlaybackFinished(data: RpcInvocationData): string { if (data.callerIdentity !== this.destinationIdentity) { this.#logger.warn( From 9d162b4a99cec3bbe3feba6ff466c2c712d1aef2 Mon Sep 17 00:00:00 2001 From: Sora Morimoto Date: Fri, 17 Jul 2026 17:52:55 +0900 Subject: [PATCH 4/4] refactor(avatar): simplify clear-buffer failure handling Keep the three distinct regression behaviours while removing duplicate assertions and a single-use helper. --- agents/src/voice/avatar/datastream_io.test.ts | 22 ++----------------- agents/src/voice/avatar/datastream_io.ts | 17 +++++--------- 2 files changed, 7 insertions(+), 32 deletions(-) diff --git a/agents/src/voice/avatar/datastream_io.test.ts b/agents/src/voice/avatar/datastream_io.test.ts index 89e58bbc9..bb5e3dd97 100644 --- a/agents/src/voice/avatar/datastream_io.test.ts +++ b/agents/src/voice/avatar/datastream_io.test.ts @@ -31,7 +31,6 @@ function createRoom(performRpcImpl: () => Promise) { const avatar = { identity: 'avatar' }; const room = new Room(); const performRpc = vi.fn(performRpcImpl); - const streamBytes = vi.fn(async () => createByteStreamWriter()); Object.defineProperties(room, { isConnected: { value: true }, @@ -39,7 +38,7 @@ function createRoom(performRpcImpl: () => Promise) { value: { performRpc, registerRpcMethod: vi.fn(), - streamBytes, + streamBytes: vi.fn(async () => createByteStreamWriter()), }, }, remoteParticipants: { value: new Map([[avatar.identity, avatar]]) }, @@ -61,8 +60,6 @@ describe('DataStreamAudioOutput.clearBuffer', () => { vi.clearAllMocks(); DataStreamAudioOutput._playbackFinishedRpcRegistered = false; DataStreamAudioOutput._playbackFinishedHandlers = {}; - DataStreamAudioOutput._playbackStartedRpcRegistered = false; - DataStreamAudioOutput._playbackStartedHandlers = {}; }); it('settles pending playout when the clear-buffer RPC rejects', async () => { @@ -92,7 +89,6 @@ describe('DataStreamAudioOutput.clearBuffer', () => { { error, destinationIdentity: 'avatar' }, 'failed to perform clear buffer rpc', ); - expect(output.pendingPlayoutSegments).toBe(0); }); it('does not let a late rejection settle a newer segment', async () => { @@ -105,34 +101,21 @@ describe('DataStreamAudioOutput.clearBuffer', () => { room, destinationIdentity: 'avatar', }); - const playbackEvents: PlaybackFinishedEvent[] = []; - output.on(AudioOutput.EVENT_PLAYBACK_FINISHED, (event) => playbackEvents.push(event)); await output.captureFrame(createFrame()); output.flush(); - const firstPlayout = output.waitForPlayout(); output.clearBuffer(); - const firstEvent = { playbackPosition: 0.01, interrupted: true }; - output.onPlaybackFinished(firstEvent); - await expect(firstPlayout).resolves.toEqual(firstEvent); + output.onPlaybackFinished({ playbackPosition: 0.01, interrupted: true }); await nextTick(); await output.captureFrame(createFrame()); output.flush(); const secondPlayout = output.waitForPlayout(); - let secondSettled = false; - void secondPlayout.then(() => { - secondSettled = true; - }); rejectRpc(new Error('Failed to send')); await vi.waitFor(() => expect(logger.warn).toHaveBeenCalledOnce()); - expect(secondSettled).toBe(false); - expect(output.pendingPlayoutSegments).toBe(1); - expect(playbackEvents).toEqual([firstEvent]); - const secondEvent = { playbackPosition: 0.02, interrupted: false }; output.onPlaybackFinished(secondEvent); await expect(secondPlayout).resolves.toEqual(secondEvent); @@ -161,6 +144,5 @@ describe('DataStreamAudioOutput.clearBuffer', () => { { playbackPosition: 0, interrupted: true }, { playbackPosition: 0, interrupted: true }, ]); - expect(output.pendingPlayoutSegments).toBe(0); }); }); diff --git a/agents/src/voice/avatar/datastream_io.ts b/agents/src/voice/avatar/datastream_io.ts index 342f60875..f9c68be60 100644 --- a/agents/src/voice/avatar/datastream_io.ts +++ b/agents/src/voice/avatar/datastream_io.ts @@ -212,9 +212,7 @@ export class DataStreamAudioOutput extends AudioOutput { clearBuffer(): void { if (!this.started) return; - // Capture the current segment high-water mark before starting the RPC. The rejection may - // arrive after a real playback-finished event or after a new segment has started, so draining - // the live pending count in the catch handler could incorrectly finish newer audio. + // A delayed rejection must not finish audio captured after this call. const playoutTarget = this.capturedPlayoutSegments; void this.room @@ -229,18 +227,13 @@ export class DataStreamAudioOutput extends AudioOutput { 'failed to perform clear buffer rpc', ); - this.settlePlayoutThrough(playoutTarget); + const finishedSegments = this.capturedPlayoutSegments - this.pendingPlayoutSegments; + for (let segment = finishedSegments; segment < playoutTarget; segment++) { + this.onPlaybackFinished({ playbackPosition: 0, interrupted: true }); + } }); } - private settlePlayoutThrough(playoutTarget: number): void { - const finishedSegments = this.capturedPlayoutSegments - this.pendingPlayoutSegments; - - for (let segment = finishedSegments; segment < playoutTarget; segment++) { - this.onPlaybackFinished({ playbackPosition: 0, interrupted: true }); - } - } - private handlePlaybackFinished(data: RpcInvocationData): string { if (data.callerIdentity !== this.destinationIdentity) { this.#logger.warn(