fix(agents): keep downstream-cancelled stream channels from leaking unhandled rejections

.changeset/stream-channel-downstream-cancel.md

@@ -0,0 +1,5 @@
+---
+'@livekit/agents': patch
+---
+
+Stream channels: tolerate the consumer cancelling the readable side. Fire-and-forget write() and close() on a downstream-cancelled channel no longer surface as unhandled promise rejections (`Error: undefined`), the `closed` getter now reflects downstream cancellation, and the audio forwarder cancels its reader with a real reason.

agents/src/log.ts

@@ -5,7 +5,7 @@ import { Writable } from 'node:stream';
 import type { DestinationStream, Logger } from 'pino';
 import { multistream, pino } from 'pino';
 import { build as pinoPretty } from 'pino-pretty';
-import { type LoggerOptions, log, loggerOptions, setLoggerState } from './log_core.js';
+import { type LoggerOptions, log, loggerOptions, setLoggerState, tryLog } from './log_core.js';
 import { type PinoLogObject, emitToOtel } from './telemetry/pino_otel_transport.js';
 
 const OTEL_ENABLED_KEY = Symbol.for('@livekit/agents:otelEnabled');
@@ -19,7 +19,7 @@ const globals = globalThis as typeof globalThis & GlobalState;
 // LiveKit Cloud injects this into deployed agents. Child processes inherit it.
 const deployedRegion = process.env.LIVEKIT_REGION_NAME || undefined;
 
-export { log, loggerOptions, type LoggerOptions };
+export { log, loggerOptions, tryLog, type LoggerOptions };
 
 const createLogger = ({ pretty, level }: LoggerOptions): Logger => {
   const logLevel = level || 'info';

agents/src/log_core.ts

@@ -34,6 +34,12 @@ export const log = (): Logger => {
   return logger;
 };
 
+/**
+ * Like {@link log}, but returns undefined before initializeLogger() has run.
+ * For best-effort logging from code that may run outside a worker.
+ */
+export const tryLog = (): Logger | undefined => globals[LOGGER_KEY];
+
 /** @internal */
 export const setLoggerState = (logger: Logger, options: LoggerOptions): void => {
   globals[LOGGER_OPTIONS_KEY] = options;

agents/src/stream/stream_channel.test.ts

@@ -163,4 +163,40 @@ describe('StreamChannel', () => {
     const result3 = await read3;
     expect(result3.done).toBe(true);
   });
+
+  it('should not produce unhandled rejections when the reader cancels mid-stream', async () => {
+    const unhandled: unknown[] = [];
+    const onUnhandled = (reason: unknown) => {
+      unhandled.push(reason);
+    };
+    process.on('unhandledRejection', onUnhandled);
+    try {
+      const channel = createStreamChannel<string>();
+      const reader = channel.stream().getReader();
+
+      await channel.write('before');
+      await reader.read();
+      await reader.cancel();
+
+      // Fire-and-forget write and close after the consumer is gone, as
+      // realtime producers do.
+      channel.write('after-cancel');
+      await channel.close();
+
+      await new Promise((resolve) => setTimeout(resolve, 20));
+      expect(unhandled).toEqual([]);
+      expect(channel.closed).toBe(true);
+    } finally {
+      process.off('unhandledRejection', onUnhandled);
+    }
+  });
+
+  it('should still reject an awaited write after the reader cancelled', async () => {
+    const channel = createStreamChannel<string>();
+    const reader = channel.stream().getReader();
+
+    await reader.cancel(new Error('consumer gone'));
+
+    await expect(channel.write('late')).rejects.toThrow('consumer gone');
+  });
 });

agents/src/stream/stream_channel.ts

@@ -2,6 +2,7 @@
 //
 // SPDX-License-Identifier: Apache-2.0
 import type { ReadableStream } from 'node:stream/web';
+import { tryLog } from '../log.js';
 import { IdentityTransform } from './identity_transform.js';
 
 export interface StreamChannel<T, E extends Error = Error> {
@@ -18,8 +19,29 @@ export function createStreamChannel<T, E extends Error = Error>(): StreamChannel
   const writer = transform.writable.getWriter();
   let isClosed = false;
 
+  // The consumer side can cancel the readable (or the stream can error), which
+  // errors the writable and would otherwise reject every later write() and
+  // close() with the cancel reason - surfacing as unhandled rejections from
+  // fire-and-forget producers.
+  writer.closed.catch((error: unknown) => {
+    if (!isClosed) {
+      isClosed = true;
+      tryLog()?.debug({ error }, 'stream channel writable errored or was cancelled downstream');
+    }
+  });
+
   return {
-    write: (chunk: T) => writer.write(chunk),
+    write: (chunk: T) => {
+      const result = writer.write(chunk);
+      // Mark the rejection as handled for fire-and-forget producers; callers
+      // that await the returned promise still observe it.
+      result.catch((error: unknown) => {
+        if (!isClosed) {
+          tryLog()?.debug({ error }, 'stream channel write failed');
+        }
+      });
+      return result;
+    },
     stream: () => transform.readable,
     abort: async (error: E) => {
       if (isClosed) return;
@@ -53,16 +75,15 @@ export function createStreamChannel<T, E extends Error = Error>(): StreamChannel
     },
     close: async () => {
       try {
-        const result = await writer.close();
-        isClosed = true;
-        return result;
+        return await writer.close();
       } catch (e) {
-        if (e instanceof Error && e.name === 'TypeError') {
-          // Ignore error if the stream is already closed
-          isClosed = true;
-          return;
+        // Ignore error if the stream is already closed or errored - either way
+        // no more data flows, which is what close() asks for.
+        if (!isClosed && !(e instanceof Error && e.name === 'TypeError')) {
+          tryLog()?.debug({ error: e }, 'stream channel close failed');
         }
-        throw e;
+      } finally {
+        isClosed = true;
       }
     },
     get closed() {

agents/src/voice/generation.ts

@@ -1049,7 +1049,9 @@ async function forwardAudio(
   const reader = ttsStream.getReader();
   let resampler: AudioResampler | null = null;
   const cancelReader = () => {
-    void reader.cancel().catch((error) => {
+    // Cancel with a reason so producers still writing into the stream see a
+    // real error instead of `undefined`.
+    void reader.cancel(new Error('audio forwarding aborted')).catch((error) => {
       logger.debug({ error }, 'failed to cancel TTS stream reader after abort');
     });
   };