fix: explicitly close AudioResampler instances too free up resources (#1210)

Author: lukasIO

.changeset/smart-kids-tickle.md

@@ -0,0 +1,8 @@
+---
+"@livekit/agents": patch
+"@livekit/agents-plugin-google": patch
+"@livekit/agents-plugin-phonic": patch
+"@livekit/agents-plugin-silero": patch
+---
+
+Explicitly close AudioResampler instances too free up resources

agents/src/inference/interruption/interruption_stream.ts

@@ -414,6 +414,7 @@ export class InterruptionStreamBase {
 
   async close(): Promise<void> {
     if (!this.inputStream.closed) await this.inputStream.close();
+    this.resampler?.close();
     this.model.removeStream(this);
   }
 }

agents/src/ipc/supervised_proc.test.ts

@@ -66,14 +66,14 @@ describe('pidusage on dead process', () => {
     const exitPromise = new Promise<void>((r) => child.on('exit', r));
 
     child.kill('SIGKILL');
+    await exitPromise;
 
     const results = await Promise.all([
       getChildMemoryUsageMB(pid),
       getChildMemoryUsageMB(pid),
       getChildMemoryUsageMB(pid),
     ]);
 
-    await exitPromise;
     expect(results.every((r) => r === 0)).toBe(true);
   });
 });

agents/src/stt/stt.ts

@@ -445,6 +445,7 @@ export abstract class SpeechStream implements AsyncIterableIterator<SpeechEvent>
     if (!this.queue.closed) this.queue.close();
     if (!this.output.closed) this.output.close();
     if (!this.abortController.signal.aborted) this.abortController.abort();
+    this.resampler?.close();
     this.closed = true;
   }
 

agents/src/tts/fallback_adapter.ts

@@ -324,6 +324,8 @@ class FallbackChunkedStream extends ChunkedStream {
         this.adapter.markUnAvailable(i);
         continue;
       }
+      const resampler = this.adapter.createResamplerForTTS(i);
+
       try {
         this._logger.debug({ tts: tts.label }, 'attempting TTS synthesis');
         const connOptions: APIConnectOptions = {
@@ -332,7 +334,6 @@ class FallbackChunkedStream extends ChunkedStream {
         };
         const stream = tts.synthesize(this.inputText, connOptions, this.abortSignal);
         let audioReceived = false;
-        const resampler = this.adapter.createResamplerForTTS(i);
         for await (const audio of stream) {
           if (this.abortController.signal.aborted) {
             stream.close();
@@ -384,6 +385,8 @@ class FallbackChunkedStream extends ChunkedStream {
         } else {
           throw error;
         }
+      } finally {
+        resampler?.close();
       }
     }
     const labels = this.adapter.ttsInstances.map((t) => t.label).join(', ');
@@ -443,6 +446,7 @@ class FallbackSynthesizeStream extends SynthesizeStream {
         this.adapter.markUnAvailable(i);
         continue;
       }
+      const resampler = this.adapter.createResamplerForTTS(i);
 
       try {
         this._logger.debug({ tts: originalTts.label }, 'attempting TTS stream');
@@ -453,7 +457,6 @@ class FallbackSynthesizeStream extends SynthesizeStream {
         };
 
         const stream = tts.stream({ connOptions });
-        const resampler = this.adapter.createResamplerForTTS(i);
         let bufferIndex = 0;
         let streamOutputCompleted = false;
         const forwardBufferToTTS = async () => {
@@ -575,6 +578,8 @@ class FallbackSynthesizeStream extends SynthesizeStream {
         } else {
           throw error;
         }
+      } finally {
+        resampler?.close();
       }
     }
     await readInputLLMStream.catch(() => {});

agents/src/utils.test.ts

@@ -147,7 +147,7 @@ describe('utils', () => {
         return 'completed';
       });
 
-      await delay(39);
+      await delay(35);
       task.cancel();
 
       expect(arr).toEqual([0, 1, 2]);

agents/src/utils.ts

@@ -723,6 +723,7 @@ export function resampleStream({
         for (const frame of resampler.flush()) {
           controller.enqueue(frame);
         }
+        resampler.close();
       }
     },
   });

agents/src/voice/generation.ts

@@ -855,6 +855,7 @@ async function forwardAudio(
 
     reader?.releaseLock();
     audioOutput.flush();
+    resampler?.close();
   }
 }
 

agents/src/voice/recorder_io/recorder_io.ts

@@ -352,6 +352,8 @@ export class RecorderIO {
     } finally {
       inReader.releaseLock();
       outReader.releaseLock();
+      this.inResampler?.close();
+      this.outResampler?.close();
 
       if (!this.closeFuture.done) {
         this.closeFuture.resolve();

plugins/google/src/beta/realtime/realtime_api.ts

@@ -810,6 +810,7 @@ export class RealtimeSession extends llm.RealtimeSession {
     this.#closed = true;
 
     this.sessionShouldClose.set();
+    this.inputResampler?.close();
 
     await this.closeActiveSession();
 
@@ -1660,6 +1661,7 @@ export class RealtimeSession extends llm.RealtimeSession {
     if (this.inputResampler) {
       if (frame.sampleRate !== this.inputResamplerInputRate) {
         // input audio changed to a different sample rate
+        this.inputResampler.close();
         this.inputResampler = undefined;
         this.inputResamplerInputRate = undefined;
       }