Improve AgentActivity resource cleanup

.changeset/hungry-schools-think.md

@@ -0,0 +1,5 @@
+---
+'@livekit/agents': patch
+---
+
+Fix improper resource cleanup inside AgentActivity by not close global STT / TTS / VAD components

agents/src/inference/stt.ts

@@ -422,8 +422,7 @@ export class SpeechStream<TModel extends STTModels> extends BaseSpeechStream {
       try {
         ws = await this.stt.connectWs(this.connOptions.timeoutMs);
 
-        // Wrap tasks for proper cancellation support using Task signals
-        const controller = new AbortController();
+        const controller = this.abortController; // Use base class abortController for proper cancellation
         const sendTask = Task.from(({ signal }) => send(ws!, signal), controller);
         const wsListenerTask = Task.from(({ signal }) => createWsListener(ws!, signal), controller);
         const recvTask = Task.from(({ signal }) => recv(signal), controller);

agents/src/ipc/job_proc_lazy_main.ts

@@ -189,7 +189,7 @@ const startJob = (
     let logger = log().child({ pid: proc.pid });
 
     process.on('unhandledRejection', (reason) => {
-      logger.error(reason);
+      logger.debug({ error: reason }, 'Unhandled promise rejection');
     });
 
     logger.debug('initializing job runner');

agents/src/stt/stream_adapter.ts

@@ -23,10 +23,14 @@ export class StreamAdapter extends STT {
     this.#stt.on('metrics_collected', (metrics) => {
       this.emit('metrics_collected', metrics);
     });
+
+    this.#stt.on('error', (error) => {
+      this.emit('error', error);
+    });
   }
 
-  _recognize(frame: AudioFrame): Promise<SpeechEvent> {
-    return this.#stt.recognize(frame);
+  _recognize(frame: AudioFrame, abortSignal?: AbortSignal): Promise<SpeechEvent> {
+    return this.#stt.recognize(frame, abortSignal);
   }
 
   stream(options?: { connOptions?: APIConnectOptions }): StreamAdapterWrapper {
@@ -46,6 +50,11 @@ export class StreamAdapterWrapper extends SpeechStream {
     this.label = `stt.StreamAdapterWrapper<${this.#stt.label}>`;
   }
 
+  close() {
+    super.close();
+    this.#vadStream.close();
+  }
+
   async monitorMetrics() {
     return; // do nothing
   }
@@ -72,7 +81,7 @@ export class StreamAdapterWrapper extends SpeechStream {
             this.output.put({ type: SpeechEventType.END_OF_SPEECH });
 
             try {
-              const event = await this.#stt.recognize(ev.frames);
+              const event = await this.#stt.recognize(ev.frames, this.abortSignal);
               if (!event.alternatives![0].text) {
                 continue;
               }
@@ -93,6 +102,6 @@ export class StreamAdapterWrapper extends SpeechStream {
       }
     };
 
-    Promise.all([forwardInput(), recognize()]);
+    await Promise.all([forwardInput(), recognize()]);
   }
 }

agents/src/stt/stt.ts

@@ -113,9 +113,9 @@ export abstract class STT extends (EventEmitter as new () => TypedEmitter<STTCal
   }
 
   /** Receives an audio buffer and returns transcription in the form of a {@link SpeechEvent} */
-  async recognize(frame: AudioBuffer): Promise<SpeechEvent> {
+  async recognize(frame: AudioBuffer, abortSignal?: AbortSignal): Promise<SpeechEvent> {
     const startTime = process.hrtime.bigint();
-    const event = await this._recognize(frame);
+    const event = await this._recognize(frame, abortSignal);
     const durationMs = Number((process.hrtime.bigint() - startTime) / BigInt(1000000));
     this.emit('metrics_collected', {
       type: 'stt_metrics',
@@ -128,7 +128,11 @@ export abstract class STT extends (EventEmitter as new () => TypedEmitter<STTCal
     });
     return event;
   }
-  protected abstract _recognize(frame: AudioBuffer): Promise<SpeechEvent>;
+
+  protected abstract _recognize(
+    frame: AudioBuffer,
+    abortSignal?: AbortSignal,
+  ): Promise<SpeechEvent>;
 
   /**
    * Returns a {@link SpeechStream} that can be used to push audio frames and receive
@@ -173,6 +177,8 @@ export abstract class SpeechStream implements AsyncIterableIterator<SpeechEvent>
   private logger = log();
   private _connOptions: APIConnectOptions;
 
+  protected abortController = new AbortController();
+
   constructor(
     stt: STT,
     sampleRate?: number,
@@ -290,6 +296,10 @@ export abstract class SpeechStream implements AsyncIterableIterator<SpeechEvent>
 
   protected abstract run(): Promise<void>;
 
+  protected get abortSignal(): AbortSignal {
+    return this.abortController.signal;
+  }
+
   updateInputStream(audioStream: ReadableStream<AudioFrame>) {
     this.deferredInputStream.setSource(audioStream);
   }
@@ -354,6 +364,7 @@ export abstract class SpeechStream implements AsyncIterableIterator<SpeechEvent>
     if (!this.input.closed) this.input.close();
     if (!this.queue.closed) this.queue.close();
     if (!this.output.closed) this.output.close();
+    if (!this.abortController.signal.aborted) this.abortController.abort();
     this.closed = true;
   }
 

agents/src/voice/agent.ts

@@ -260,28 +260,41 @@ export class Agent<UserData = any> {
       let wrapped_stt = activity.stt;
 
       if (!wrapped_stt.capabilities.streaming) {
-        if (!agent.vad) {
+        const vad = agent.vad || activity.vad;
+        if (!vad) {
           throw new Error(
             'STT does not support streaming, add a VAD to the AgentTask/VoiceAgent to enable streaming',
           );
         }
-        wrapped_stt = new STTStreamAdapter(wrapped_stt, agent.vad);
+        wrapped_stt = new STTStreamAdapter(wrapped_stt, vad);
       }
 
       const connOptions = activity.agentSession.connOptions.sttConnOptions;
       const stream = wrapped_stt.stream({ connOptions });
       stream.updateInputStream(audio);
 
+      let cleaned = false;
+      const cleanup = () => {
+        if (cleaned) return;
+        cleaned = true;
+        stream.detachInputStream();
+        stream.close();
+      };
+
       return new ReadableStream({
         async start(controller) {
-          for await (const event of stream) {
-            controller.enqueue(event);
+          try {
+            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();
           }
-          controller.close();
         },
         cancel() {
-          stream.detachInputStream();
-          stream.close();
+          cleanup();
         },
       });
     },
@@ -314,15 +327,27 @@ export class Agent<UserData = any> {
         connOptions,
         parallelToolCalls: true,
       });
+
+      let cleaned = false;
+      const cleanup = () => {
+        if (cleaned) return;
+        cleaned = true;
+        stream.close();
+      };
+
       return new ReadableStream({
         async start(controller) {
-          for await (const chunk of stream) {
-            controller.enqueue(chunk);
+          try {
+            for await (const chunk of stream) {
+              controller.enqueue(chunk);
+            }
+            controller.close();
+          } finally {
+            cleanup();
           }
-          controller.close();
         },
         cancel() {
-          stream.close();
+          cleanup();
         },
       });
     },
@@ -347,18 +372,29 @@ export class Agent<UserData = any> {
       const stream = wrapped_tts.stream({ connOptions });
       stream.updateInputStream(text);
 
+      let cleaned = false;
+      const cleanup = () => {
+        if (cleaned) return;
+        cleaned = true;
+        stream.close();
+      };
+
       return new ReadableStream({
         async start(controller) {
-          for await (const chunk of stream) {
-            if (chunk === SynthesizeStream.END_OF_STREAM) {
-              break;
+          try {
+            for await (const chunk of stream) {
+              if (chunk === SynthesizeStream.END_OF_STREAM) {
+                break;
+              }
+              controller.enqueue(chunk.frame);
             }
-            controller.enqueue(chunk.frame);
+            controller.close();
+          } finally {
+            cleanup();
           }
-          controller.close();
         },
         cancel() {
-          stream.close();
+          cleanup();
         },
       });
     },

agents/src/voice/agent_activity.ts

@@ -2259,15 +2259,12 @@ export class AgentActivity implements RecognitionHooks {
       }
       if (this.stt instanceof STT) {
         this.stt.off('metrics_collected', this.onMetricsCollected);
-        await this.stt.close();
       }
       if (this.tts instanceof TTS) {
         this.tts.off('metrics_collected', this.onMetricsCollected);
-        await this.tts.close();
       }
       if (this.vad instanceof VAD) {
         this.vad.off('metrics_collected', this.onMetricsCollected);
-        await this.vad.close();
       }
 
       this.detachAudioInput();

agents/src/voice/agent_session.ts

@@ -328,15 +328,6 @@ export class AgentSession<
         );
       }
 
-      this.logger.info(
-        {
-          input: this.input.audio,
-          output: this.output.audio,
-          enableRecording: this._enableRecording,
-        },
-        'Recording audio input and output',
-      );
-
       if (this.input.audio && this.output.audio && this._enableRecording) {
         this._recorderIO = new RecorderIO({ agentSession: this });
         this.input.audio = this._recorderIO.recordInput(this.input.audio);

agents/src/worker.ts

@@ -729,7 +729,7 @@ export class AgentServer {
 
     const req = new JobRequest(msg.job!, onReject, onAccept);
     this.#logger
-      .child({ job: msg.job, resuming: msg.resuming, agentName: this.#opts.agentName })
+      .child({ jobId: msg.job?.id, resuming: msg.resuming, agentName: this.#opts.agentName })
       .info('received job request');
 
     const jobRequestTask = async () => {

examples/src/basic_tool_call_agent.ts

@@ -4,14 +4,15 @@
 import {
   type JobContext,
   type JobProcess,
-  WorkerOptions,
+  ServerOptions,
   cli,
   defineAgent,
   llm,
   voice,
 } from '@livekit/agents';
 import * as livekit from '@livekit/agents-plugin-livekit';
 import * as silero from '@livekit/agents-plugin-silero';
+import { BackgroundVoiceCancellation } from '@livekit/noise-cancellation-node';
 import { fileURLToPath } from 'node:url';
 import { z } from 'zod';
 
@@ -148,8 +149,11 @@ export default defineAgent({
     await session.start({
       agent: routerAgent,
       room: ctx.room,
+      inputOptions: {
+        noiseCancellation: BackgroundVoiceCancellation(),
+      },
     });
   },
 });
 
-cli.runApp(new WorkerOptions({ agent: fileURLToPath(import.meta.url) }));
+cli.runApp(new ServerOptions({ agent: fileURLToPath(import.meta.url) }));