diff --git a/.changeset/calm-fallback-errors.md b/.changeset/calm-fallback-errors.md new file mode 100644 index 000000000..e972e1aff --- /dev/null +++ b/.changeset/calm-fallback-errors.md @@ -0,0 +1,5 @@ +--- +'@livekit/agents': patch +--- + +Keep child TTS failures internal while the fallback adapter switches providers, including when non-streaming providers are wrapped for streaming. diff --git a/agents/src/tts/fallback_adapter.test.ts b/agents/src/tts/fallback_adapter.test.ts index d7d0287d6..8d7082183 100644 --- a/agents/src/tts/fallback_adapter.test.ts +++ b/agents/src/tts/fallback_adapter.test.ts @@ -9,7 +9,7 @@ import { initializeLogger } from '../log.js'; import type { APIConnectOptions } from '../types.js'; import { USERDATA_TTS_STARTED_TIME } from '../types.js'; import { FallbackAdapter } from './fallback_adapter.js'; -import { ChunkedStream, SynthesizeStream, TTS } from './tts.js'; +import { ChunkedStream, SynthesizeStream, TTS, type TTSError } from './tts.js'; const SAMPLE_RATE = 24000; @@ -105,8 +105,8 @@ class MockTTS extends TTS { /** The started time the stream recorded when it "sent" text to the provider. */ lastMarkedTime?: number; - constructor(label: string, sampleRate: number = SAMPLE_RATE) { - super(sampleRate, 1, { streaming: true }); + constructor(label: string, sampleRate: number = SAMPLE_RATE, streaming = true) { + super(sampleRate, 1, { streaming }); this.label = label; } @@ -172,6 +172,136 @@ describe('TTS FallbackAdapter', () => { await adapter.close(); }); + it('does not forward a child error when the fallback TTS succeeds', async () => { + const primary = new MockTTS('primary'); + primary.shouldFail = true; + const secondary = new MockTTS('secondary'); + const adapter = new FallbackAdapter({ + ttsInstances: [primary, secondary], + maxRetryPerTTS: 0, + recoveryDelayMs: 60_000, + }); + const errors: TTSError[] = []; + adapter.on('error', (error) => errors.push(error)); + expect(primary.listenerCount('error')).toBe(1); + + const stream = adapter.stream(); + stream.updateInputStream( + new ReadableStream({ + start(controller) { + controller.enqueue('hello world'); + controller.close(); + }, + }), + ); + + let frameCount = 0; + for await (const event of stream) { + if (event === SynthesizeStream.END_OF_STREAM) break; + frameCount++; + } + + expect(frameCount).toBeGreaterThan(0); + expect(errors).toEqual([]); + + stream.close(); + await adapter.close(); + expect(primary.listenerCount('error')).toBe(0); + }); + + it('cleans up every private error sink when the same TTS is listed more than once', async () => { + const shared = new MockTTS('shared'); + const externalErrorListener = () => {}; + shared.on('error', externalErrorListener); + const adapter = new FallbackAdapter({ + ttsInstances: [shared, shared], + }); + + expect(shared.listenerCount('error')).toBe(3); + + await adapter.close(); + + expect(shared.listenerCount('error')).toBe(1); + shared.off('error', externalErrorListener); + }); + + it('falls back from a non-streaming TTS without forwarding errors or leaking listeners', async () => { + const primary = new MockTTS('primary', SAMPLE_RATE, false); + primary.shouldFail = true; + const secondary = new MockTTS('secondary'); + const adapter = new FallbackAdapter({ + ttsInstances: [primary, secondary], + maxRetryPerTTS: 0, + recoveryDelayMs: 60_000, + }); + const errors: TTSError[] = []; + adapter.on('error', (error) => errors.push(error)); + expect(primary.listenerCount('error')).toBe(1); + + const stream = adapter.stream(); + stream.updateInputStream( + new ReadableStream({ + start(controller) { + controller.enqueue('hello world'); + controller.close(); + }, + }), + ); + + let frameCount = 0; + for await (const event of stream) { + if (event === SynthesizeStream.END_OF_STREAM) break; + frameCount++; + } + + expect(frameCount).toBeGreaterThan(0); + expect(errors).toEqual([]); + expect(primary.listenerCount('error')).toBe(1); + + stream.close(); + await adapter.close(); + expect(primary.listenerCount('error')).toBe(0); + }); + + it('emits the adapter terminal error when every TTS fails', async () => { + const primary = new MockTTS('primary'); + primary.shouldFail = true; + const secondary = new MockTTS('secondary'); + secondary.shouldFail = true; + const adapter = new FallbackAdapter({ + ttsInstances: [primary, secondary], + maxRetryPerTTS: 0, + recoveryDelayMs: 60_000, + }); + const errors: TTSError[] = []; + adapter.on('error', (error) => errors.push(error)); + + const stream = adapter.stream(); + stream.updateInputStream( + new ReadableStream({ + start(controller) { + controller.enqueue('hello world'); + controller.close(); + }, + }), + ); + + for await (const event of stream) { + if (event === SynthesizeStream.END_OF_STREAM) break; + } + + expect(errors).toHaveLength(1); + expect(errors[0]).toMatchObject({ + type: 'tts_error', + label: 'tts.FallbackAdapter', + recoverable: false, + }); + expect(errors[0]!.error.message).toBe('all TTS instances failed (primary, secondary)'); + + stream.close(); + await adapter.close(); + }); + it('should fall back when the primary has a mismatched sample rate and emits no audio', async () => { // Primary runs at 22050Hz, adapter aggregates at 24000Hz → a resampler is // created for the primary. The primary throws with no frames ever pushed, diff --git a/agents/src/tts/fallback_adapter.ts b/agents/src/tts/fallback_adapter.ts index 903231bd7..02a1a4ae5 100644 --- a/agents/src/tts/fallback_adapter.ts +++ b/agents/src/tts/fallback_adapter.ts @@ -9,7 +9,13 @@ import { basic } from '../tokenize/index.js'; import { type APIConnectOptions, DEFAULT_API_CONNECT_OPTIONS } from '../types.js'; import { Task, cancelAndWait } from '../utils.js'; import { StreamAdapter } from './stream_adapter.js'; -import { ChunkedStream, SynthesizeStream, TTS, type TTSCapabilities } from './tts.js'; +import { + ChunkedStream, + SynthesizeStream, + TTS, + type TTSCapabilities, + type TTSError, +} from './tts.js'; /** * Internal status tracking for each TTS instance. @@ -90,6 +96,7 @@ export class FallbackAdapter extends TTS { private _status: TTSStatus[] = []; private _logger = log(); private _recoveryTimeouts: Map = new Map(); + private _errorSinks: { tts: TTS; listener: (error: TTSError) => void }[] = []; label: string = `tts.FallbackAdapter`; @@ -121,13 +128,18 @@ export class FallbackAdapter extends TTS { } private setupEventForwarding(): void { + // Child errors are handled by the fallback streams below. Keep a listener + // attached because Node treats listenerless `error` events as exceptions, + // but do not forward them: consumers could treat a provider failure as + // terminal before the adapter switches providers. If every provider fails, + // the adapter's own stream emits the terminal error instead. this.ttsInstances.forEach((tts) => { tts.on('metrics_collected', (metrics) => { this.emit('metrics_collected', metrics); }); - tts.on('error', (error) => { - this.emit('error', error); - }); + const errorSink: (error: TTSError) => void = () => {}; + this._errorSinks.push({ tts, listener: errorSink }); + tts.on('error', errorSink); }); } @@ -281,8 +293,11 @@ export class FallbackAdapter extends TTS { // Remove event listeners for (const tts of this.ttsInstances) { tts.removeAllListeners('metrics_collected'); - tts.removeAllListeners('error'); } + for (const { tts, listener } of this._errorSinks) { + tts.off('error', listener); + } + this._errorSinks = []; // Close all TTS instances await ThrowsPromise.all(this.ttsInstances.map((tts) => tts.close())); @@ -440,7 +455,6 @@ class FallbackSynthesizeStream extends SynthesizeStream { })(); for (let i = 0; i < this.adapter.ttsInstances.length; i++) { - const tts = this.adapter.getStreamingInstance(i); const originalTts = this.adapter.ttsInstances[i]!; const status = this.adapter.status[i]!; let lastRequestId: string = ''; @@ -450,15 +464,26 @@ class FallbackSynthesizeStream extends SynthesizeStream { this.adapter.markUnAvailable(i); continue; } - const resampler = this.adapter.createResamplerForTTS(i); + const tts = this.adapter.getStreamingInstance(i); + const isTransientStreamAdapter = tts !== originalTts; + const transientErrorSink: ((error: TTSError) => void) | undefined = isTransientStreamAdapter + ? () => {} + : undefined; + if (transientErrorSink) { + // StreamAdapter forwards errors from its wrapped TTS onto itself. + // Keep that transient emitter safe while fallback handles the failure. + tts.on('error', transientErrorSink); + } // ttfb measures the fallback adapter as a whole: anchor on the first // time a sentence was handed to any underlying TTS — even one that // failed before emitting audio — and never overwrite it when falling // back to another TTS (markStarted only takes effect once). let captureStartedTime: () => void = () => {}; + let resampler: AudioResampler | null = null; try { + resampler = this.adapter.createResamplerForTTS(i); this._logger.debug({ tts: originalTts.label }, 'attempting TTS stream'); const connOptions: APIConnectOptions = { @@ -611,6 +636,10 @@ class FallbackSynthesizeStream extends SynthesizeStream { // its started time must still anchor the fallback's ttfb captureStartedTime(); resampler?.close(); + if (transientErrorSink) { + tts.off('error', transientErrorSink); + await tts.close(); + } } } await readInputLLMStream.catch(() => {});