fix(tts): keep fallback child errors internal

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

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<string>({
+        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<string>({
+        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<string>({
+        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,

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<number, NodeJS.Timeout> = 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(() => {});