diff --git a/.changeset/close-speech-stream-adapters.md b/.changeset/close-speech-stream-adapters.md new file mode 100644 index 000000000..7a3e2a3a1 --- /dev/null +++ b/.changeset/close-speech-stream-adapters.md @@ -0,0 +1,5 @@ +--- +'@livekit/agents': patch +--- + +Close temporary speech stream adapters when their owning pipeline or fallback lifecycle ends. diff --git a/agents/src/stt/fallback_adapter.test.ts b/agents/src/stt/fallback_adapter.test.ts index db616664b..38f992428 100644 --- a/agents/src/stt/fallback_adapter.test.ts +++ b/agents/src/stt/fallback_adapter.test.ts @@ -7,6 +7,7 @@ import { APIConnectionError, APIError } from '../_exceptions.js'; import { initializeLogger } from '../log.js'; import type { APIConnectOptions } from '../types.js'; import { type AudioBuffer, delay } from '../utils.js'; +import { VAD, VADStream } from '../vad.js'; import { FallbackAdapter } from './fallback_adapter.js'; import { STT, type SpeechEvent, SpeechEventType, SpeechStream } from './stt.js'; import { FakeSTT, RecognizeSentinel, emptyAudioFrame } from './testing/fake_stt.js'; @@ -82,6 +83,30 @@ class RetryTimelineStream extends SpeechStream { } } +class FakeVAD extends VAD { + label = 'fake-vad'; + + constructor() { + super({ updateInterval: 100 }); + } + + stream(): VADStream { + return new (class extends VADStream {})(this); + } +} + +class NonStreamingSTT extends FakeSTT { + closeCount = 0; + + constructor() { + super({ capabilities: { streaming: false, interimResults: false } }); + } + + override async close(): Promise { + this.closeCount++; + } +} + describe('FallbackAdapter', () => { beforeAll(() => { initializeLogger({ pretty: false }); @@ -281,6 +306,19 @@ describe('FallbackAdapter', () => { expect(received).toHaveLength(0); }); + + it('close closes automatically created stream adapters', async () => { + const stt = new NonStreamingSTT(); + const baseline = stt.listenerCount('metrics_collected'); + const adapter = new FallbackAdapter({ sttInstances: [stt], vad: new FakeVAD() }); + + expect(stt.listenerCount('metrics_collected')).toBe(baseline + 1); + + await adapter.close(); + + expect(stt.listenerCount('metrics_collected')).toBe(baseline); + expect(stt.closeCount).toBe(0); + }); }); describe('FallbackSpeechStream (streaming path)', () => { diff --git a/agents/src/stt/fallback_adapter.ts b/agents/src/stt/fallback_adapter.ts index f733545bb..06265a5eb 100644 --- a/agents/src/stt/fallback_adapter.ts +++ b/agents/src/stt/fallback_adapter.ts @@ -98,6 +98,7 @@ export class FallbackAdapter extends STT { private _status: STTStatus[] = []; private _logger = log(); private _metricsForwarders = new Map void>(); + private _ownedStreamAdapters: StreamAdapter[] = []; // Last child that produced output or returned a recognize result. Surfaced // via the dynamic label/model/provider getters so OTel attributes like // `gen_ai.request.model` on `user_turn` (refreshed on every STT event by @@ -124,9 +125,13 @@ export class FallbackAdapter extends STT { ); } - const wrapped = opts.sttInstances.map((s) => - s.capabilities.streaming ? s : new StreamAdapter(s, opts.vad!), - ); + const ownedStreamAdapters: StreamAdapter[] = []; + const wrapped = opts.sttInstances.map((s) => { + if (s.capabilities.streaming) return s; + const adapter = new StreamAdapter(s, opts.vad!); + ownedStreamAdapters.push(adapter); + return adapter; + }); // Pick the primary's granularity only if every instance supports aligned // transcripts — otherwise consumers can't rely on a consistent format. @@ -145,6 +150,7 @@ export class FallbackAdapter extends STT { }); this.sttInstances = wrapped; + this._ownedStreamAdapters = ownedStreamAdapters; this.attemptTimeoutMs = opts.attemptTimeoutMs ?? 10_000; this.maxRetryPerSTT = opts.maxRetryPerSTT ?? 1; this.retryIntervalMs = opts.retryIntervalMs ?? 5_000; @@ -327,6 +333,7 @@ export class FallbackAdapter extends STT { if (m) s.off('metrics_collected' as keyof STTCallbacks, m); } this._metricsForwarders.clear(); + await Promise.all(this._ownedStreamAdapters.map((adapter) => adapter.close())); } } diff --git a/agents/src/stt/stream_adapter.ts b/agents/src/stt/stream_adapter.ts index 535029024..48c7381f7 100644 --- a/agents/src/stt/stream_adapter.ts +++ b/agents/src/stt/stream_adapter.ts @@ -4,12 +4,13 @@ import type { AudioFrame } from '@livekit/rtc-node'; import { ThrowsPromise } from '@livekit/throws-transformer/throws'; import { log } from '../log.js'; +import type { STTMetrics } from '../metrics/base.js'; import type { APIConnectOptions } from '../types.js'; import { isStreamClosedError } from '../utils.js'; import type { VAD, VADStream } from '../vad.js'; import { VADEventType } from '../vad.js'; import type { ConversationItemAddedEvent } from '../voice/events.js'; -import type { SpeechEvent } from './stt.js'; +import type { STTError, SpeechEvent } from './stt.js'; import { STT, SpeechEventType, SpeechStream } from './stt.js'; export class StreamAdapter extends STT { @@ -17,6 +18,14 @@ export class StreamAdapter extends STT { #vad: VAD; label: string; + #forwardMetrics = (metrics: STTMetrics) => { + this.emit('metrics_collected', metrics); + }; + + #forwardError = (error: STTError) => { + this.emit('error', error); + }; + constructor(stt: STT, vad: VAD) { super({ streaming: true, @@ -28,13 +37,14 @@ export class StreamAdapter extends STT { this.#vad = vad; this.label = `stt.StreamAdapter<${this.#stt.label}>`; - this.#stt.on('metrics_collected', (metrics) => { - this.emit('metrics_collected', metrics); - }); + this.#stt.on('metrics_collected', this.#forwardMetrics); + this.#stt.on('error', this.#forwardError); + } - this.#stt.on('error', (error) => { - this.emit('error', error); - }); + async close(): Promise { + this.#stt.off('metrics_collected', this.#forwardMetrics); + this.#stt.off('error', this.#forwardError); + await super.close(); } _recognize(frame: AudioFrame, abortSignal?: AbortSignal): Promise { diff --git a/agents/src/tts/fallback_adapter.test.ts b/agents/src/tts/fallback_adapter.test.ts index d7d0287d6..fe70f737d 100644 --- a/agents/src/tts/fallback_adapter.test.ts +++ b/agents/src/tts/fallback_adapter.test.ts @@ -19,6 +19,7 @@ class MockSynthesizeStream extends SynthesizeStream { constructor( private mockTts: MockTTS, private shouldFail: boolean, + private blocked: boolean, connOptions?: APIConnectOptions, ) { super(mockTts, connOptions); @@ -36,6 +37,13 @@ class MockSynthesizeStream extends SynthesizeStream { } protected async run(): Promise { + if (this.blocked) { + if (this.abortSignal.aborted) return; + await new Promise((resolve) => { + this.abortSignal.addEventListener('abort', () => resolve(), { once: true }); + }); + return; + } if (this.shouldFail) { if (this.mockTts.failAfterInput) { // Simulate a provider that receives text but dies before emitting @@ -78,11 +86,19 @@ class MockChunkedStream extends ChunkedStream { private mockTts: MockTTS, text: string, private shouldFail: boolean, + private blocked: boolean, connOptions?: APIConnectOptions, ) { super(text, mockTts, connOptions); } protected async run(): Promise { + if (this.blocked) { + if (this.abortSignal.aborted) return; + await new Promise((resolve) => { + this.abortSignal.addEventListener('abort', () => resolve(), { once: true }); + }); + return; + } if (this.shouldFail) { throw new APIError('mock TTS failed immediately'); } @@ -98,6 +114,7 @@ class MockChunkedStream extends ChunkedStream { class MockTTS extends TTS { label: string; shouldFail = false; + blocked = false; /** When failing, first consume a token (and mark started) before throwing. */ failAfterInput = false; /** Simulated latency between receiving text and sending it to the provider. */ @@ -105,18 +122,42 @@ 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 }); + closeCount = 0; + + constructor(label: string, sampleRate: number = SAMPLE_RATE, streaming = true) { + super(sampleRate, 1, { streaming }); this.label = label; } synthesize(text: string, connOptions?: APIConnectOptions): ChunkedStream { - return new MockChunkedStream(this, text, this.shouldFail, connOptions); + return new MockChunkedStream(this, text, this.shouldFail, this.blocked, connOptions); } stream(options?: { connOptions?: APIConnectOptions }): SynthesizeStream { - return new MockSynthesizeStream(this, this.shouldFail, options?.connOptions); + return new MockSynthesizeStream(this, this.shouldFail, this.blocked, options?.connOptions); + } + + override async close(): Promise { + this.closeCount++; + } +} + +function textInput(text = 'hello test'): ReadableStream { + return new ReadableStream({ + start(controller) { + controller.enqueue(text); + controller.close(); + }, + }); +} + +async function consume(stream: SynthesizeStream): Promise { + stream.updateInputStream(textInput()); + let frames = 0; + for await (const event of stream) { + if (event !== SynthesizeStream.END_OF_STREAM) frames++; } + return frames; } describe('TTS FallbackAdapter', () => { @@ -126,6 +167,74 @@ describe('TTS FallbackAdapter', () => { process.on('unhandledRejection', () => {}); }); + it('closes temporary stream adapters after each request', async () => { + const nonStreaming = new MockTTS('non-streaming', SAMPLE_RATE, false); + const adapter = new FallbackAdapter({ ttsInstances: [nonStreaming] }); + const baseline = nonStreaming.listenerCount('metrics_collected'); + + try { + for (let i = 0; i < 3; i++) { + expect(await consume(adapter.stream())).toBeGreaterThan(0); + expect(nonStreaming.listenerCount('metrics_collected')).toBe(baseline); + } + expect(nonStreaming.closeCount).toBe(0); + } finally { + await adapter.close(); + } + }); + + it('closes a temporary stream adapter after failure and fallback', async () => { + const nonStreaming = new MockTTS('non-streaming', SAMPLE_RATE, false); + nonStreaming.shouldFail = true; + const adapter = new FallbackAdapter({ + ttsInstances: [nonStreaming, new MockTTS('fallback')], + maxRetryPerTTS: 0, + recoveryDelayMs: 60_000, + }); + const baseline = nonStreaming.listenerCount('metrics_collected'); + + try { + expect(await consume(adapter.stream())).toBeGreaterThan(0); + expect(nonStreaming.listenerCount('metrics_collected')).toBe(baseline); + expect(nonStreaming.closeCount).toBe(0); + } finally { + await adapter.close(); + } + }); + + it('closes a temporary stream adapter after cancellation', async () => { + const nonStreaming = new MockTTS('non-streaming', SAMPLE_RATE, false); + nonStreaming.blocked = true; + const adapter = new FallbackAdapter({ ttsInstances: [nonStreaming] }); + const baseline = nonStreaming.listenerCount('metrics_collected'); + const stream = adapter.stream(); + stream.updateInputStream(textInput()); + + try { + const deadline = Date.now() + 1_000; + while ( + nonStreaming.listenerCount('metrics_collected') === baseline && + Date.now() < deadline + ) { + await new Promise((resolve) => setTimeout(resolve, 5)); + } + expect(nonStreaming.listenerCount('metrics_collected')).toBeGreaterThan(baseline); + + stream.close(); + const closeDeadline = Date.now() + 1_000; + while ( + nonStreaming.listenerCount('metrics_collected') > baseline && + Date.now() < closeDeadline + ) { + await new Promise((resolve) => setTimeout(resolve, 5)); + } + expect(nonStreaming.listenerCount('metrics_collected')).toBe(baseline); + } finally { + stream.close(); + await adapter.close(); + } + }); + it('should fall back to the next TTS when the primary stream fails before any pushText', async () => { const primary = new MockTTS('primary'); primary.shouldFail = true; diff --git a/agents/src/tts/fallback_adapter.ts b/agents/src/tts/fallback_adapter.ts index 903231bd7..f07997653 100644 --- a/agents/src/tts/fallback_adapter.ts +++ b/agents/src/tts/fallback_adapter.ts @@ -440,7 +440,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,7 +449,11 @@ class FallbackSynthesizeStream extends SynthesizeStream { this.adapter.markUnAvailable(i); continue; } - const resampler = this.adapter.createResamplerForTTS(i); + const tts = this.adapter.getStreamingInstance(i); + let stream!: SynthesizeStream; + let resampler: AudioResampler | null = null; + const closeStream = () => stream?.close(); + this.abortSignal.addEventListener('abort', closeStream, { once: true }); // ttfb measures the fallback adapter as a whole: anchor on the first // time a sentence was handed to any underlying TTS — even one that @@ -459,6 +462,7 @@ class FallbackSynthesizeStream extends SynthesizeStream { let captureStartedTime: () => void = () => {}; try { + resampler = this.adapter.createResamplerForTTS(i); this._logger.debug({ tts: originalTts.label }, 'attempting TTS stream'); const connOptions: APIConnectOptions = { @@ -466,7 +470,8 @@ class FallbackSynthesizeStream extends SynthesizeStream { maxRetry: this.adapter.maxRetryPerTTS, }; - const stream = tts.stream({ connOptions }); + stream = tts.stream({ connOptions }); + if (this.abortSignal.aborted) stream.close(); let bufferIndex = 0; let streamOutputCompleted = false; @@ -610,7 +615,12 @@ class FallbackSynthesizeStream extends SynthesizeStream { // the stream may have received text and failed before emitting audio; // its started time must still anchor the fallback's ttfb captureStartedTime(); + this.abortSignal.removeEventListener('abort', closeStream); + stream?.close(); resampler?.close(); + if (tts !== originalTts) { + await tts.close(); + } } } await readInputLLMStream.catch(() => {}); diff --git a/agents/src/voice/agent.test.ts b/agents/src/voice/agent.test.ts index 50973e58b..bc291a7e3 100644 --- a/agents/src/voice/agent.test.ts +++ b/agents/src/voice/agent.test.ts @@ -1,14 +1,22 @@ // SPDX-FileCopyrightText: 2025 LiveKit, Inc. // // SPDX-License-Identifier: Apache-2.0 -import type { AudioFrame } from '@livekit/rtc-node'; +import { AudioFrame } from '@livekit/rtc-node'; import { ReadableStream } from 'node:stream/web'; import { describe, expect, it, vi } from 'vitest'; import { z } from 'zod'; +import { APIConnectionError } from '../_exceptions.js'; import { ChatContext, ChatMessage, ToolError, tool } from '../llm/index.js'; import { initializeLogger } from '../log.js'; -import { SynthesizeStream } from '../tts/index.js'; +import { FakeSTT } from '../stt/testing/fake_stt.js'; +import { + type ChunkedStream, + SynthesizeStream, + TTS, + StreamAdapter as TTSStreamAdapter, +} from '../tts/index.js'; import { Task } from '../utils.js'; +import { VAD, VADStream } from '../vad.js'; import { Agent, AgentTask, _setActivityTaskInfo } from './agent.js'; import { AgentActivity, agentActivityStorage } from './agent_activity.js'; import { AgentSession } from './agent_session.js'; @@ -48,6 +56,61 @@ class TimeoutError extends Error { } } +class FakeVAD extends VAD { + label = 'fake-vad'; + + constructor() { + super({ updateInterval: 100 }); + } + + stream(): VADStream { + return new (class extends VADStream {})(this); + } +} + +class NonStreamingSTT extends FakeSTT { + closeCount = 0; + + constructor() { + super({ capabilities: { streaming: false, interimResults: false } }); + } + + override async close(): Promise { + this.closeCount++; + } +} + +class NonStreamingTTS extends TTS { + label = 'non-streaming-tts'; + closeCount = 0; + + constructor() { + super(24_000, 1, { streaming: false }); + } + + synthesize(): ChunkedStream { + throw new Error('not used by the node lifecycle test'); + } + + stream(): SynthesizeStream { + throw new Error('non-streaming test TTS'); + } + + override async close(): Promise { + this.closeCount++; + } +} + +function bindTestActivity(agent: Agent, models: { stt?: FakeSTT; tts?: TTS; vad?: VAD }): void { + (agent as any)._agentActivity = { + ...models, + agentSession: { + connOptions: { sttConnOptions: {}, ttsConnOptions: {} }, + }, + _resolveExpressiveOptions: () => undefined, + }; +} + async function closeWithTimeout(session: AgentSession): Promise { try { await withTimeout(session.close(), 30_000); @@ -465,6 +528,84 @@ describe('Agent', () => { }); }); + describe('temporary speech stream adapters', () => { + it('closes the STT adapter when the node is cancelled', async () => { + const stt = new NonStreamingSTT(); + const baseline = stt.listenerCount('metrics_collected'); + const agent = new Agent({ instructions: 'test' }); + bindTestActivity(agent, { stt, vad: new FakeVAD() }); + const input = new ReadableStream(); + + const output = await Agent.default.sttNode(agent, input, {}); + expect(output).not.toBeNull(); + expect(stt.listenerCount('metrics_collected')).toBeGreaterThan(baseline); + + await output!.cancel(); + + expect(stt.listenerCount('metrics_collected')).toBe(baseline); + expect(stt.closeCount).toBe(0); + }); + + it.each(['success', 'failure', 'blocked'] as const)( + 'closes the TTS adapter after %s', + async (mode) => { + const tts = new NonStreamingTTS(); + tts.on('error', () => {}); + const baseline = tts.listenerCount('metrics_collected'); + const agent = new Agent({ instructions: 'test' }); + bindTestActivity(agent, { tts }); + const frame = new AudioFrame(new Int16Array(160), 24_000, 1, 160); + let unblock!: () => void; + const blocked = new Promise((resolve) => { + unblock = resolve; + }); + const adaptedStream = { + updateInputStream() {}, + close: unblock, + async *[Symbol.asyncIterator]() { + if (mode === 'failure') { + throw new APIConnectionError({ message: 'probe failure' }); + } + if (mode === 'blocked') { + await blocked; + return; + } + yield { frame, timedTranscripts: [] }; + yield SynthesizeStream.END_OF_STREAM; + }, + } as unknown as SynthesizeStream; + const streamSpy = vi + .spyOn(TTSStreamAdapter.prototype, 'stream') + .mockReturnValue(adaptedStream); + const input = new ReadableStream({ + start(controller) { + controller.enqueue('Hello world, this is a complete sentence.'); + controller.close(); + }, + }); + + try { + const output = await Agent.default.ttsNode(agent, input, {}); + expect(output).not.toBeNull(); + + if (mode === 'success') { + expect(await collectReadableStream(output!)).toHaveLength(1); + } else if (mode === 'failure') { + await expect(collectReadableStream(output!)).rejects.toThrow('probe failure'); + } else { + await output!.cancel(); + } + + expect(tts.listenerCount('metrics_collected')).toBe(baseline); + expect(tts.closeCount).toBe(0); + } finally { + unblock(); + streamSpy.mockRestore(); + } + }, + ); + }); + it('should require AgentTask to run inside task context', async () => { class TestTask extends AgentTask { constructor() { diff --git a/agents/src/voice/agent.ts b/agents/src/voice/agent.ts index 2fc5127ed..29aeeb36f 100644 --- a/agents/src/voice/agent.ts +++ b/agents/src/voice/agent.ts @@ -477,6 +477,7 @@ export class Agent { } let wrappedStt = activity.stt; + let temporaryAdapter: STTStreamAdapter | undefined; if (!wrappedStt.capabilities.streaming) { const vad = agent.vad || activity.vad; @@ -485,30 +486,40 @@ export class Agent { 'STT does not support streaming, add a VAD to the AgentTask/VoiceAgent to enable streaming', ); } - wrappedStt = new STTStreamAdapter(wrappedStt, vad); + temporaryAdapter = new STTStreamAdapter(wrappedStt, vad); + wrappedStt = temporaryAdapter; } const connOptions = activity.agentSession.connOptions.sttConnOptions; - const stream = wrappedStt.stream({ connOptions }); - - // Set startTimeOffset to provide linear timestamps across reconnections - const audioInputStartedAt = - activity.inputStartedAt ?? // Use input started at proxied from AudioRecognition if available - activity.agentSession._recorderIO?.recordingStartedAt ?? // Fallback to recording start time if available - activity.agentSession._startedAt ?? // Fallback to session start time - Date.now(); // Fallback to current time - - stream.startTimeOffset = (Date.now() - audioInputStartedAt) / 1000; - - stream.updateInputStream(input); - + let stream!: ReturnType; let cleaned = false; - const cleanup = () => { + const cleanup = async () => { if (cleaned) return; cleaned = true; - stream.detachInputStream(); - stream.close(); + try { + stream?.detachInputStream(); + stream?.close(); + } finally { + await temporaryAdapter?.close(); + } }; + try { + stream = wrappedStt.stream({ connOptions }); + + // Set startTimeOffset to provide linear timestamps across reconnections + const audioInputStartedAt = + activity.inputStartedAt ?? // Use input started at proxied from AudioRecognition if available + activity.agentSession._recorderIO?.recordingStartedAt ?? // Fallback to recording start time if available + activity.agentSession._startedAt ?? // Fallback to session start time + Date.now(); // Fallback to current time + + stream.startTimeOffset = (Date.now() - audioInputStartedAt) / 1000; + + stream.updateInputStream(input); + } catch (error) { + await cleanup(); + throw error; + } return new ReadableStream({ async start(controller) { @@ -516,14 +527,14 @@ export class Agent { for await (const event of stream) { controller.enqueue(event); } - controller.close(); } finally { // Always clean up the STT stream, whether it ends naturally or is cancelled - cleanup(); + await cleanup(); } + controller.close(); }, cancel() { - cleanup(); + return cleanup(); }, }); }, @@ -613,19 +624,30 @@ export class Agent { activity.tts._setExpressive(expressiveActive); const connOptions = activity.agentSession.connOptions.ttsConnOptions; - const stream = wrappedTts.stream({ connOptions }); - stream.updateInputStream(input); - + let stream!: SynthesizeStream; let cleaned = false; const cleanup = async () => { if (cleaned) return; cleaned = true; - stream.close(); - await input.cancel('tts node cleanup').catch(() => {}); - if (wrappedTts !== activity.tts) { - await wrappedTts.close(); + try { + stream?.close(); + } finally { + try { + if (wrappedTts !== activity.tts) { + await wrappedTts.close(); + } + } finally { + await input.cancel('tts node cleanup').catch(() => {}); + } } }; + try { + stream = wrappedTts.stream({ connOptions }); + stream.updateInputStream(input); + } catch (error) { + await cleanup(); + throw error; + } return new ReadableStream({ async start(controller) { @@ -640,10 +662,10 @@ export class Agent { } controller.enqueue(chunk.frame); } - controller.close(); } finally { await cleanup(); } + controller.close(); }, cancel() { return cleanup();