fix: add idle timeouts to TTS stream reads to prevent agent stuck in speaking state (#1174)

Author: toubatbrian

.changeset/gorgeous-seas-grin.md

@@ -0,0 +1,5 @@
+---
+"@livekit/agents": patch
+---
+
+fix: add idle timeouts to TTS stream reads to prevent agent stuck in speaking state

agents/src/inference/tts.ts

@@ -3,7 +3,7 @@
 // SPDX-License-Identifier: Apache-2.0
 import type { AudioFrame } from '@livekit/rtc-node';
 import { WebSocket } from 'ws';
-import { APIError, APIStatusError } from '../_exceptions.js';
+import { APIError, APIStatusError, APITimeoutError } from '../_exceptions.js';
 import { AudioByteStream } from '../audio.js';
 import { ConnectionPool } from '../connection_pool.js';
 import { type LanguageCode, normalizeLanguage } from '../language.js';
@@ -13,7 +13,15 @@ import { basic as tokenizeBasic } from '../tokenize/index.js';
 import type { ChunkedStream } from '../tts/index.js';
 import { SynthesizeStream as BaseSynthesizeStream, TTS as BaseTTS } from '../tts/index.js';
 import { type APIConnectOptions, DEFAULT_API_CONNECT_OPTIONS } from '../types.js';
-import { Event, Future, Task, cancelAndWait, combineSignals, shortuuid } from '../utils.js';
+import {
+  Event,
+  Future,
+  Task,
+  cancelAndWait,
+  combineSignals,
+  shortuuid,
+  waitUntilTimeout,
+} from '../utils.js';
 import {
   type TtsClientEvent,
   type TtsServerEvent,
@@ -578,6 +586,7 @@ export class SynthesizeStream<TModel extends TTSModels> extends BaseSynthesizeSt
 
     const createRecvTask = async (signal: AbortSignal) => {
       let currentSessionId: string | null = null;
+      const recvTimeoutMs = this.connOptions.timeoutMs;
 
       const bstream = new AudioByteStream(this.opts.sampleRate, NUM_CHANNELS);
       const serverEventStream = eventChannel.stream();
@@ -587,7 +596,12 @@ export class SynthesizeStream<TModel extends TTSModels> extends BaseSynthesizeSt
         await inputSentEvent.wait();
 
         while (!this.closed && !signal.aborted) {
-          const result = await reader.read();
+          const result = await waitUntilTimeout(
+            reader.read(),
+            recvTimeoutMs,
+            () => new APITimeoutError({ message: 'TTS recv idle timeout' }),
+          );
+
           if (signal.aborted) return;
           if (result.done) return;
 
@@ -632,6 +646,14 @@ export class SynthesizeStream<TModel extends TTSModels> extends BaseSynthesizeSt
               break;
           }
         }
+      } catch (e) {
+        if (e instanceof APITimeoutError) {
+          this.#logger.warn('TTS recv task timed out waiting for server message');
+          await resourceCleanup();
+          completionFuture.reject(e);
+          return;
+        }
+        throw e;
       } finally {
         reader.releaseLock();
         try {

agents/src/utils.ts

@@ -9,6 +9,7 @@ import type {
   TrackKind,
 } from '@livekit/rtc-node';
 import { AudioFrame, AudioResampler, RoomEvent } from '@livekit/rtc-node';
+import type { Throws } from '@livekit/throws-transformer/throws';
 import { AsyncLocalStorage } from 'node:async_hooks';
 import { EventEmitter, once } from 'node:events';
 import type { ReadableStream } from 'node:stream/web';
@@ -804,6 +805,33 @@ export function delay(ms: number, options: DelayOptions = {}): Promise<void> {
   });
 }
 
+export class IdleTimeoutError extends Error {
+  constructor(message = 'idle timeout') {
+    super(message);
+    this.name = 'IdleTimeoutError';
+  }
+}
+
+/**
+ * Race a promise against an idle timeout. If the promise does not settle within
+ * `timeoutMs` milliseconds, the returned promise rejects with {@link IdleTimeoutError}
+ * (or the error returned by `throwError` when provided).
+ * The timer is properly cleaned up on settlement to avoid leaking handles.
+ */
+export function waitUntilTimeout<T, E extends Error = IdleTimeoutError>(
+  promise: Promise<T>,
+  timeoutMs: number,
+  throwError?: () => E,
+): Promise<Throws<T, E>> {
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  return Promise.race([
+    promise,
+    new Promise<never>((_, reject) => {
+      timer = setTimeout(() => reject(throwError?.() ?? new IdleTimeoutError()), timeoutMs);
+    }),
+  ]).finally(() => clearTimeout(timer)) as Promise<Throws<T, E>>;
+}
+
 /**
  * Returns a participant that matches the given identity. If identity is None, the first
  * participant that joins the room will be returned.

agents/src/voice/generation.ts

@@ -25,7 +25,15 @@ import { log } from '../log.js';
 import { IdentityTransform } from '../stream/identity_transform.js';
 import { traceTypes, tracer } from '../telemetry/index.js';
 import { USERDATA_TIMED_TRANSCRIPT } from '../types.js';
-import { Future, Task, shortuuid, toError, waitForAbort } from '../utils.js';
+import {
+  Future,
+  IdleTimeoutError,
+  Task,
+  shortuuid,
+  toError,
+  waitForAbort,
+  waitUntilTimeout,
+} from '../utils.js';
 import {
   type Agent,
   type ModelSettings,
@@ -46,6 +54,8 @@ import {
 import { RunContext } from './run_context.js';
 import type { SpeechHandle } from './speech_handle.js';
 
+const TTS_READ_IDLE_TIMEOUT_MS = 10_000;
+
 /** @internal */
 export class _LLMGenerationData {
   generatedText: string = '';
@@ -550,6 +560,7 @@ export function performTTSInference(
   model?: string,
   provider?: string,
 ): [Task<void>, _TTSGenerationData] {
+  const logger = log();
   const audioStream = new IdentityTransform<AudioFrame>();
   const outputWriter = audioStream.writable.getWriter();
   const audioOutputStream = audioStream.readable;
@@ -624,12 +635,15 @@ export function performTTSInference(
       // JS currently only does single inference, so initialPushedDuration is always 0.
       // TODO: Add FlushSentinel + multi-segment loop
       const initialPushedDuration = pushedDuration;
-
       while (true) {
         if (signal.aborted) {
           break;
         }
-        const { done, value: frame } = await ttsStreamReader.read();
+
+        const { done, value: frame } = await waitUntilTimeout(
+          ttsStreamReader.read(),
+          TTS_READ_IDLE_TIMEOUT_MS,
+        );
         if (done) {
           break;
         }
@@ -671,14 +685,15 @@ export function performTTSInference(
         pushedDuration += frameDuration;
       }
     } catch (error) {
-      if (error instanceof DOMException && error.name === 'AbortError') {
-        // Abort signal was triggered, handle gracefully
+      if (error instanceof IdleTimeoutError) {
+        logger.warn('TTS stream stalled after producing audio, forcing close');
+      } else if (error instanceof DOMException && error.name === 'AbortError') {
         return;
+      } else {
+        throw error;
       }
-      throw error;
     } finally {
       if (!timedTextsFut.done) {
-        // Ensure downstream consumers don't hang on errors.
         timedTextsFut.resolve(null);
       }
       ttsStreamReader?.releaseLock();
@@ -773,9 +788,12 @@ async function forwardAudio(
   out: _AudioOut,
   signal?: AbortSignal,
 ): Promise<void> {
+  const logger = log();
   const reader = ttsStream.getReader();
   let resampler: AudioResampler | null = null;
 
+  const FORWARD_AUDIO_IDLE_TIMEOUT_MS = 10_000;
+
   const onPlaybackStarted = (ev: { createdAt: number }) => {
     if (!out.firstFrameFut.done) {
       out.firstFrameFut.resolve(ev.createdAt);
@@ -791,7 +809,10 @@ async function forwardAudio(
         break;
       }
 
-      const { done, value: frame } = await reader.read();
+      const { done, value: frame } = await waitUntilTimeout(
+        reader.read(),
+        FORWARD_AUDIO_IDLE_TIMEOUT_MS,
+      );
       if (done) break;
 
       out.audio.push(frame);
@@ -819,6 +840,12 @@ async function forwardAudio(
         await audioOutput.captureFrame(f);
       }
     }
+  } catch (e) {
+    if (e instanceof IdleTimeoutError) {
+      logger.warn('audio forwarding stalled waiting for TTS frames, forcing close');
+    } else {
+      throw e;
+    }
   } finally {
     audioOutput.off(AudioOutput.EVENT_PLAYBACK_STARTED, onPlaybackStarted);
 

agents/src/voice/generation_tts_timeout.test.ts

@@ -0,0 +1,112 @@
+// SPDX-FileCopyrightText: 2026 LiveKit, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+import { AudioFrame } from '@livekit/rtc-node';
+import { ReadableStream } from 'stream/web';
+import { describe, expect, it, vi } from 'vitest';
+import { initializeLogger } from '../log.js';
+import { performAudioForwarding, performTTSInference } from './generation.js';
+import { AudioOutput } from './io.js';
+
+function createSilentFrame(sampleRate = 24000, channels = 1, durationMs = 20): AudioFrame {
+  const samplesPerChannel = Math.floor((sampleRate * durationMs) / 1000);
+  const data = new Int16Array(samplesPerChannel * channels);
+  return new AudioFrame(data, sampleRate, channels, samplesPerChannel);
+}
+
+class MockAudioOutput extends AudioOutput {
+  capturedFrames: AudioFrame[] = [];
+
+  constructor() {
+    super(24000);
+  }
+
+  async captureFrame(frame: AudioFrame): Promise<void> {
+    await super.captureFrame(frame);
+    this.capturedFrames.push(frame);
+    this.onPlaybackStarted(Date.now());
+  }
+
+  clearBuffer(): void {
+    // no-op for mock
+  }
+}
+
+describe('TTS stream idle timeout', () => {
+  initializeLogger({ pretty: false, level: 'silent' });
+
+  it('forwardAudio completes when TTS stream stalls after producing frames', async () => {
+    const stalledStream = new ReadableStream<AudioFrame>({
+      start(controller) {
+        controller.enqueue(createSilentFrame());
+        controller.enqueue(createSilentFrame());
+      },
+    });
+
+    const audioOutput = new MockAudioOutput();
+    const controller = new AbortController();
+
+    const [task, audioOut] = performAudioForwarding(stalledStream, audioOutput, controller);
+
+    vi.useFakeTimers();
+
+    const taskPromise = task.result;
+    await vi.advanceTimersByTimeAsync(11_000);
+    await taskPromise;
+
+    vi.useRealTimers();
+
+    expect(audioOutput.capturedFrames.length).toBe(2);
+    expect(audioOut.firstFrameFut.done).toBe(true);
+  }, 10_000);
+
+  it('forwardAudio completes normally when TTS stream closes properly', async () => {
+    const normalStream = new ReadableStream<AudioFrame>({
+      start(controller) {
+        controller.enqueue(createSilentFrame());
+        controller.enqueue(createSilentFrame());
+        controller.enqueue(createSilentFrame());
+        controller.close();
+      },
+    });
+
+    const audioOutput = new MockAudioOutput();
+    const controller = new AbortController();
+
+    const [task, audioOut] = performAudioForwarding(normalStream, audioOutput, controller);
+
+    await task.result;
+
+    expect(audioOutput.capturedFrames.length).toBe(3);
+    expect(audioOut.firstFrameFut.done).toBe(true);
+  });
+
+  it('performTTSInference completes when TTS node returns stalled stream', async () => {
+    const stalledTtsStream = new ReadableStream<AudioFrame>({
+      start(controller) {
+        controller.enqueue(createSilentFrame());
+      },
+    });
+
+    const ttsNode = async () => stalledTtsStream;
+    const textInput = new ReadableStream<string>({
+      start(controller) {
+        controller.enqueue('Hello world');
+        controller.close();
+      },
+    });
+
+    const controller = new AbortController();
+    const [task, genData] = performTTSInference(ttsNode, textInput, {}, controller);
+
+    vi.useFakeTimers();
+
+    const taskPromise = task.result;
+    await vi.advanceTimersByTimeAsync(11_000);
+    await taskPromise;
+
+    vi.useRealTimers();
+
+    expect(genData.ttfb).toBeDefined();
+  }, 10_000);
+});