fix(avatar): handle clear-buffer RPC failures

.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.

agents/src/voice/avatar/datastream_io.test.ts

@@ -0,0 +1,166 @@
+// 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<Uint8Array>(), {
+    streamId: 'test-stream',
+    mimeType: 'application/octet-stream',
+    topic: 'lk.audio_stream',
+    timestamp: 0,
+    name: 'audio',
+  });
+}
+
+function createRoom(performRpcImpl: () => Promise<string>) {
+  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,
+        registerRpcMethod: vi.fn(),
+        streamBytes,
+      },
+    },
+    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<void> {
+  return new Promise((resolve) => setImmediate(resolve));
+}
+
+describe('DataStreamAudioOutput.clearBuffer', () => {
+  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<string>((_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();
+
+    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);
+  });
+});

agents/src/voice/avatar/datastream_io.ts

@@ -212,11 +212,33 @@ export class DataStreamAudioOutput extends AudioOutput {
   clearBuffer(): void {
     if (!this.started) return;
 
-    this.room.localParticipant!.performRpc({
-      destinationIdentity: this.destinationIdentity,
-      method: RPC_CLEAR_BUFFER,
-      payload: '',
-    });
+    // 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,
+        method: RPC_CLEAR_BUFFER,
+        payload: '',
+      })
+      .catch((error) => {
+        this.#logger.warn(
+          { 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 {