fix(agents): prevent inference STT frame retention

.changeset/quiet-streams-release.md

@@ -0,0 +1,5 @@
+---
+'@livekit/agents': patch
+---
+
+Prevent inference STT streams from retaining previously consumed audio frames while waiting for cancellation.

agents/src/inference/stt.test.ts

@@ -1,7 +1,7 @@
 // SPDX-FileCopyrightText: 2025 LiveKit, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
-import { beforeAll, describe, expect, it } from 'vitest';
+import { beforeAll, describe, expect, it, vi } from 'vitest';
 import * as agents from '../index.js';
 import { normalizeLanguage } from '../language.js';
 import { initializeLogger } from '../log.js';
@@ -129,6 +129,95 @@ describe('Inference STT start of speech', () => {
   });
 });
 
+describe('Inference STT abortable iteration', () => {
+  it('removes each abort listener after consuming an input', async () => {
+    const { stream } = makeSpeechStream();
+    const controller = new AbortController();
+    const addListener = vi.spyOn(controller.signal, 'addEventListener');
+    const removeListener = vi.spyOn(controller.signal, 'removeEventListener');
+    let value = 0;
+    const iterator: AsyncIterator<number> = {
+      next: async () => ({ done: false, value: value++ }),
+    };
+
+    for (let i = 0; i < 50; i++) {
+      await expect(stream['nextUntilAborted'](iterator, controller.signal)).resolves.toEqual({
+        done: false,
+        value: i,
+      });
+    }
+
+    expect(addListener).toHaveBeenCalledTimes(50);
+    expect(removeListener).toHaveBeenCalledTimes(50);
+  });
+
+  it('stops a pending iteration and consumes its later rejection after abort', async () => {
+    const { stream } = makeSpeechStream();
+    const controller = new AbortController();
+    const removeListener = vi.spyOn(controller.signal, 'removeEventListener');
+    let rejectNext!: (error: Error) => void;
+    const iterator: AsyncIterator<number> = {
+      next: () =>
+        new Promise<IteratorResult<number>>((_, reject) => {
+          rejectNext = reject;
+        }),
+    };
+
+    const next = stream['nextUntilAborted'](iterator, controller.signal);
+    controller.abort();
+
+    await expect(next).resolves.toBeUndefined();
+    expect(removeListener).toHaveBeenCalledTimes(1);
+
+    // The abandoned iterator may still settle after cancellation; its rejection must stay handled.
+    rejectNext(new Error('late iterator failure'));
+    await Promise.resolve();
+  });
+
+  it('propagates an iterator rejection and removes its abort listener', async () => {
+    const { stream } = makeSpeechStream();
+    const controller = new AbortController();
+    const removeListener = vi.spyOn(controller.signal, 'removeEventListener');
+    const iterator: AsyncIterator<number> = {
+      next: async () => {
+        throw new Error('input failed');
+      },
+    };
+
+    await expect(stream['nextUntilAborted'](iterator, controller.signal)).rejects.toThrow(
+      'input failed',
+    );
+    expect(removeListener).toHaveBeenCalledTimes(1);
+  });
+
+  it.skipIf(typeof globalThis.gc !== 'function')(
+    'releases consumed inputs while the abort signal remains pending',
+    async () => {
+      const { stream } = makeSpeechStream();
+      const controller = new AbortController();
+      const refs: WeakRef<object>[] = [];
+      const iterator: AsyncIterator<object> = {
+        next: async () => {
+          const value = { payload: new Uint8Array(1024) };
+          refs.push(new WeakRef(value));
+          return { done: false, value };
+        },
+      };
+
+      for (let i = 0; i < 20; i++) {
+        await stream['nextUntilAborted'](iterator, controller.signal);
+      }
+
+      globalThis.gc!();
+      await new Promise<void>((resolve) => setTimeout(resolve, 0));
+      globalThis.gc!();
+
+      expect(refs.filter((ref) => ref.deref() !== undefined).length).toBeLessThanOrEqual(1);
+      controller.abort();
+    },
+  );
+});
+
 describe('parseSTTModelString', () => {
   it('simple model without language', () => {
     const [model, language] = parseSTTModelString('deepgram');

agents/src/inference/stt.ts

@@ -796,6 +796,58 @@ export class SpeechStream<TModel extends STTModels> extends BaseSpeechStream {
     }
   }
 
+  private nextUntilAborted<T>(
+    iterator: AsyncIterator<T>,
+    signal: AbortSignal,
+  ): Promise<IteratorResult<T> | undefined> {
+    if (signal.aborted) return Promise.resolve(undefined);
+
+    // A shared abort promise used in Promise.race accumulates one reaction per input and keeps
+    // each settled iterator result reachable until the stream closes. Keep cancellation scoped
+    // to the current read and remove its listener as soon as either side settles.
+    return new Promise<IteratorResult<T> | undefined>((resolve, reject) => {
+      let settled = false;
+      const cleanup = () => signal.removeEventListener('abort', onAbort);
+      const onAbort = () => {
+        if (settled) return;
+        settled = true;
+        cleanup();
+        resolve(undefined);
+      };
+
+      signal.addEventListener('abort', onAbort, { once: true });
+      if (signal.aborted) {
+        onAbort();
+        return;
+      }
+
+      let next: PromiseLike<IteratorResult<T>>;
+      try {
+        next = iterator.next();
+      } catch (error) {
+        settled = true;
+        cleanup();
+        reject(error);
+        return;
+      }
+
+      void next.then(
+        (result) => {
+          if (settled) return;
+          settled = true;
+          cleanup();
+          resolve(result);
+        },
+        (error: unknown) => {
+          if (settled) return;
+          settled = true;
+          cleanup();
+          reject(error);
+        },
+      );
+    });
+  }
+
   protected async run(): Promise<void> {
     while (true) {
       const vad = await this.stt.vadPromise;
@@ -858,74 +910,44 @@ export class SpeechStream<TModel extends STTModels> extends BaseSpeechStream {
           Math.floor(this.opts.sampleRate / 20), // 50ms
         );
 
-        // Create abort promise once to avoid memory leak
-        const abortPromise = new ThrowsPromise<never, Error>((_, reject) => {
-          if (signal.aborted) {
-            return reject(new Error('Send aborted'));
-          }
-          const onAbort = () => reject(new Error('Send aborted'));
-          signal.addEventListener('abort', onAbort, { once: true });
-        });
-
         // Manual iteration to support cancellation
         const iterator = this.input[Symbol.asyncIterator]();
-        try {
-          while (true) {
-            const result = await ThrowsPromise.race([iterator.next(), abortPromise]);
-
-            if (result.done) break;
-            const ev = result.value;
-
-            let frames: AudioFrame[];
-            if (ev === SpeechStream.FLUSH_SENTINEL) {
-              frames = audioStream.flush();
-            } else {
-              const frame = ev as AudioFrame;
-              vadStream?.pushFrame(frame);
-              frames = audioStream.write(new Int16Array(frame.data).buffer);
-            }
-
-            for (const frame of frames) {
-              this.speechDuration += frame.samplesPerChannel / frame.sampleRate;
-              const base64 = Buffer.from(frame.data.buffer).toString('base64');
-              const msg = { type: 'input_audio', audio: base64 };
-              socket.send(JSON.stringify(msg));
-            }
+        while (true) {
+          const result = await this.nextUntilAborted(iterator, signal);
+          if (result === undefined) return;
+          if (result.done) break;
+          const ev = result.value;
+
+          let frames: AudioFrame[];
+          if (ev === SpeechStream.FLUSH_SENTINEL) {
+            frames = audioStream.flush();
+          } else {
+            const frame = ev as AudioFrame;
+            vadStream?.pushFrame(frame);
+            frames = audioStream.write(new Int16Array(frame.data).buffer);
           }
 
-          closing = true;
-          vadStream?.endInput();
-          socket.send(JSON.stringify({ type: 'session.finalize' }));
-        } catch (e) {
-          if ((e as Error).message === 'Send aborted') {
-            // Expected abort, don't log
-            return;
+          for (const frame of frames) {
+            this.speechDuration += frame.samplesPerChannel / frame.sampleRate;
+            const base64 = Buffer.from(frame.data.buffer).toString('base64');
+            const msg = { type: 'input_audio', audio: base64 };
+            socket.send(JSON.stringify(msg));
           }
-          throw e;
         }
+
+        closing = true;
+        vadStream?.endInput();
+        socket.send(JSON.stringify({ type: 'session.finalize' }));
       };
 
       const processVAD = async (stream: VADStream, socket: WebSocket, signal: AbortSignal) => {
-        const abortPromise = new ThrowsPromise<never, Error>((_, reject) => {
-          if (signal.aborted) {
-            return reject(new Error('VAD aborted'));
-          }
-          const onAbort = () => reject(new Error('VAD aborted'));
-          signal.addEventListener('abort', onAbort, { once: true });
-        });
-
         const iterator = stream[Symbol.asyncIterator]();
-        try {
-          while (true) {
-            const result = await ThrowsPromise.race([iterator.next(), abortPromise]);
-            if (result.done) break;
-            if (result.value.type !== VADEventType.END_OF_SPEECH) continue;
-            if (socket.readyState !== 1) return;
-            socket.send(JSON.stringify({ type: 'session.finalize' }));
-          }
-        } catch (e) {
-          if ((e as Error).message === 'VAD aborted') return;
-          throw e;
+        while (true) {
+          const result = await this.nextUntilAborted(iterator, signal);
+          if (result === undefined || result.done) return;
+          if (result.value.type !== VADEventType.END_OF_SPEECH) continue;
+          if (socket.readyState !== 1) return;
+          socket.send(JSON.stringify({ type: 'session.finalize' }));
         }
       };