feat(synthesia): add interactive avatar plugin

.changeset/fuzzy-avatars-swap.md

@@ -0,0 +1,8 @@
+---
+'@livekit/agents-plugin-synthesia': minor
+'@livekit/agents': patch
+---
+
+Add Synthesia interactive avatar sessions with precomputed mid-session avatar swaps.
+
+Add audio output tail replacement and data-stream output cleanup for avatar session lifecycle.

agents/etc/agents.api.md

@@ -3009,6 +3009,7 @@ function createWarmTransferTask(input?: WarmTransferTaskOptions): AgentTask<Warm
 // @public
 export class DataStreamAudioOutput extends AudioOutput {
     constructor(opts: DataStreamAudioOutputOptions);
+    aclose(): Promise<void>;
     // (undocumented)
     captureFrame(frame: AudioFrame): Promise<void>;
     // (undocumented)

agents/src/voice/avatar/datastream_io.ts

@@ -60,6 +60,9 @@ export class DataStreamAudioOutput extends AudioOutput {
   private lock = new Mutex();
   private startTask?: Task<void>;
   private firstFrameEmitted: boolean = false;
+  private closed: boolean = false;
+  private playbackFinishedHandler = (data: RpcInvocationData) => this.handlePlaybackFinished(data);
+  private playbackStartedHandler = (data: RpcInvocationData) => this.handlePlaybackStarted(data);
 
   #logger = log();
 
@@ -73,23 +76,20 @@ export class DataStreamAudioOutput extends AudioOutput {
     this.waitRemoteTrack = waitRemoteTrack;
     this.waitPlaybackStart = waitPlaybackStart ?? false;
 
-    const onRoomConnected = async () => {
-      if (this.startTask) return;
-
-      await this.roomConnectedFuture.await;
-
+    const onRoomConnected = () => {
+      if (this.startTask || !this.room.isConnected || this.closed) return;
       // register the rpc method right after the room is connected
       DataStreamAudioOutput.registerPlaybackFinishedRpc({
         room,
         callerIdentity: this.destinationIdentity,
-        handler: (data) => this.handlePlaybackFinished(data),
+        handler: this.playbackFinishedHandler,
       });
 
       if (this.waitPlaybackStart) {
         DataStreamAudioOutput.registerPlaybackStartedRpc({
           room,
           callerIdentity: this.destinationIdentity,
-          handler: (data) => this.handlePlaybackStarted(data),
+          handler: this.playbackStartedHandler,
         });
       }
 
@@ -98,11 +98,13 @@ export class DataStreamAudioOutput extends AudioOutput {
 
     this.roomConnectedFuture = new Future<void>();
 
-    this.room.on(RoomEvent.ConnectionStateChanged, (_) => {
+    this.onRoomConnectionStateChanged = () => {
       if (room.isConnected && !this.roomConnectedFuture.done) {
         this.roomConnectedFuture.resolve(undefined);
       }
-    });
+      onRoomConnected();
+    };
+    this.room.on(RoomEvent.ConnectionStateChanged, this.onRoomConnectionStateChanged);
 
     if (this.room.isConnected) {
       this.roomConnectedFuture.resolve(undefined);
@@ -111,7 +113,9 @@ export class DataStreamAudioOutput extends AudioOutput {
     onRoomConnected();
   }
 
-  private async _start(_abortSignal: AbortSignal) {
+  private onRoomConnectionStateChanged: () => void;
+
+  private async _start(abortSignal: AbortSignal) {
     const unlock = await this.lock.lock();
 
     try {
@@ -129,6 +133,7 @@ export class DataStreamAudioOutput extends AudioOutput {
       await waitForParticipant({
         room: this.room,
         identity: this.destinationIdentity,
+        signal: abortSignal,
       });
 
       if (this.waitRemoteTrack) {
@@ -144,6 +149,7 @@ export class DataStreamAudioOutput extends AudioOutput {
           room: this.room,
           identity: this.destinationIdentity,
           kind: this.waitRemoteTrack,
+          signal: abortSignal,
         });
       }
 
@@ -161,6 +167,7 @@ export class DataStreamAudioOutput extends AudioOutput {
   }
 
   async captureFrame(frame: AudioFrame): Promise<void> {
+    if (this.closed) throw new Error('DataStreamAudioOutput is closed');
     if (!this.startTask) {
       this.startTask = Task.from(({ signal }) => this._start(signal));
     }
@@ -219,6 +226,30 @@ export class DataStreamAudioOutput extends AudioOutput {
     });
   }
 
+  /** Release resources owned by this data-stream output. */
+  async aclose(): Promise<void> {
+    if (this.closed) return;
+    this.closed = true;
+    this.room.off(RoomEvent.ConnectionStateChanged, this.onRoomConnectionStateChanged);
+    this.startTask?.cancel();
+    if (this.streamWriter) {
+      await this.streamWriter.close();
+      this.streamWriter = undefined;
+    }
+    if (
+      DataStreamAudioOutput._playbackFinishedHandlers[this.destinationIdentity] ===
+      this.playbackFinishedHandler
+    ) {
+      delete DataStreamAudioOutput._playbackFinishedHandlers[this.destinationIdentity];
+    }
+    if (
+      DataStreamAudioOutput._playbackStartedHandlers[this.destinationIdentity] ===
+      this.playbackStartedHandler
+    ) {
+      delete DataStreamAudioOutput._playbackStartedHandlers[this.destinationIdentity];
+    }
+  }
+
   private handlePlaybackFinished(data: RpcInvocationData): string {
     if (data.callerIdentity !== this.destinationIdentity) {
       this.#logger.warn(

agents/src/voice/io.test.ts

@@ -1,8 +1,9 @@
 // SPDX-FileCopyrightText: 2026 LiveKit, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
+import type { AudioFrame } from '@livekit/rtc-node';
 import { describe, expect, it, vi } from 'vitest';
-import { AgentInput, AudioInput } from './io.js';
+import { AgentInput, AgentOutput, AudioInput, AudioOutput } from './io.js';
 
 class TestAudioInput extends AudioInput {
   override setAttached = vi.fn();
@@ -59,3 +60,66 @@ describe('AgentInput', () => {
     expect(order).toEqual(['setAttached:true', 'onAttached', 'setAttached:false', 'onDetached']);
   });
 });
+
+class TestAudioOutput extends AudioOutput {
+  captureFrame = vi.fn(async (frame: AudioFrame) => super.captureFrame(frame));
+  override flush = vi.fn(() => super.flush());
+  override clearBuffer = vi.fn();
+  override onAttached = vi.fn();
+  override onDetached = vi.fn();
+}
+
+class TestAudioWrapper extends AudioOutput {
+  constructor(next: AudioOutput) {
+    super(next.sampleRate, next, { pause: true });
+  }
+
+  override async captureFrame(frame: AudioFrame): Promise<void> {
+    await super.captureFrame(frame);
+    await this.nextInChain!.captureFrame(frame);
+  }
+
+  override flush(): void {
+    super.flush();
+    this.nextInChain!.flush();
+  }
+
+  override clearBuffer(): void {
+    this.nextInChain!.clearBuffer();
+  }
+}
+
+describe('AgentOutput.replaceAudioTail', () => {
+  it('replaces a bare output directly', () => {
+    const output = new AgentOutput(() => {});
+    const original = new TestAudioOutput();
+    const replacement = new TestAudioOutput();
+    output.audio = original;
+
+    output.replaceAudioTail(replacement);
+
+    expect(output.audio).toBe(replacement);
+    expect(original.onDetached).toHaveBeenCalledOnce();
+    expect(replacement.onAttached).toHaveBeenCalledOnce();
+  });
+
+  it('keeps wrappers and settles a flushed segment when swapping the leaf', async () => {
+    const output = new AgentOutput(() => {});
+    const original = new TestAudioOutput();
+    const replacement = new TestAudioOutput();
+    const wrapper = new TestAudioWrapper(original);
+    output.audio = wrapper;
+    const frame = { samplesPerChannel: 480, sampleRate: 24000 } as AudioFrame;
+    await wrapper.captureFrame(frame);
+    wrapper.flush();
+
+    output.replaceAudioTail(replacement);
+
+    expect(output.audio).toBe(wrapper);
+    expect(original.flush).toHaveBeenCalledOnce();
+    expect(original.clearBuffer).toHaveBeenCalledOnce();
+    await expect(wrapper.waitForPlayout()).resolves.toMatchObject({ interrupted: true });
+    await wrapper.captureFrame(frame);
+    expect(replacement.captureFrame).toHaveBeenCalledWith(frame);
+  });
+});

agents/src/voice/io.ts

@@ -136,15 +136,16 @@ export abstract class AudioOutput extends EventEmitter {
   ) {
     super();
     this.capabilities = capabilities;
+    this.attachNextInChainListeners();
+  }
 
-    if (this.nextInChain) {
-      this.nextInChain.on(AudioOutput.EVENT_PLAYBACK_STARTED, (ev: PlaybackStartedEvent) =>
-        this.onPlaybackStarted(ev.createdAt),
-      );
-      this.nextInChain.on(AudioOutput.EVENT_PLAYBACK_FINISHED, (ev: PlaybackFinishedEvent) =>
-        this.onPlaybackFinished(ev),
-      );
-    }
+  private onNextPlaybackStarted = (ev: PlaybackStartedEvent) =>
+    this.onPlaybackStarted(ev.createdAt);
+  private onNextPlaybackFinished = (ev: PlaybackFinishedEvent) => this.onPlaybackFinished(ev);
+
+  private attachNextInChainListeners(): void {
+    this.nextInChain?.on(AudioOutput.EVENT_PLAYBACK_STARTED, this.onNextPlaybackStarted);
+    this.nextInChain?.on(AudioOutput.EVENT_PLAYBACK_FINISHED, this.onNextPlaybackFinished);
   }
 
   /**
@@ -451,6 +452,42 @@ export class AgentOutput {
     }
   }
 
+  /**
+   * Replace the leaf audio sink while retaining recorder and transcription wrappers.
+   * Falls back to replacing the whole output when no wrapper chain is installed.
+   */
+  replaceAudioTail(sink: AudioOutput): void {
+    let current = this._audioSink;
+    while (current) {
+      const internals = current as unknown as {
+        nextInChain?: AudioOutput;
+        onNextPlaybackStarted: (event: PlaybackStartedEvent) => void;
+        onNextPlaybackFinished: (event: PlaybackFinishedEvent) => void;
+        attachNextInChainListeners: () => void;
+        _capturing: boolean;
+      };
+      const next = internals.nextInChain;
+      if (next && !(next as unknown as { nextInChain?: AudioOutput }).nextInChain) {
+        next.off(AudioOutput.EVENT_PLAYBACK_STARTED, internals.onNextPlaybackStarted);
+        next.off(AudioOutput.EVENT_PLAYBACK_FINISHED, internals.onNextPlaybackFinished);
+        if (current.pendingPlayoutSegments > 0) {
+          if (internals._capturing) next.flush();
+          next.clearBuffer();
+        }
+        if (this._audioEnabled) next.onDetached();
+        internals.nextInChain = sink;
+        internals.attachNextInChainListeners();
+        if (this._audioEnabled) sink.onAttached();
+        if (current.pendingPlayoutSegments > 0) {
+          current.onPlaybackFinished({ playbackPosition: 0, interrupted: true });
+        }
+        return;
+      }
+      current = next ?? null;
+    }
+    this.audio = sink;
+  }
+
   get transcription(): TextOutput | null {
     return this._transcriptionSink;
   }

agents/src/voice/transcription/synchronizer.ts

@@ -760,7 +760,7 @@ class SyncedAudioOutput extends AudioOutput {
 
   constructor(
     public synchronizer: TranscriptionSynchronizer,
-    private nextInChainAudio: AudioOutput,
+    nextInChainAudio: AudioOutput,
   ) {
     super(nextInChainAudio.sampleRate, nextInChainAudio, { pause: true });
   }
@@ -799,10 +799,10 @@ class SyncedAudioOutput extends AudioOutput {
       this.segmentOpen = true;
       this.segmentAccepted = false;
     }
-    const downstreamCapturedBefore = this.nextInChainAudio.capturedPlayoutSegments;
+    const downstreamCapturedBefore = this.nextInChain!.capturedPlayoutSegments;
     await super.captureFrame(frame);
-    await this.nextInChainAudio.captureFrame(frame); // passthrough audio
-    if (this.nextInChainAudio.capturedPlayoutSegments > downstreamCapturedBefore) {
+    await this.nextInChain!.captureFrame(frame); // passthrough audio
+    if (this.nextInChain!.capturedPlayoutSegments > downstreamCapturedBefore) {
       this.segmentAccepted = true;
     }
 
@@ -841,7 +841,7 @@ class SyncedAudioOutput extends AudioOutput {
 
   flush() {
     super.flush();
-    this.nextInChainAudio.flush();
+    this.nextInChain!.flush();
     if (this.segmentOpen) {
       this.segmentOpen = false;
       this.lastSegmentAccepted = this.segmentAccepted;
@@ -877,11 +877,11 @@ class SyncedAudioOutput extends AudioOutput {
   }
 
   clearBuffer() {
-    this.nextInChainAudio.clearBuffer();
+    this.nextInChain!.clearBuffer();
   }
 
   async waitForPlayout(): Promise<PlaybackFinishedEvent> {
-    const drift = this.pendingPlayoutSegments - this.nextInChainAudio.pendingPlayoutSegments;
+    const drift = this.pendingPlayoutSegments - this.nextInChain!.pendingPlayoutSegments;
     for (let i = 0; i < drift; i++) {
       this.settleDriftFinish();
     }

plugins/synthesia/README.md

@@ -0,0 +1,46 @@
+# Synthesia plugin for LiveKit Agents
+
+Attach a [Synthesia](https://www.synthesia.io/) interactive avatar to a LiveKit voice agent. The avatar joins the room and lip-syncs the agent's speech in real time.
+
+See the [Synthesia integration docs](https://docs.livekit.io/agents/models/avatar/plugins/synthesia/) for more information.
+
+## Installation
+
+```bash
+npm install @livekit/agents-plugin-synthesia
+```
+
+## Pre-requisites
+
+You'll need an API key from Synthesia. It can be set as the `SYNTHESIA_API_KEY` environment variable.
+
+## Usage
+
+```typescript
+import { voice } from '@livekit/agents';
+import * as synthesia from '@livekit/agents-plugin-synthesia';
+
+const session = new voice.AgentSession(/* ... */);
+const avatar = new synthesia.AvatarSession(
+  new synthesia.AvatarConfig({
+    avatarIds: ['03cee7ec-ac90-45ec-8c20-74a399cf3dc4'],
+  }),
+);
+
+await avatar.start(session, ctx.room); // before session.start()
+await session.start({ agent: new Agent(/* ... */), room: ctx.room });
+```
+
+Set your Synthesia workspace API key in `SYNTHESIA_API_KEY`, or pass `apiKey`. `avatarIds` takes one to five gallery IDs available to your workspace. The first is active and the rest are precomputed so `swapAvatar()` can switch to them during the session. An inaccessible ID raises `SynthesiaError` with `type` set to `ErrorType.UNKNOWN_AVATAR`.
+
+```typescript
+await avatar.swapAvatar('<another-id-from-avatarIds>');
+await avatar.swapAvatar('default');
+```
+
+## Parameters
+
+| Parameter                   | Default                    | Description                                                                                     |
+| --------------------------- | -------------------------- | ----------------------------------------------------------------------------------------------- |
+| `avatarParticipantIdentity` | `"synthesia-avatar-agent"` | The LiveKit identity the avatar joins under. It must be unique per concurrent avatar in a room. |
+| `avatarParticipantName`     | `"Synthesia avatar"`       | The LiveKit display name the avatar joins under.                                                |

plugins/synthesia/api-extractor.json

@@ -0,0 +1,8 @@
+/**
+ * Config file for API Extractor. For more info, visit https://api-extractor.com.
+ */
+{
+  "$schema": "https://developer.microsoft.com/json-schemas/api-extractor/v7/api-extractor.schema.json",
+  "extends": "../../api-extractor-shared.json",
+  "mainEntryPointFilePath": "./dist/index.d.ts"
+}