wip injectable stream

agents/package.json

@@ -41,6 +41,7 @@
     "@livekit/mutex": "^1.1.1",
     "@livekit/protocol": "^1.29.1",
     "@livekit/typed-emitter": "^3.0.0",
+    "@std/streams": "jsr:^1.0.9",
     "@std/async": "npm:@jsr/std__async@^1.0.13",
     "commander": "^12.0.0",
     "heap-js": "^2.6.0",

agents/src/stream/deferred_stream.ts

@@ -1,7 +1,6 @@
 // SPDX-FileCopyrightText: 2025 LiveKit, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
-import { type ReadableStream } from 'node:stream/web';
 import { IdentityTransform } from './identity_transform.js';
 
 /**

agents/src/stream/injectable_stream.ts

@@ -0,0 +1,98 @@
+import { Mutex } from '@livekit/mutex';
+import { mergeReadableStreams } from '@std/streams';
+import type { ReadableStream } from 'node:stream/web';
+import { IdentityTransform } from './identity_transform.js';
+
+export class InjectableStream<T> {
+  private source: ReadableStream<T>;
+  private identityStream: IdentityTransform<T>;
+  private mergedStream: ReadableStream<T>;
+  private injectMutex = new Mutex();
+  private writer: WritableStreamDefaultWriter<T>;
+  private closed = false;
+
+  constructor(source: ReadableStream<T>) {
+    this.source = source;
+    this.identityStream = new IdentityTransform<T>();
+    this.mergedStream = mergeReadableStreams<T>(this.source, this.identityStream.readable);
+    this.writer = this.identityStream.writable.getWriter();
+  }
+
+  async inject(value: T) {
+    const unlock = await this.injectMutex.lock();
+
+    if (this.closed) {
+      throw new Error('Cannot inject into a closed stream');
+    }
+
+    try {
+      await this.writer.write(value);
+    } finally {
+      unlock();
+    }
+  }
+
+  async close() {
+    const unlock = await this.injectMutex.lock();
+
+    if (this.closed) {
+      return;
+    }
+
+    try {
+      // this will not cancel the source stream but instead keep the readable open until the source finishes
+      this.writer.releaseLock();
+      await this.identityStream.writable.close();
+      this.closed = true;
+    } finally {
+      unlock();
+    }
+  }
+
+  async cancel(reason?: any) {
+    await this.close();
+    await Promise.all([
+      this.mergedStream.cancel(reason),
+      this.identityStream.writable.abort(reason),
+    ]);
+  }
+
+  get readable() {
+    return this.mergedStream;
+  }
+}
+
+// // Copied from @std/streams/merge-readable-streams.ts to avoid incompetible ReadableStream types
+// export function mergeReadableStreams<T>(
+//   ...streams: ReadableStream<T>[]
+// ): ReadableStream<T> {
+//   const resolvePromises = streams.map(() => Promise.withResolvers<void>());
+//   return new ReadableStream<T>({
+//     start(controller) {
+//       let mustClose = false;
+//       Promise.all(resolvePromises.map(({ promise }) => promise))
+//         .then(() => {
+//           controller.close();
+//         })
+//         .catch((error) => {
+//           mustClose = true;
+//           controller.error(error);
+//         });
+//       for (const [index, stream] of streams.entries()) {
+//         (async () => {
+//           try {
+//             for await (const data of stream) {
+//               if (mustClose) {
+//                 break;
+//               }
+//               controller.enqueue(data);
+//             }
+//             resolvePromises[index]!.resolve();
+//           } catch (error) {
+//             resolvePromises[index]!.reject(error);
+//           }
+//         })();
+//       }
+//     },
+//   });
+// }

agents/src/voice/agent_activity.ts

@@ -12,8 +12,8 @@ import type { STT, SpeechEvent } from '../stt/stt.js';
 import type { TTS } from '../tts/tts.js';
 import { Future } from '../utils.js';
 import type { VADEvent } from '../vad.js';
-import { StopResponse } from './agent.js';
 import type { Agent } from './agent.js';
+import { StopResponse } from './agent.js';
 import type { AgentSession } from './agent_session.js';
 import {
   AudioRecognition,

agents/src/voice/agent_session.ts

@@ -3,8 +3,8 @@
 // SPDX-License-Identifier: Apache-2.0
 import type { AudioFrame, AudioSource, Room } from '@livekit/rtc-node';
 import type { ReadableStream } from 'node:stream/web';
-import { ChatContext } from '../llm/chat_context.js';
 import type { ChatMessage } from '../llm/chat_context.js';
+import { ChatContext } from '../llm/chat_context.js';
 import type { LLM } from '../llm/index.js';
 import { log } from '../log.js';
 import type { AgentState } from '../pipeline/index.js';

agents/src/voice/generation.ts

@@ -3,7 +3,7 @@
 // SPDX-License-Identifier: Apache-2.0
 import type { AudioFrame, AudioSource } from '@livekit/rtc-node';
 import { randomUUID } from 'node:crypto';
-import type { ReadableStream } from 'stream/web';
+import type { ReadableStream } from 'node:stream/web';
 import type { ChatContext } from '../llm/chat_context.js';
 import { IdentityTransform } from '../stream/identity_transform.js';
 import { Future } from '../utils.js';