diff --git a/.changeset/fuzzy-avatars-swap.md b/.changeset/fuzzy-avatars-swap.md new file mode 100644 index 000000000..87b3446b1 --- /dev/null +++ b/.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. diff --git a/agents/etc/agents.api.md b/agents/etc/agents.api.md index 9036ea036..b89817eb0 100644 --- a/agents/etc/agents.api.md +++ b/agents/etc/agents.api.md @@ -3009,6 +3009,7 @@ function createWarmTransferTask(input?: WarmTransferTaskOptions): AgentTask; // (undocumented) captureFrame(frame: AudioFrame): Promise; // (undocumented) diff --git a/agents/src/voice/avatar/datastream_io.ts b/agents/src/voice/avatar/datastream_io.ts index e500998cc..67257e08c 100644 --- a/agents/src/voice/avatar/datastream_io.ts +++ b/agents/src/voice/avatar/datastream_io.ts @@ -60,6 +60,9 @@ export class DataStreamAudioOutput extends AudioOutput { private lock = new Mutex(); private startTask?: Task; 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(); - 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 { + 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 { + 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( diff --git a/agents/src/voice/io.test.ts b/agents/src/voice/io.test.ts index da311820f..067dd3d4d 100644 --- a/agents/src/voice/io.test.ts +++ b/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 { + 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); + }); +}); diff --git a/agents/src/voice/io.ts b/agents/src/voice/io.ts index b5c82d319..cf87fb2cf 100644 --- a/agents/src/voice/io.ts +++ b/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; } diff --git a/agents/src/voice/transcription/synchronizer.ts b/agents/src/voice/transcription/synchronizer.ts index 482787761..2fcbf4fb7 100644 --- a/agents/src/voice/transcription/synchronizer.ts +++ b/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 { - const drift = this.pendingPlayoutSegments - this.nextInChainAudio.pendingPlayoutSegments; + const drift = this.pendingPlayoutSegments - this.nextInChain!.pendingPlayoutSegments; for (let i = 0; i < drift; i++) { this.settleDriftFinish(); } diff --git a/plugins/synthesia/README.md b/plugins/synthesia/README.md new file mode 100644 index 000000000..7ee51696a --- /dev/null +++ b/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(''); +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. | diff --git a/plugins/synthesia/api-extractor.json b/plugins/synthesia/api-extractor.json new file mode 100644 index 000000000..8a589b91c --- /dev/null +++ b/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" +} diff --git a/plugins/synthesia/etc/agents-plugin-synthesia.api.md b/plugins/synthesia/etc/agents-plugin-synthesia.api.md new file mode 100644 index 000000000..8f31708c1 --- /dev/null +++ b/plugins/synthesia/etc/agents-plugin-synthesia.api.md @@ -0,0 +1,150 @@ +## API Report File for "@livekit/agents-plugin-synthesia" + +> Do not edit this file. It is a report generated by [API Extractor](https://api-extractor.com/). + +```ts + +import { AgentSession as AgentSession_2 } from '@livekit/protocol'; +import { AudioFrame } from '@livekit/rtc-node'; +import { AudioResampler } from '@livekit/rtc-node'; +import type { Context } from '@opentelemetry/api'; +import { EventEmitter } from 'events'; +import { EventEmitter as EventEmitter_2 } from 'node:events'; +import type { EventMap } from '@livekit/typed-emitter'; +import { FrameProcessor } from '@livekit/rtc-node'; +import { JsonObject } from '@bufbuild/protobuf'; +import type { JSONSchema7 } from 'json-schema'; +import { LocalTrackPublication } from '@livekit/rtc-node'; +import { Logger } from 'pino'; +import { NoiseCancellationOptions } from '@livekit/rtc-node'; +import { Participant } from '@livekit/rtc-node'; +import { ParticipantKind } from '@livekit/rtc-node'; +import { ReadableStream as ReadableStream_2 } from 'node:stream/web'; +import type { ReadableStreamDefaultReader as ReadableStreamDefaultReader_2 } from 'node:stream/web'; +import { RemoteParticipant } from '@livekit/rtc-node'; +import { RemoteTrackPublication } from '@livekit/rtc-node'; +import { Room } from '@livekit/rtc-node'; +import { RpcInvocationData } from '@livekit/rtc-node'; +import type { Span } from '@opentelemetry/api'; +import type { TextStreamInfo } from '@livekit/rtc-node'; +import { Throws } from '@livekit/throws-transformer/throws'; +import { ThrowsPromise } from '@livekit/throws-transformer/throws'; +import { TrackKind } from '@livekit/rtc-node'; +import { TrackPublishOptions } from '@livekit/rtc-node'; +import { TransformStream as TransformStream_2 } from 'node:stream/web'; +import type { TypedEventEmitter } from '@livekit/typed-emitter'; +import type { VideoFrame as VideoFrame_2 } from '@livekit/rtc-node'; +import type { WritableStreamDefaultWriter as WritableStreamDefaultWriter_2 } from 'node:stream/web'; +import { z } from 'zod'; + +// @public +export class AvatarConfig { + constructor(input: AvatarConfigOptions); + // (undocumented) + readonly avatarIds: readonly string[]; +} + +// @public (undocumented) +export interface AvatarConfigOptions { + avatarIds: readonly string[]; +} + +// Warning: (ae-forgotten-export) The symbol "voice" needs to be exported by the entry point index.d.ts +// +// @public +export class AvatarSession extends voice.AvatarSession { + constructor(avatarConfig: AvatarConfig, options?: AvatarSessionOptions); + // (undocumented) + aclose(): Promise; + // (undocumented) + get avatarIdentity(): string; + // (undocumented) + get provider(): string; + get sessionId(): string | null; + // (undocumented) + start(agentSession: voice.AgentSession, room: Room, options?: StartOptions): Promise; + swapAvatar(avatarId: string, input?: { + timeout?: number | undefined; + }): Promise; +} + +// @public +export interface AvatarSessionOptions { + apiKey?: string | null; + apiUrl?: string | null; + avatarParticipantIdentity?: string | null; + avatarParticipantName?: string | null; + joinTimeout?: number; +} + +// @public +export enum ErrorType { + // (undocumented) + AUTH = "auth", + // (undocumented) + CONCURRENCY_LIMIT = "concurrency_limit", + // (undocumented) + CONNECTION = "connection", + // (undocumented) + FEATURE_NOT_IN_PLAN = "feature_not_in_plan", + // (undocumented) + INVALID_ROOM_TOKEN = "invalid_room_token", + // (undocumented) + INVALID_SESSION_REQUEST = "invalid_session_request", + // (undocumented) + LIVEKIT_CREDENTIALS_REJECTED = "livekit_credentials_rejected", + // (undocumented) + QUOTA_EXCEEDED = "quota_exceeded", + // (undocumented) + RATE_LIMITED = "rate_limited", + // (undocumented) + TIMEOUT = "timeout", + // (undocumented) + UNKNOWN_AVATAR = "unknown_avatar" +} + +// @public +export interface StartOptions { + // (undocumented) + livekitApiKey?: string | null; + // (undocumented) + livekitApiSecret?: string | null; + // (undocumented) + livekitUrl?: string | null; +} + +// Warning: (ae-forgotten-export) The symbol "APIError" needs to be exported by the entry point index.d.ts +// +// @public +export class SynthesiaError extends APIError { + constructor(message: unknown, options?: SynthesiaErrorOptions); + // (undocumented) + readonly requestId: string | null; + // (undocumented) + readonly retryAfter: number | null; + // (undocumented) + readonly status: number | null; + // (undocumented) + readonly type: ErrorType | null; +} + +// @public (undocumented) +export interface SynthesiaErrorOptions { + // (undocumented) + body?: object | null; + // (undocumented) + cause?: unknown; + // (undocumented) + requestId?: string | null; + // (undocumented) + retryable?: boolean; + retryAfter?: number | null; + // (undocumented) + status?: number | null; + // (undocumented) + type?: ErrorType | null; +} + +// (No @packageDocumentation comment for this package) + +``` diff --git a/plugins/synthesia/package.json b/plugins/synthesia/package.json new file mode 100644 index 000000000..5ec7a13ac --- /dev/null +++ b/plugins/synthesia/package.json @@ -0,0 +1,51 @@ +{ + "name": "@livekit/agents-plugin-synthesia", + "version": "1.8.1", + "description": "Synthesia interactive avatar plugin for LiveKit Node Agents", + "main": "dist/index.js", + "require": "dist/index.cjs", + "types": "dist/index.d.ts", + "exports": { + "import": { + "types": "./dist/index.d.ts", + "default": "./dist/index.js" + }, + "require": { + "types": "./dist/index.d.cts", + "default": "./dist/index.cjs" + } + }, + "author": "LiveKit", + "type": "module", + "repository": "git@github.com:livekit/agents-js.git", + "license": "Apache-2.0", + "files": [ + "dist", + "src", + "README.md" + ], + "scripts": { + "build": "tsup --onSuccess \"pnpm build:types\"", + "build:types": "tsc --declaration --emitDeclarationOnly && node ../../scripts/copyDeclarationOutput.js", + "clean": "rm -rf dist", + "clean:build": "pnpm clean && pnpm build", + "lint": "eslint -f unix \"src/**/*.{ts,js}\"", + "api:check": "api-extractor run --typescript-compiler-folder ../../node_modules/typescript", + "api:update": "api-extractor run --local --typescript-compiler-folder ../../node_modules/typescript --verbose" + }, + "devDependencies": { + "@livekit/agents": "workspace:*", + "@livekit/rtc-node": "catalog:", + "@microsoft/api-extractor": "^7.58.12", + "pino": "^8.19.0", + "tsup": "^8.3.5", + "typescript": "^5.0.0" + }, + "dependencies": { + "livekit-server-sdk": "^2.13.3" + }, + "peerDependencies": { + "@livekit/agents": "workspace:*", + "@livekit/rtc-node": "catalog:" + } +} diff --git a/plugins/synthesia/src/api.test.ts b/plugins/synthesia/src/api.test.ts new file mode 100644 index 000000000..1ec07ce2a --- /dev/null +++ b/plugins/synthesia/src/api.test.ts @@ -0,0 +1,225 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { DEFAULT_API_CONNECT_OPTIONS } from '@livekit/agents'; +import { describe, expect, it, vi } from 'vitest'; +import { SESSION_PATH, SynthesiaAPI } from './api.js'; +import { ErrorType, SynthesiaError } from './errors.js'; + +const API_URL = 'https://api.example'; +const REQUEST = { + avatarIds: ['avatar-1'], + livekitUrl: 'wss://room.livekit.cloud', + livekitToken: 'lk-token-secret', +}; +const NO_SLEEP = { ...DEFAULT_API_CONNECT_OPTIONS, retryIntervalMs: 0 }; + +function response(status: number, body?: unknown, headers?: Record): Response { + return new Response(body === undefined ? '' : JSON.stringify(body), { status, headers }); +} + +function client(outcomes: Array, connOptions = NO_SLEEP) { + const fetch = vi.fn(async () => { + const outcome = outcomes.shift(); + if (outcome instanceof Error) throw outcome; + return outcome!; + }); + return { + api: new SynthesiaAPI({ apiKey: 'sk-secret-key', apiUrl: API_URL, fetch, connOptions }), + fetch, + }; +} + +function problem( + code: string, + status: number, + options: { detail?: unknown; withCode?: boolean; requestId?: string } = {}, +) { + return { + type: `https://developers.synthesia.io/errors/${code}`, + title: 'Problem title', + status, + ...(options.withCode === false ? {} : { code }), + ...(options.detail === undefined ? {} : { detail: options.detail }), + ...(options.requestId ? { requestId: options.requestId } : {}), + }; +} + +describe('SynthesiaAPI', () => { + it.each([{ id: 'ses_1' }, { session_id: 'ses_2' }])( + 'accepts published session id shapes', + async (body) => { + const { api } = client([response(201, body)]); + await expect(api.startSession(REQUEST)).resolves.toEqual({ + sessionId: body.id ?? body.session_id, + }); + }, + ); + + it('uses the published URL, authorization, and wire payload', async () => { + const { api, fetch } = client([response(201, { id: 'ses_1' })]); + await api.startSession({ ...REQUEST, avatarIds: ['ada-uuid', 'av_prefixed'] }); + expect(fetch).toHaveBeenCalledWith( + API_URL + SESSION_PATH, + expect.objectContaining({ + method: 'POST', + headers: { Authorization: 'sk-secret-key', 'Content-Type': 'application/json' }, + body: JSON.stringify({ + avatarIds: ['av_ada-uuid', 'av_prefixed'], + livekitUrl: REQUEST.livekitUrl, + livekitToken: REQUEST.livekitToken, + }), + }), + ); + }); + + it.each([ + ['validation_error', 400, ErrorType.INVALID_SESSION_REQUEST], + ['unknown_reference', 404, ErrorType.UNKNOWN_AVATAR], + ['quota_exceeded', 402, ErrorType.QUOTA_EXCEEDED], + ['rate_limited', 429, ErrorType.RATE_LIMITED], + ['unauthorized', 401, ErrorType.AUTH], + ['invalid_api_key', 401, ErrorType.AUTH], + ['unknown_avatar', 404, ErrorType.UNKNOWN_AVATAR], + ['avatar_not_accessible', 404, ErrorType.UNKNOWN_AVATAR], + ['invalid_token', 400, ErrorType.INVALID_ROOM_TOKEN], + ['invalid_livekit_credentials', 401, ErrorType.LIVEKIT_CREDENTIALS_REJECTED], + ['unauthenticated', 401, ErrorType.AUTH], + ['not_authorized', 403, ErrorType.AUTH], + ['forbidden', 403, ErrorType.AUTH], + ['insufficient_scope', 403, ErrorType.AUTH], + ['feature_not_in_plan', 403, ErrorType.FEATURE_NOT_IN_PLAN], + ['payment_required', 402, ErrorType.QUOTA_EXCEEDED], + ['concurrency_limit', 429, ErrorType.CONCURRENCY_LIMIT], + ['concurrency_limit_exceeded', 429, ErrorType.CONCURRENCY_LIMIT], + ] as const)('maps %s to %s', async (code, status, expected) => { + const { api } = client([response(status, { error: { code, message: 'nope' } })]); + await expect(api.startSession(REQUEST)).rejects.toMatchObject({ type: expected, status }); + }); + + it('uses problem detail, status, request id, and does not guess unknown problem codes', async () => { + const body = problem('new_backend_code', 404, { + detail: 'Detailed failure', + requestId: 'req_1', + }); + const { api } = client([response(404, body)]); + await expect(api.startSession(REQUEST)).rejects.toMatchObject({ + message: 'Detailed failure', + type: null, + body, + status: 404, + requestId: 'req_1', + }); + }); + + it('falls back from an unusable problem detail to title and from no code to status', async () => { + const { api } = client([ + response(404, problem('not_found', 404, { detail: [{ msg: 'x' }], withCode: false })), + ]); + await expect(api.startSession(REQUEST)).rejects.toMatchObject({ + message: 'Problem title', + type: ErrorType.UNKNOWN_AVATAR, + }); + }); + + it.each([ + [{ error: 'Forbidden', context: 'User is not authenticated' }, 'User is not authenticated'], + [{ error: 'Forbidden' }, 'Forbidden'], + [ + { + code: 'validation_error', + context: { livekit_url: ['Must be a wss:// URL'] }, + error: 'InvalidSessionRequestError', + }, + 'validation_error: {"livekit_url":["Must be a wss:// URL"]}', + ], + ])('renders legacy error bodies', async (body, message) => { + const { api } = client([response(403, body)]); + await expect(api.startSession(REQUEST)).rejects.toMatchObject({ message, body }); + }); + + it.each([ + [{ 'Retry-After': '12' }, { error: { code: 'rate_limited' } }, 12_000], + [{}, { error: { code: 'rate_limited' }, retry_after: 3.5 }, 3_500], + [{}, { error: { code: 'rate_limited', retry_after: 9 } }, 9_000], + ] as const)('reads retry-after in JS milliseconds', async (headers, body, expected) => { + const { api } = client([response(429, body, headers)]); + await expect(api.startSession(REQUEST)).rejects.toMatchObject({ retryAfter: expected }); + }); + + it('does not retry terminal errors', async () => { + const { api, fetch } = client([response(401, { error: { code: 'unauthorized' } })]); + await expect(api.startSession(REQUEST)).rejects.toBeInstanceOf(SynthesiaError); + expect(fetch).toHaveBeenCalledTimes(1); + }); + + it('retries 5xx and transport failures and can recover', async () => { + const { api, fetch } = client([ + new TypeError('network down'), + response(503), + response(200, { session_id: 'sess_ok' }), + ]); + await expect(api.startSession(REQUEST)).resolves.toEqual({ sessionId: 'sess_ok' }); + expect(fetch).toHaveBeenCalledTimes(3); + }); + + it('reports the final 5xx detail and request id when retries are exhausted', async () => { + const body = problem('service_unavailable', 503, { + detail: 'Retry later', + requestId: 'req_last', + }); + const { api } = client([response(503, body)], { ...NO_SLEEP, maxRetry: 0 }); + await expect(api.startSession(REQUEST)).rejects.toMatchObject({ + message: + 'could not start a Synthesia session; last attempt: HTTP 503: Retry later (request req_last)', + type: ErrorType.CONNECTION, + body, + status: 503, + requestId: 'req_last', + }); + }); + + it('retries when a response body fails while being read', async () => { + const truncated = { + status: 503, + ok: false, + text: vi.fn().mockRejectedValue(new Error('truncated body')), + headers: new Headers(), + } as unknown as Response; + const { api, fetch } = client([truncated, response(200, { id: 'sess_ok' })], { + ...NO_SLEEP, + maxRetry: 1, + }); + await expect(api.startSession(REQUEST)).resolves.toEqual({ sessionId: 'sess_ok' }); + expect(fetch).toHaveBeenCalledTimes(2); + }); + + it('lets the final mixed outcome decide the exhausted message', async () => { + const first = client([new TypeError('offline'), response(503, { message: 'unavailable' })], { + ...NO_SLEEP, + maxRetry: 1, + }); + await expect(first.api.startSession(REQUEST)).rejects.toMatchObject({ + message: + 'could not start a Synthesia session after 2 attempts; last attempt: HTTP 503: unavailable', + status: 503, + }); + + const second = client([response(503), new TypeError('offline')], { + ...NO_SLEEP, + maxRetry: 1, + }); + await expect(second.api.startSession(REQUEST)).rejects.toMatchObject({ + message: + 'could not start a Synthesia session after 2 attempts; last attempt: connection error', + status: null, + }); + }); + + it('rejects malformed successful responses', async () => { + const { api } = client([response(200, { unexpected: true })]); + await expect(api.startSession(REQUEST)).rejects.toThrow( + 'Synthesia response did not contain a session id', + ); + }); +}); diff --git a/plugins/synthesia/src/api.ts b/plugins/synthesia/src/api.ts new file mode 100644 index 000000000..6078b6311 --- /dev/null +++ b/plugins/synthesia/src/api.ts @@ -0,0 +1,227 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { type APIConnectOptions, DEFAULT_API_CONNECT_OPTIONS } from '@livekit/agents'; +import { ErrorType, SynthesiaError } from './errors.js'; +import { log } from './log.js'; +import type { StartSessionRequest, StartSessionResponse } from './types.js'; + +/** @internal */ +export const SESSION_PATH = '/api/interactive-avatars/sessions'; +const AVATAR_ID_PREFIX = 'av_'; + +const CODE_TO_ERROR: Record = { + unauthorized: ErrorType.AUTH, + invalid_api_key: ErrorType.AUTH, + unknown_avatar: ErrorType.UNKNOWN_AVATAR, + avatar_not_accessible: ErrorType.UNKNOWN_AVATAR, + quota_exceeded: ErrorType.QUOTA_EXCEEDED, + rate_limited: ErrorType.RATE_LIMITED, + invalid_token: ErrorType.INVALID_ROOM_TOKEN, + invalid_livekit_credentials: ErrorType.LIVEKIT_CREDENTIALS_REJECTED, + validation_error: ErrorType.INVALID_SESSION_REQUEST, + bad_request: ErrorType.INVALID_SESSION_REQUEST, + unknown_reference: ErrorType.UNKNOWN_AVATAR, + unauthenticated: ErrorType.AUTH, + forbidden: ErrorType.AUTH, + insufficient_scope: ErrorType.AUTH, + not_authorized: ErrorType.AUTH, + feature_not_in_plan: ErrorType.FEATURE_NOT_IN_PLAN, + payment_required: ErrorType.QUOTA_EXCEEDED, + concurrency_limit: ErrorType.CONCURRENCY_LIMIT, + concurrency_limit_exceeded: ErrorType.CONCURRENCY_LIMIT, +}; + +const STATUS_TO_ERROR: Record = { + 401: ErrorType.AUTH, + 403: ErrorType.AUTH, + 402: ErrorType.QUOTA_EXCEEDED, + 404: ErrorType.UNKNOWN_AVATAR, + 429: ErrorType.RATE_LIMITED, +}; + +/** @internal */ +export interface SynthesiaAPIOptions { + apiKey: string; + apiUrl: string; + connOptions?: APIConnectOptions; + fetch?: typeof globalThis.fetch; +} + +/** Async client for the Synthesia interactive-avatar session API. @internal */ +export class SynthesiaAPI { + #apiKey: string; + private apiUrl: string; + private connOptions: APIConnectOptions; + private fetch: typeof globalThis.fetch; + + constructor(options: SynthesiaAPIOptions) { + this.#apiKey = options.apiKey; + this.apiUrl = options.apiUrl.replace(/\/+$/, ''); + this.connOptions = options.connOptions ?? DEFAULT_API_CONNECT_OPTIONS; + this.fetch = options.fetch ?? globalThis.fetch; + } + + async startSession( + request: StartSessionRequest, + connOptions: APIConnectOptions = this.connOptions, + ): Promise { + const url = this.apiUrl + SESSION_PATH; + const payload = { + avatarIds: request.avatarIds.map(publicAvatarId), + livekitUrl: request.livekitUrl, + livekitToken: request.livekitToken, + }; + + let lastStatus: number | null = null; + let lastBody: unknown = null; + let lastCause: unknown; + + for (let attempt = 0; attempt <= connOptions.maxRetry; attempt++) { + let status: number | null = null; + try { + const response = await this.fetch(url, { + method: 'POST', + headers: { Authorization: this.#apiKey, 'Content-Type': 'application/json' }, + body: JSON.stringify(payload), + signal: AbortSignal.timeout(connOptions.timeoutMs), + }); + status = response.status; + const body = await readJson(response); + if (response.ok) return parseSuccess(body); + if (status < 500) throw mappedError(response, body); + lastStatus = status; + lastBody = body; + lastCause = undefined; + log().debug({ status }, 'synthesia session request failed, retrying'); + } catch (error) { + if (error instanceof SynthesiaError) throw error; + lastStatus = status !== null && status >= 500 ? status : null; + lastBody = null; + lastCause = error; + log().debug({ error: errorName(error) }, 'synthesia session request errored, retrying'); + } + + if (attempt < connOptions.maxRetry) { + await new Promise((resolve) => setTimeout(resolve, connOptions.retryIntervalMs)); + } + } + + throw new SynthesiaError(exhaustedMessage(connOptions.maxRetry + 1, lastStatus, lastBody), { + type: ErrorType.CONNECTION, + body: isRecord(lastBody) ? lastBody : null, + status: lastStatus, + requestId: bodyRequestId(lastBody), + cause: lastCause, + }); + } +} + +function parseSuccess(body: unknown): StartSessionResponse { + if (isRecord(body)) { + const sessionId = body.id || body.session_id; + if (typeof sessionId === 'string' && sessionId) return { sessionId }; + } + throw new SynthesiaError('Synthesia response did not contain a session id'); +} + +function mappedError(response: Response, body: unknown): SynthesiaError { + const code = bodyCode(body); + let errorType = code ? CODE_TO_ERROR[code] : undefined; + if (errorType === undefined && !(isProblem(body) && code !== null)) { + errorType = STATUS_TO_ERROR[response.status]; + } + const message = + bodyMessage(body) ?? `Synthesia request failed (${code ?? `HTTP ${response.status}`})`; + const retryAfter = + errorType === ErrorType.RATE_LIMITED || errorType === ErrorType.CONCURRENCY_LIMIT + ? parseRetryAfter(response, body) + : null; + return new SynthesiaError(message, { + type: errorType, + retryAfter, + body: isRecord(body) ? body : null, + status: response.status, + requestId: bodyRequestId(body), + }); +} + +function exhaustedMessage(attempts: number, status: number | null, body: unknown): string { + let message = 'could not start a Synthesia session'; + if (attempts > 1) message += ` after ${attempts} attempts`; + if (status === null) return `${message}; last attempt: connection error`; + message += `; last attempt: HTTP ${status}`; + const detail = bodyMessage(body); + if (detail !== null) message += `: ${detail}`; + const requestId = bodyRequestId(body); + if (requestId !== null) message += ` (request ${requestId})`; + return message; +} + +function isProblem(body: unknown): boolean { + return isRecord(body) && typeof body.type === 'string'; +} + +function bodyCode(body: unknown): string | null { + if (!isRecord(body)) return null; + const nested = isRecord(body.error) ? body.error.code : undefined; + const code = nested || body.code; + return typeof code === 'string' ? code : null; +} + +function bodyMessage(body: unknown): string | null { + if (!isRecord(body)) return null; + if (isProblem(body)) { + if (typeof body.detail === 'string' && body.detail) return body.detail; + return typeof body.title === 'string' && body.title ? body.title : null; + } + const nested = isRecord(body.error) ? body.error.message : undefined; + const detail = nested || body.message || body.context; + if (detail === undefined || detail === null) { + return typeof body.error === 'string' ? body.error : null; + } + if (typeof detail === 'string') return detail; + const rendered = JSON.stringify(detail); + const code = bodyCode(body); + return code ? `${code}: ${rendered}` : rendered; +} + +function bodyRequestId(body: unknown): string | null { + return isRecord(body) && typeof body.requestId === 'string' ? body.requestId : null; +} + +function publicAvatarId(avatarId: string): string { + return avatarId.startsWith(AVATAR_ID_PREFIX) ? avatarId : AVATAR_ID_PREFIX + avatarId; +} + +function parseRetryAfter(response: Response, body: unknown): number | null { + const header = response.headers.get('Retry-After'); + if (header !== null) { + const seconds = Number(header); + if (Number.isFinite(seconds)) return seconds * 1000; + } + if (isRecord(body)) { + const nested = isRecord(body.error) ? body.error.retry_after : undefined; + for (const value of [body.retry_after, nested]) { + if (typeof value === 'number' && Number.isFinite(value)) return value * 1000; + } + } + return null; +} + +async function readJson(response: Response): Promise { + const text = await response.text(); + try { + return JSON.parse(text) as unknown; + } catch { + return null; + } +} + +function isRecord(value: unknown): value is Record { + return value !== null && typeof value === 'object' && !Array.isArray(value); +} + +function errorName(value: unknown): string { + return value instanceof Error ? value.name : typeof value; +} diff --git a/plugins/synthesia/src/avatar.test.ts b/plugins/synthesia/src/avatar.test.ts new file mode 100644 index 000000000..8467a4ea3 --- /dev/null +++ b/plugins/synthesia/src/avatar.test.ts @@ -0,0 +1,437 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { voice } from '@livekit/agents'; +import type { Room } from '@livekit/rtc-node'; +import { RoomEvent, TrackKind } from '@livekit/rtc-node'; +import { EventEmitter } from 'node:events'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { SynthesiaAPI } from './api.js'; +import { AvatarSession } from './avatar.js'; +import { ErrorType, SynthesiaError } from './errors.js'; +import * as logModule from './log.js'; +import { AVATAR_IDENTITY, AVATAR_NAME, AvatarConfig } from './types.js'; + +const LIVEKIT = { + livekitUrl: 'wss://dev.livekit.cloud', + livekitApiKey: 'lk-api-key', + livekitApiSecret: 'lk-api-secret-never-leaks', +}; +const ADA_ID = '03cee7ec-ac90-45ec-8c20-74a399cf3dc4'; +const SECOND_ID = '6d999451-039c-4bf2-9b88-c769ac2faa78'; + +type RpcOptions = { + destinationIdentity: string; + method: string; + payload: string; + responseTimeout?: number; +}; + +function fakeRoom({ identity = 'dev-agent', connected = true } = {}) { + const emitter = new EventEmitter(); + const rpcCalls: RpcOptions[] = []; + let rpcResponse = JSON.stringify({ status: 'ok', avatar_id: SECOND_ID }); + let rpcError: unknown; + const remoteParticipant = { + identity: AVATAR_IDENTITY, + trackPublications: new Map([['video', { kind: TrackKind.KIND_VIDEO }]]), + }; + const room = { + name: 'dev-room', + isConnected: connected, + localParticipant: { + identity, + registerRpcMethod: vi.fn(), + performRpc: vi.fn(async (options: RpcOptions) => { + rpcCalls.push(options); + if (rpcError) throw rpcError; + return rpcResponse; + }), + }, + remoteParticipants: new Map([[remoteParticipant.identity, remoteParticipant]]), + on: vi.fn((event: string | symbol, listener: (...args: unknown[]) => void) => { + emitter.on(event, listener); + return room; + }), + off: vi.fn((event: string | symbol, listener: (...args: unknown[]) => void) => { + emitter.off(event, listener); + return room; + }), + emit: (event: string | symbol, ...args: unknown[]) => emitter.emit(event, ...args), + listenerCount: (event: string | symbol) => emitter.listenerCount(event), + } as unknown as Room; + return { + room, + rpcCalls, + emit(event: string | symbol, ...args: unknown[]) { + emitter.emit(event, ...args); + }, + setRpcResponse(value: string) { + rpcResponse = value; + }, + setRpcError(value: unknown) { + rpcError = value; + }, + }; +} + +function fakeAgentSession() { + const emitter = new EventEmitter(); + const output = { + audio: null as voice.AudioOutput | null, + replaceAudioTail(sink: voice.AudioOutput) { + this.audio = sink; + }, + }; + const session = { + _started: false, + output, + on: vi.fn((event: string | symbol, listener: (...args: unknown[]) => void) => { + emitter.on(event, listener); + return session; + }), + off: vi.fn((event: string | symbol, listener: (...args: unknown[]) => void) => { + emitter.off(event, listener); + return session; + }), + emit: vi.fn(), + } as unknown as voice.AgentSession; + return session; +} + +function avatar(options: ConstructorParameters[1] = {}) { + return new AvatarSession(new AvatarConfig({ avatarIds: [ADA_ID, SECOND_ID] }), { + apiKey: 'syn-key', + ...options, + }); +} + +function decodeJwt(token: string): Record { + return JSON.parse(Buffer.from(token.split('.')[1]!, 'base64url').toString()) as Record< + string, + unknown + >; +} + +describe('Synthesia AvatarSession', () => { + beforeEach(() => { + vi.spyOn(SynthesiaAPI.prototype, 'startSession').mockResolvedValue({ sessionId: 'sess_123' }); + vi.spyOn(voice.AvatarSession.prototype, 'waitForJoin').mockResolvedValue(); + }); + + afterEach(() => { + vi.restoreAllMocks(); + delete process.env.SYNTHESIA_API_KEY; + delete process.env.SYNTHESIA_API_URL; + delete process.env.LIVEKIT_URL; + delete process.env.LIVEKIT_API_KEY; + delete process.env.LIVEKIT_API_SECRET; + voice.DataStreamAudioOutput._playbackFinishedRpcRegistered = false; + voice.DataStreamAudioOutput._playbackFinishedHandlers = {}; + voice.DataStreamAudioOutput._playbackStartedRpcRegistered = false; + voice.DataStreamAudioOutput._playbackStartedHandlers = {}; + }); + + it('requires a key and accepts it from the environment', () => { + expect(() => new AvatarSession(new AvatarConfig({ avatarIds: [ADA_ID] }))).toThrow( + SynthesiaError, + ); + process.env.SYNTHESIA_API_KEY = 'env-key'; + expect(() => new AvatarSession(new AvatarConfig({ avatarIds: [ADA_ID] }))).not.toThrow(); + }); + + it('does not expose its API key when serialized', () => { + expect(JSON.stringify(avatar({ apiKey: 'super-secret' }))).not.toContain('super-secret'); + }); + + it.each([ + ['avatarParticipantIdentity', ''], + ['avatarParticipantIdentity', ' '], + ['avatarParticipantName', ''], + ['avatarParticipantName', ' '], + ] as const)('rejects blank %s', (key, value) => { + expect(() => avatar({ [key]: value })).toThrow(key); + }); + + it('calls the base start before provisioning', async () => { + const sentinel = new Error('super-start-called'); + vi.spyOn(voice.AvatarSession.prototype, 'start').mockRejectedValueOnce(sentinel); + await expect(avatar().start(fakeAgentSession(), fakeRoom().room, LIVEKIT)).rejects.toThrow( + sentinel, + ); + expect(SynthesiaAPI.prototype.startSession).not.toHaveBeenCalled(); + }); + + it('mints the expected worker token, provisions in order, and routes audio', async () => { + const agent = fakeAgentSession(); + const { room } = fakeRoom(); + const session = avatar(); + await session.start(agent, room, LIVEKIT); + + const request = vi.mocked(SynthesiaAPI.prototype.startSession).mock.calls[0]![0]; + expect(request.avatarIds).toEqual([ADA_ID, SECOND_ID]); + expect(request.livekitUrl).toBe(LIVEKIT.livekitUrl); + const claims = decodeJwt(request.livekitToken) as { + sub: string; + name: string; + kind: string; + video: Record; + attributes: Record; + exp: number; + }; + expect(claims).toMatchObject({ + sub: AVATAR_IDENTITY, + name: AVATAR_NAME, + kind: 'agent', + video: { + roomJoin: true, + room: 'dev-room', + canPublish: true, + canSubscribe: true, + canPublishData: true, + }, + attributes: { 'lk.publish_on_behalf': 'dev-agent' }, + }); + expect(claims.exp).toBeGreaterThan(Math.floor(Date.now() / 1000) + 6 * 60 * 60 - 5); + expect(session.sessionId).toBe('sess_123'); + expect(agent.output.audio).toBeInstanceOf(voice.DataStreamAudioOutput); + expect(agent.output.audio).toMatchObject({ + destinationIdentity: AVATAR_IDENTITY, + waitRemoteTrack: TrackKind.KIND_VIDEO, + }); + await session.aclose(); + }); + + it.each([ + ['https://proj.livekit.cloud', 'wss://proj.livekit.cloud'], + ['http://localhost:7880', 'ws://localhost:7880'], + ['wss://proj.livekit.cloud', 'wss://proj.livekit.cloud'], + ['ws://localhost:7880', 'ws://localhost:7880'], + ])('normalizes %s to %s', async (given, expected) => { + const session = avatar(); + await session.start(fakeAgentSession(), fakeRoom().room, { ...LIVEKIT, livekitUrl: given }); + expect(vi.mocked(SynthesiaAPI.prototype.startSession).mock.calls[0]![0].livekitUrl).toBe( + expected, + ); + await session.aclose(); + }); + + it.each(['host:7880', 'tcp://host:7880', 'proj.livekit.cloud'])( + 'rejects unusable URL %s', + async (url) => { + await expect( + avatar().start(fakeAgentSession(), fakeRoom().room, { ...LIVEKIT, livekitUrl: url }), + ).rejects.toThrow('livekitUrl'); + expect(SynthesiaAPI.prototype.startSession).not.toHaveBeenCalled(); + }, + ); + + it.each([ + { livekitUrl: '', livekitApiKey: 'key', livekitApiSecret: 'secret' }, + { livekitUrl: 'wss://host', livekitApiKey: ' ', livekitApiSecret: 'secret' }, + { livekitUrl: 'wss://host', livekitApiKey: 'key', livekitApiSecret: ' ' }, + ])('rejects missing or blank LiveKit credentials before launch', async (options) => { + await expect(avatar().start(fakeAgentSession(), fakeRoom().room, options)).rejects.toThrow( + 'LiveKit', + ); + expect(SynthesiaAPI.prototype.startSession).not.toHaveBeenCalled(); + }); + + it('uses LiveKit credentials from the environment', async () => { + process.env.LIVEKIT_URL = LIVEKIT.livekitUrl; + process.env.LIVEKIT_API_KEY = 'env-key'; + process.env.LIVEKIT_API_SECRET = 'env-secret'; + const session = avatar(); + await session.start(fakeAgentSession(), fakeRoom().room); + const token = vi.mocked(SynthesiaAPI.prototype.startSession).mock.calls[0]![0].livekitToken; + expect(decodeJwt(token).iss).toBe('env-key'); + await session.aclose(); + }); + + it('fails before launch when the standalone room has no local identity', async () => { + await expect( + avatar().start(fakeAgentSession(), fakeRoom({ identity: ' ' }).room, LIVEKIT), + ).rejects.toThrow('local participant'); + expect(SynthesiaAPI.prototype.startSession).not.toHaveBeenCalled(); + }); + + it('supports custom participant identity and name everywhere', async () => { + const session = avatar({ + avatarParticipantIdentity: 'avatar-host', + avatarParticipantName: 'Host Avatar', + }); + const room = fakeRoom(); + room.setRpcResponse(JSON.stringify({ status: 'ok', avatar_id: SECOND_ID })); + const agent = fakeAgentSession(); + await session.start(agent, room.room, LIVEKIT); + const token = vi.mocked(SynthesiaAPI.prototype.startSession).mock.calls[0]![0].livekitToken; + expect(decodeJwt(token)).toMatchObject({ sub: 'avatar-host', name: 'Host Avatar' }); + expect(session.avatarIdentity).toBe('avatar-host'); + expect(agent.output.audio).toMatchObject({ destinationIdentity: 'avatar-host' }); + await expect(session.swapAvatar(SECOND_ID)).resolves.toBe(SECOND_ID); + expect(room.rpcCalls[0]).toMatchObject({ destinationIdentity: 'avatar-host' }); + await session.aclose(); + }); + + it('sends swap RPCs and supports default', async () => { + const room = fakeRoom(); + const session = avatar(); + await session.start(fakeAgentSession(), room.room, LIVEKIT); + await expect(session.swapAvatar(SECOND_ID)).resolves.toBe(SECOND_ID); + expect(room.rpcCalls[0]).toEqual({ + destinationIdentity: AVATAR_IDENTITY, + method: 'swapAvatar', + payload: JSON.stringify({ avatar_id: SECOND_ID }), + responseTimeout: 15_000, + }); + room.setRpcResponse(JSON.stringify({ status: 'ok', avatar_id: ADA_ID })); + await expect(session.swapAvatar('default')).resolves.toBe(ADA_ID); + await session.aclose(); + }); + + it('rejects swaps not precomputed without making an RPC', async () => { + const room = fakeRoom(); + const session = avatar(); + await session.start(fakeAgentSession(), room.room, LIVEKIT); + await expect(session.swapAvatar('not-in-list')).rejects.toMatchObject({ + type: ErrorType.UNKNOWN_AVATAR, + }); + expect(room.rpcCalls).toHaveLength(0); + await session.aclose(); + }); + + it.each([ + ['worker error', JSON.stringify({ error: 'swap timeout' }), 'swap timeout'], + ['unrecognized response', JSON.stringify({ status: 'weird' }), 'weird'], + ['malformed response', 'not json', 'malformed'], + ])('surfaces a %s', async (_case, raw, message) => { + const room = fakeRoom(); + room.setRpcResponse(raw); + const session = avatar(); + await session.start(fakeAgentSession(), room.room, LIVEKIT); + await expect(session.swapAvatar(SECOND_ID)).rejects.toThrow(message); + await session.aclose(); + }); + + it('maps swap transport failures to connection errors', async () => { + const room = fakeRoom(); + room.setRpcError(new Error('rpc transport down')); + const session = avatar(); + await session.start(fakeAgentSession(), room.room, LIVEKIT); + await expect(session.swapAvatar(SECOND_ID)).rejects.toMatchObject({ + type: ErrorType.CONNECTION, + }); + await session.aclose(); + }); + + it('rejects swaps before start and after close', async () => { + const session = avatar(); + await expect(session.swapAvatar(SECOND_ID)).rejects.toThrow('started'); + await session.start(fakeAgentSession(), fakeRoom().room, LIVEKIT); + await session.aclose(); + await expect(session.swapAvatar(SECOND_ID)).rejects.toThrow('started'); + }); + + it('maps join timeout and tears down its exact audio output', async () => { + vi.mocked(voice.AvatarSession.prototype.waitForJoin).mockRejectedValueOnce( + new Error('timed out waiting for avatar participant'), + ); + const close = vi.spyOn(voice.DataStreamAudioOutput.prototype, 'aclose'); + const session = avatar({ joinTimeout: 50 }); + await expect(session.start(fakeAgentSession(), fakeRoom().room, LIVEKIT)).rejects.toMatchObject( + { + type: ErrorType.TIMEOUT, + message: 'avatar did not join within 50ms', + }, + ); + expect(close).toHaveBeenCalledTimes(1); + }); + + it('tears down after a mapped launch failure and can retry', async () => { + vi.mocked(SynthesiaAPI.prototype.startSession) + .mockRejectedValueOnce(new SynthesiaError('bad key', { type: ErrorType.AUTH })) + .mockResolvedValueOnce({ sessionId: 'sess_retry' }); + const session = avatar(); + const room = fakeRoom(); + const agent = fakeAgentSession(); + await expect(session.start(agent, room.room, LIVEKIT)).rejects.toMatchObject({ + type: ErrorType.AUTH, + }); + expect(agent.output.audio).toBeNull(); + await session.start(agent, room.room, LIVEKIT); + expect(session.sessionId).toBe('sess_retry'); + await session.aclose(); + }); + + it('is idempotent across double start and double close', async () => { + const session = avatar(); + const room = fakeRoom(); + const agent = fakeAgentSession(); + await session.start(agent, room.room, LIVEKIT); + await session.start(agent, room.room, LIVEKIT); + expect(SynthesiaAPI.prototype.startSession).toHaveBeenCalledTimes(1); + await session.aclose(); + await session.aclose(); + }); + + it.each([ + ['track', RoomEvent.TrackUnpublished], + ['participant', RoomEvent.ParticipantDisconnected], + ])('logs an unexpected %s loss once and tears down', async (kind, event) => { + const logger = logModule.log(); + const warn = vi.spyOn(logger, 'warn'); + vi.spyOn(logModule, 'log').mockReturnValue(logger); + const close = vi.spyOn(voice.DataStreamAudioOutput.prototype, 'aclose'); + const room = fakeRoom(); + const session = avatar(); + await session.start(fakeAgentSession(), room.room, LIVEKIT); + const participant = { identity: AVATAR_IDENTITY }; + if (event === RoomEvent.TrackUnpublished) { + room.emit(event, { kind: TrackKind.KIND_VIDEO }, participant); + } else { + room.emit(event, participant); + } + await vi.waitFor(() => expect(close).toHaveBeenCalledTimes(1)); + expect(warn).toHaveBeenCalledWith('avatar left the room unexpectedly'); + expect(kind).toBeTruthy(); + }); + + it('logs a clean room end, tears down, and unregisters lifecycle handlers', async () => { + const logger = logModule.log(); + const info = vi.spyOn(logger, 'info'); + vi.spyOn(logModule, 'log').mockReturnValue(logger); + const room = fakeRoom(); + const session = avatar(); + await session.start(fakeAgentSession(), room.room, LIVEKIT); + expect(room.room.listenerCount(RoomEvent.Disconnected)).toBeGreaterThan(0); + room.emit(RoomEvent.Disconnected); + await vi.waitFor(() => expect(room.room.listenerCount(RoomEvent.TrackUnpublished)).toBe(0)); + expect(info).toHaveBeenCalledWith('avatar session ended'); + }); + + it('ignores another avatar identity disconnect', async () => { + const logger = logModule.log(); + const warn = vi.spyOn(logger, 'warn'); + vi.spyOn(logModule, 'log').mockReturnValue(logger); + const room = fakeRoom(); + const session = avatar({ avatarParticipantIdentity: 'avatar-host' }); + await session.start(fakeAgentSession(), room.room, LIVEKIT); + room.emit(RoomEvent.ParticipantDisconnected, { identity: AVATAR_IDENTITY }); + expect(warn).not.toHaveBeenCalledWith('avatar left the room unexpectedly'); + await session.aclose(); + }); + + it('mirrors the README attach-before-agent usage', async () => { + process.env.SYNTHESIA_API_KEY = 'syn_live_key'; + process.env.LIVEKIT_URL = LIVEKIT.livekitUrl; + process.env.LIVEKIT_API_KEY = LIVEKIT.livekitApiKey; + process.env.LIVEKIT_API_SECRET = LIVEKIT.livekitApiSecret; + const session = fakeAgentSession(); + const room = fakeRoom(); + const attachedAvatar = new AvatarSession(new AvatarConfig({ avatarIds: [ADA_ID] })); + await attachedAvatar.start(session, room.room); + expect(session.output.audio).not.toBeNull(); + expect(session.output.audio).toMatchObject({ destinationIdentity: AVATAR_IDENTITY }); + await attachedAvatar.aclose(); + }); +}); diff --git a/plugins/synthesia/src/avatar.ts b/plugins/synthesia/src/avatar.ts new file mode 100644 index 000000000..270d23327 --- /dev/null +++ b/plugins/synthesia/src/avatar.ts @@ -0,0 +1,351 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { DEFAULT_API_CONNECT_OPTIONS, getJobContext, voice } from '@livekit/agents'; +import type { RemoteParticipant, RemoteTrackPublication, Room } from '@livekit/rtc-node'; +import { RoomEvent, TrackKind } from '@livekit/rtc-node'; +import type { VideoGrant } from 'livekit-server-sdk'; +import { AccessToken } from 'livekit-server-sdk'; +import { SynthesiaAPI } from './api.js'; +import { ErrorType, SynthesiaError } from './errors.js'; +import { log } from './log.js'; +import { + AVATAR_IDENTITY, + AVATAR_NAME, + type AvatarConfig, + DEFAULT_API_URL, + DEFAULT_JOIN_TIMEOUT, + DEFAULT_SWAP_TIMEOUT, + TOKEN_TTL, +} from './types.js'; + +const ATTRIBUTE_PUBLISH_ON_BEHALF = 'lk.publish_on_behalf'; + +enum State { + IDLE, + STARTED, + CLOSED, +} + +/** Options for configuring an AvatarSession. @public */ +export interface AvatarSessionOptions { + /** Synthesia API key. Falls back to `SYNTHESIA_API_KEY`. */ + apiKey?: string | null; + /** Synthesia API URL. Falls back to `SYNTHESIA_API_URL`. */ + apiUrl?: string | null; + /** Maximum time in milliseconds to wait for the avatar to join. */ + joinTimeout?: number; + /** Identity for the avatar participant. */ + avatarParticipantIdentity?: string | null; + /** Display name for the avatar participant. */ + avatarParticipantName?: string | null; +} + +/** Optional LiveKit credentials for {@link AvatarSession.start}. @public */ +export interface StartOptions { + livekitUrl?: string | null; + livekitApiKey?: string | null; + livekitApiSecret?: string | null; +} + +/** A Synthesia interactive avatar for a LiveKit voice agent. @public */ +export class AvatarSession extends voice.AvatarSession { + private avatarIds: readonly string[]; + #apiKey: string; + private apiUrl: string; + private joinTimeout: number; + private avatarParticipantIdentity: string; + private avatarParticipantName: string; + private state = State.IDLE; + private starting = false; + private room?: Room; + private sessionIdValue: string | null = null; + private audioOutput?: voice.DataStreamAudioOutput; + private closePromise?: Promise; + private teardownPromise?: Promise; + + constructor(avatarConfig: AvatarConfig, options: AvatarSessionOptions = {}) { + super(); + if (options.avatarParticipantIdentity !== undefined) { + requirePresent(options.avatarParticipantIdentity, 'avatarParticipantIdentity'); + } + if (options.avatarParticipantName !== undefined) { + requirePresent(options.avatarParticipantName, 'avatarParticipantName'); + } + const apiKey = options.apiKey ?? process.env.SYNTHESIA_API_KEY; + if (!apiKey) { + throw new SynthesiaError( + 'a Synthesia API key is required: pass apiKey or set SYNTHESIA_API_KEY', + ); + } + this.avatarIds = [...avatarConfig.avatarIds]; + this.#apiKey = apiKey; + this.apiUrl = options.apiUrl ?? process.env.SYNTHESIA_API_URL ?? DEFAULT_API_URL; + this.joinTimeout = options.joinTimeout ?? DEFAULT_JOIN_TIMEOUT; + this.avatarParticipantIdentity = options.avatarParticipantIdentity ?? AVATAR_IDENTITY; + this.avatarParticipantName = options.avatarParticipantName ?? AVATAR_NAME; + } + + override get avatarIdentity(): string { + return this.avatarParticipantIdentity; + } + + override get provider(): string { + return 'synthesia'; + } + + /** Synthesia session ID after provisioning succeeds, otherwise `null`. */ + get sessionId(): string | null { + return this.sessionIdValue; + } + + async start( + agentSession: voice.AgentSession, + room: Room, + options: StartOptions = {}, + ): Promise { + if (this.starting) throw new SynthesiaError('start() is already in progress'); + this.starting = true; + try { + if (this.closePromise) { + try { + await this.closePromise; + } catch (cause) { + throw new SynthesiaError( + 'the previous avatar session did not tear down cleanly; not restarting', + { cause }, + ); + } + this.closePromise = undefined; + this.teardownPromise = undefined; + } + if (this.state === State.STARTED) return; + + const livekitUrl = options.livekitUrl ?? process.env.LIVEKIT_URL; + const livekitApiKey = options.livekitApiKey ?? process.env.LIVEKIT_API_KEY; + const livekitApiSecret = options.livekitApiSecret ?? process.env.LIVEKIT_API_SECRET; + if (![livekitUrl, livekitApiKey, livekitApiSecret].every(isPresent)) { + throw new SynthesiaError( + 'LiveKit url, API key, and API secret are required: pass them or set LIVEKIT_URL, LIVEKIT_API_KEY, LIVEKIT_API_SECRET', + ); + } + const normalizedUrl = toWebSocketUrl(livekitUrl!); + if (!normalizedUrl.startsWith('ws://') && !normalizedUrl.startsWith('wss://')) { + throw new SynthesiaError( + `livekitUrl ${JSON.stringify(normalizedUrl)} is not a ws:// or wss:// URL`, + ); + } + + this.state = State.IDLE; + this.room = room; + try { + await super.start(agentSession, room); + const api = new SynthesiaAPI({ apiKey: this.#apiKey, apiUrl: this.apiUrl }); + const response = await api.startSession( + { + avatarIds: this.avatarIds, + livekitUrl: normalizedUrl, + livekitToken: await this.mintToken(room, livekitApiKey!, livekitApiSecret!), + }, + { ...DEFAULT_API_CONNECT_OPTIONS, timeoutMs: this.joinTimeout, maxRetry: 0 }, + ); + this.sessionIdValue = response.sessionId; + const audioOutput = new voice.DataStreamAudioOutput({ + room, + destinationIdentity: this.avatarIdentity, + waitRemoteTrack: TrackKind.KIND_VIDEO, + }); + agentSession.output.replaceAudioTail(audioOutput); + this.audioOutput = audioOutput; + await this.waitForJoin({ timeout: this.joinTimeout }); + } catch (error) { + await this.aclose().catch(() => undefined); + await this.discardPartialStart(); + if (isJoinTimeout(error)) { + throw new SynthesiaError(`avatar did not join within ${this.joinTimeout}ms`, { + type: ErrorType.TIMEOUT, + cause: error, + }); + } + throw error; + } + + if (this.isClosed()) { + await this.closePromise; + await this.discardPartialStart(); + throw new SynthesiaError('avatar session was closed while starting'); + } + room.on(RoomEvent.Disconnected, this.onRoomDisconnected); + room.on(RoomEvent.TrackUnpublished, this.onTrackUnpublished); + room.on(RoomEvent.ParticipantDisconnected, this.onParticipantDisconnected); + this.state = State.STARTED; + } finally { + this.starting = false; + } + } + + /** Switch the rendered avatar during a started session. */ + async swapAvatar(avatarId: string, { timeout = DEFAULT_SWAP_TIMEOUT } = {}): Promise { + if (this.state !== State.STARTED || this.teardownPromise || !this.room) { + throw new SynthesiaError('swapAvatar() requires a started avatar session'); + } + if (avatarId !== 'default' && !this.avatarIds.includes(avatarId)) { + throw new SynthesiaError(`avatar ${JSON.stringify(avatarId)} was not in initial avatarIds`, { + type: ErrorType.UNKNOWN_AVATAR, + }); + } + + let raw: string; + try { + raw = await this.room.localParticipant!.performRpc({ + destinationIdentity: this.avatarIdentity, + method: 'swapAvatar', + payload: JSON.stringify({ avatar_id: avatarId }), + responseTimeout: timeout, + }); + } catch (cause) { + throw new SynthesiaError(`avatar swap RPC failed: ${String(cause)}`, { + type: ErrorType.CONNECTION, + cause, + }); + } + + let response: unknown; + try { + response = JSON.parse(raw) as unknown; + } catch (cause) { + throw new SynthesiaError('avatar swap returned a malformed response', { cause }); + } + const result = isRecord(response) ? response.avatar_id : undefined; + if (!isRecord(response) || response.error || typeof result !== 'string') { + const detail = isRecord(response) ? response.error : undefined; + throw new SynthesiaError(`avatar swap failed: ${detail || raw}`); + } + return result; + } + + override async aclose(): Promise { + if (this.closePromise) { + await this.closePromise.catch(() => undefined); + return; + } + this.state = State.CLOSED; + this.sessionIdValue = null; + this.closePromise = this.closeImpl(); + return this.closePromise; + } + + private async closeImpl(): Promise { + try { + const audioOutput = this.audioOutput; + this.audioOutput = undefined; + await audioOutput?.aclose(); + if (this.room) { + this.room.off(RoomEvent.Disconnected, this.onRoomDisconnected); + this.room.off(RoomEvent.TrackUnpublished, this.onTrackUnpublished); + this.room.off(RoomEvent.ParticipantDisconnected, this.onParticipantDisconnected); + } + await super.aclose(); + } finally { + this.room = undefined; + } + } + + private async discardPartialStart(): Promise { + const audioOutput = this.audioOutput; + this.audioOutput = undefined; + await audioOutput?.aclose(); + this.sessionIdValue = null; + } + + private async mintToken(room: Room, apiKey: string, apiSecret: string): Promise { + const jobContext = getJobContext(false); + const agentIdentity = + jobContext?.agent?.identity ?? + room.localParticipant?.identity ?? + jobContext?.info.acceptArguments.identity; + if (!isPresent(agentIdentity)) { + throw new SynthesiaError( + "the room's local participant has no identity; connect the room before starting the avatar session", + ); + } + const roomName = room.name || jobContext?.job.room?.name; + if (!isPresent(roomName)) { + throw new SynthesiaError( + 'failed to get room name; connect the room before starting outside a job context', + ); + } + const token = new AccessToken(apiKey, apiSecret, { + identity: this.avatarIdentity, + name: this.avatarParticipantName, + ttl: TOKEN_TTL, + }); + token.kind = 'agent'; + token.addGrant({ + roomJoin: true, + room: roomName, + canPublish: true, + canSubscribe: true, + canPublishData: true, + } as VideoGrant); + token.attributes = { [ATTRIBUTE_PUBLISH_ON_BEHALF]: agentIdentity }; + return token.toJwt(); + } + + private onRoomDisconnected = () => { + if (this.teardownPromise || this.state === State.CLOSED) return; + log().info('avatar session ended'); + this.beginTeardown(); + }; + + private onTrackUnpublished = ( + publication: RemoteTrackPublication, + participant: RemoteParticipant, + ) => { + if (participant.identity === this.avatarIdentity && publication.kind === TrackKind.KIND_VIDEO) { + this.reportAvatarLost(); + } + }; + + private onParticipantDisconnected = (participant: RemoteParticipant) => { + if (participant.identity === this.avatarIdentity) this.reportAvatarLost(); + }; + + private reportAvatarLost() { + if (this.teardownPromise || this.state === State.CLOSED || !this.room?.isConnected) return; + log().warn('avatar left the room unexpectedly'); + this.beginTeardown(); + } + + private beginTeardown() { + if (this.teardownPromise || this.state === State.CLOSED) return; + this.teardownPromise = this.aclose(); + void this.teardownPromise.catch((error) => log().error({ error }, 'avatar teardown failed')); + } + + private isClosed(): boolean { + return this.state === State.CLOSED; + } +} + +function isPresent(value: string | null | undefined): value is string { + return Boolean(value?.trim()); +} + +function requirePresent(value: string | null, name: string): asserts value is string { + if (!isPresent(value)) throw new SynthesiaError(`${name} must be a non-empty string`); +} + +function toWebSocketUrl(url: string): string { + if (url.startsWith('https://')) return 'wss://' + url.slice('https://'.length); + if (url.startsWith('http://')) return 'ws://' + url.slice('http://'.length); + return url; +} + +function isJoinTimeout(error: unknown): boolean { + return error instanceof Error && error.message === 'timed out waiting for avatar participant'; +} + +function isRecord(value: unknown): value is Record { + return value !== null && typeof value === 'object' && !Array.isArray(value); +} diff --git a/plugins/synthesia/src/errors.test.ts b/plugins/synthesia/src/errors.test.ts new file mode 100644 index 000000000..c1cbb6e29 --- /dev/null +++ b/plugins/synthesia/src/errors.test.ts @@ -0,0 +1,59 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { APIError } from '@livekit/agents'; +import { describe, expect, it } from 'vitest'; +import { ErrorType, SynthesiaError } from './errors.js'; + +describe('Synthesia errors', () => { + const nonRetryable = [ + ErrorType.AUTH, + ErrorType.FEATURE_NOT_IN_PLAN, + ErrorType.UNKNOWN_AVATAR, + ErrorType.QUOTA_EXCEEDED, + ErrorType.INVALID_ROOM_TOKEN, + ErrorType.LIVEKIT_CREDENTIALS_REJECTED, + ErrorType.INVALID_SESSION_REQUEST, + ]; + const retryable = [ + ErrorType.RATE_LIMITED, + ErrorType.CONCURRENCY_LIMIT, + ErrorType.TIMEOUT, + ErrorType.CONNECTION, + ]; + + it('extends APIError and defaults untyped errors to non-retryable', () => { + const error = new SynthesiaError('boom'); + expect(error).toBeInstanceOf(APIError); + expect(error).toMatchObject({ type: null, retryable: false, retryAfter: null }); + }); + + it.each(nonRetryable)('defaults %s to non-retryable', (type) => { + expect(new SynthesiaError('boom', { type }).retryable).toBe(false); + }); + + it.each(retryable)('defaults %s to retryable', (type) => { + expect(new SynthesiaError('boom', { type }).retryable).toBe(true); + }); + + it('allows retryability overrides and carries metadata', () => { + expect(new SynthesiaError('boom', { type: ErrorType.AUTH, retryable: true }).retryable).toBe( + true, + ); + expect( + new SynthesiaError('boom', { type: ErrorType.CONNECTION, retryable: false }).retryable, + ).toBe(false); + expect( + new SynthesiaError('throttled', { + type: ErrorType.RATE_LIMITED, + retryAfter: 12_500, + status: 429, + requestId: 'req_1', + }), + ).toMatchObject({ retryAfter: 12_500, status: 429, requestId: 'req_1' }); + }); + + it('safely stringifies non-string messages', () => { + expect(new SynthesiaError({ context: ['boom'] }).message).toContain('boom'); + }); +}); diff --git a/plugins/synthesia/src/errors.ts b/plugins/synthesia/src/errors.ts new file mode 100644 index 000000000..deac2ce61 --- /dev/null +++ b/plugins/synthesia/src/errors.ts @@ -0,0 +1,70 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { APIError } from '@livekit/agents'; + +/** What went wrong, for callers that want to branch on it. @public */ +export enum ErrorType { + AUTH = 'auth', + FEATURE_NOT_IN_PLAN = 'feature_not_in_plan', + INVALID_ROOM_TOKEN = 'invalid_room_token', + LIVEKIT_CREDENTIALS_REJECTED = 'livekit_credentials_rejected', + INVALID_SESSION_REQUEST = 'invalid_session_request', + UNKNOWN_AVATAR = 'unknown_avatar', + QUOTA_EXCEEDED = 'quota_exceeded', + RATE_LIMITED = 'rate_limited', + CONCURRENCY_LIMIT = 'concurrency_limit', + TIMEOUT = 'timeout', + CONNECTION = 'connection', +} + +const RETRYABLE_TYPES = new Set([ + ErrorType.RATE_LIMITED, + ErrorType.CONCURRENCY_LIMIT, + ErrorType.TIMEOUT, + ErrorType.CONNECTION, +]); + +/** @public */ +export interface SynthesiaErrorOptions { + type?: ErrorType | null; + body?: object | null; + retryable?: boolean; + /** Server-provided backoff in milliseconds. */ + retryAfter?: number | null; + status?: number | null; + requestId?: string | null; + cause?: unknown; +} + +/** Every error raised by the Synthesia plugin. @public */ +export class SynthesiaError extends APIError { + readonly type: ErrorType | null; + readonly retryAfter: number | null; + readonly status: number | null; + readonly requestId: string | null; + + constructor(message: unknown, options: SynthesiaErrorOptions = {}) { + const type = options.type ?? null; + super(stringify(message), { + body: options.body, + retryable: options.retryable ?? RETRYABLE_TYPES.has(type as ErrorType), + }); + this.name = 'SynthesiaError'; + this.type = type; + this.retryAfter = options.retryAfter ?? null; + this.status = options.status ?? null; + this.requestId = options.requestId ?? null; + if (options.cause !== undefined) this.cause = options.cause; + Error.captureStackTrace(this, SynthesiaError); + } +} + +function stringify(value: unknown): string { + if (typeof value === 'string') return value; + try { + return JSON.stringify(value); + } catch { + return String(value); + } +} diff --git a/plugins/synthesia/src/index.test.ts b/plugins/synthesia/src/index.test.ts new file mode 100644 index 000000000..2ff5aaeb5 --- /dev/null +++ b/plugins/synthesia/src/index.test.ts @@ -0,0 +1,28 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { Plugin } from '@livekit/agents'; +import { describe, expect, it } from 'vitest'; +import { AvatarConfig, ErrorType, SynthesiaError } from './index.js'; + +describe('Synthesia plugin and config', () => { + it('registers the plugin and exports its error taxonomy', () => { + expect(Plugin.registeredPlugins.some((plugin) => plugin.title === 'synthesia')).toBe(true); + expect(ErrorType.UNKNOWN_AVATAR).toBe('unknown_avatar'); + expect(new SynthesiaError('x')).toBeInstanceOf(SynthesiaError); + }); + + it.each([0, 6])('rejects %i avatar IDs', (length) => { + expect( + () => new AvatarConfig({ avatarIds: Array.from({ length }, (_, i) => `a-${i}`) }), + ).toThrow('between 1 and 5'); + }); + + it('rejects a bare string and copies the input while preserving order', () => { + expect(() => new AvatarConfig({ avatarIds: 'lucas' as unknown as string[] })).toThrow('list'); + const ids = ['first', 'second']; + const config = new AvatarConfig({ avatarIds: ids }); + ids.push('third'); + expect(config.avatarIds).toEqual(['first', 'second']); + }); +}); diff --git a/plugins/synthesia/src/index.ts b/plugins/synthesia/src/index.ts new file mode 100644 index 000000000..f9402a826 --- /dev/null +++ b/plugins/synthesia/src/index.ts @@ -0,0 +1,23 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { Plugin } from '@livekit/agents'; + +export { AvatarSession } from './avatar.js'; +export type { AvatarSessionOptions, StartOptions } from './avatar.js'; +export { ErrorType, SynthesiaError } from './errors.js'; +export type { SynthesiaErrorOptions } from './errors.js'; +export { AvatarConfig } from './types.js'; +export type { AvatarConfigOptions } from './types.js'; + +class SynthesiaPlugin extends Plugin { + constructor() { + super({ + title: 'synthesia', + version: __PACKAGE_VERSION__, + package: '@livekit/agents-plugin-synthesia', + }); + } +} + +Plugin.registerPlugin(new SynthesiaPlugin()); diff --git a/plugins/synthesia/src/log.ts b/plugins/synthesia/src/log.ts new file mode 100644 index 000000000..7b12e48cb --- /dev/null +++ b/plugins/synthesia/src/log.ts @@ -0,0 +1,7 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { log as agentsLog } from '@livekit/agents'; +import type { Logger } from 'pino'; + +export const log = (): Logger => agentsLog().child({ plugin: 'synthesia' }); diff --git a/plugins/synthesia/src/types.ts b/plugins/synthesia/src/types.ts new file mode 100644 index 000000000..bb885605f --- /dev/null +++ b/plugins/synthesia/src/types.ts @@ -0,0 +1,53 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 + +/** @public */ +export const DEFAULT_API_URL = 'https://developers.synthesia.io'; +/** @public */ +export const DEFAULT_JOIN_TIMEOUT = 30_000; +/** @public */ +export const DEFAULT_SWAP_TIMEOUT = 15_000; +/** @public */ +export const AVATAR_IDENTITY = 'synthesia-avatar-agent'; +/** @public */ +export const AVATAR_NAME = 'Synthesia avatar'; +/** @public */ +export const MAX_AVATAR_IDS = 5; +/** @internal */ +export const TOKEN_TTL = '6h'; + +/** @public */ +export interface AvatarConfigOptions { + /** One to five gallery avatar IDs. The first ID is initially active. */ + avatarIds: readonly string[]; +} + +/** The avatars to render in the room. @public */ +export class AvatarConfig { + readonly avatarIds: readonly string[]; + + constructor({ avatarIds }: AvatarConfigOptions) { + if (!Array.isArray(avatarIds)) { + throw new TypeError('avatarIds must be a list of ids, not a single string'); + } + if (avatarIds.length < 1 || avatarIds.length > MAX_AVATAR_IDS) { + throw new RangeError( + `avatarIds must contain between 1 and ${MAX_AVATAR_IDS} ids, got ${avatarIds.length}`, + ); + } + this.avatarIds = [...avatarIds]; + } +} + +/** @internal */ +export interface StartSessionRequest { + avatarIds: readonly string[]; + livekitUrl: string; + livekitToken: string; +} + +/** @internal */ +export interface StartSessionResponse { + sessionId: string; +} diff --git a/plugins/synthesia/tsconfig.json b/plugins/synthesia/tsconfig.json new file mode 100644 index 000000000..f7f211347 --- /dev/null +++ b/plugins/synthesia/tsconfig.json @@ -0,0 +1,15 @@ +{ + "extends": "../../tsconfig.json", + "include": ["./src"], + "compilerOptions": { + "rootDir": "./src", + "declarationDir": "./dist", + "outDir": "./dist" + }, + "typedocOptions": { + "name": "plugins/agents-plugin-synthesia", + "entryPointStrategy": "resolve", + "readme": "none", + "entryPoints": ["src/index.ts"] + } +} diff --git a/plugins/synthesia/tsup.config.ts b/plugins/synthesia/tsup.config.ts new file mode 100644 index 000000000..46011fa8c --- /dev/null +++ b/plugins/synthesia/tsup.config.ts @@ -0,0 +1,9 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { defineConfig } from 'tsup'; +import defaults from '../../tsup.config.js'; + +export default defineConfig({ + ...defaults, +}); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 9acd3d512..4f67ff53d 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1443,6 +1443,31 @@ importers: specifier: ^5.0.0 version: 5.9.3 + plugins/synthesia: + dependencies: + livekit-server-sdk: + specifier: ^2.13.3 + version: 2.14.1 + devDependencies: + '@livekit/agents': + specifier: workspace:* + version: link:../../agents + '@livekit/rtc-node': + specifier: 'catalog:' + version: 0.13.34 + '@microsoft/api-extractor': + specifier: ^7.58.12 + version: 7.58.12(@types/node@25.6.0) + pino: + specifier: ^8.19.0 + version: 8.21.0 + tsup: + specifier: ^8.3.5 + version: 8.4.0(@microsoft/api-extractor@7.58.12(@types/node@25.6.0))(postcss@8.5.23)(tsx@4.23.1)(typescript@5.9.3) + typescript: + specifier: ^5.0.0 + version: 5.9.3 + plugins/tavus: dependencies: livekit-server-sdk: diff --git a/turbo.json b/turbo.json index 8babfa9f0..3bf9ed1c3 100644 --- a/turbo.json +++ b/turbo.json @@ -99,6 +99,8 @@ "RUNWAY_AVATAR_PRESET_ID", "SARVAM_API_KEY", "SONIOX_API_KEY", + "SYNTHESIA_API_KEY", + "SYNTHESIA_API_URL", "SIP_PARTICIPANT_IDENTITY", "SIP_PHONE_NUMBER", "LK_OPENAI_DEBUG",