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",
     "commander": "^12.0.0",
     "livekit-server-sdk": "^2.9.2",
     "pino": "^8.19.0",

agents/src/stream/identity_transform.ts

@@ -1,7 +1,6 @@
 // SPDX-FileCopyrightText: 2025 LiveKit, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
-import { TransformStream } from 'node:stream/web';
 
 export class IdentityTransform<T> extends TransformStream<T, T> {
   constructor() {

agents/src/stream/injectable_stream.ts

@@ -0,0 +1,48 @@
+import { Mutex } from '@livekit/mutex';
+import { mergeReadableStreams } from '@std/streams';
+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();
+
+  constructor(source: ReadableStream<T>) {
+    this.source = source;
+    this.identityStream = new IdentityTransform<T>();
+    this.mergedStream = mergeReadableStreams<T>(this.source, this.identityStream.readable);
+  }
+
+  async inject(value: T) {
+    const unlock = await this.injectMutex.lock();
+    try {
+      const writer = this.identityStream.writable.getWriter();
+      await writer.write(value);
+      await writer.close();
+    } finally {
+      unlock();
+    }
+  }
+
+  async close() {
+    const unlock = await this.injectMutex.lock();
+    try {
+      // this will not cancel the source stream but instead keep the readable open until the source finishes
+      await this.identityStream.writable.close();
+    } finally {
+      unlock();
+    }
+  }
+
+  async cancel(reason?: any) {
+    await Promise.all([
+      this.mergedStream.cancel(reason),
+      this.identityStream.writable.abort(reason),
+    ]);
+  }
+
+  get readable() {
+    return this.mergedStream;
+  }
+}

package.json

@@ -21,6 +21,7 @@
     "examples:minimal": "pnpm exec tsx examples/src/multimodal_agent.ts"
   },
   "devDependencies": {
+    "@types/node": "^22.13.10",
     "@changesets/cli": "^2.27.1",
     "@livekit/changesets-changelog-github": "^0.0.4",
     "@rushstack/heft": "^0.66.0",
@@ -44,5 +45,5 @@
     "typescript": "^5.4.5",
     "vitest": "^1.6.0"
   },
-  "packageManager": "pnpm@9.7.0"
+  "packageManager": "pnpm@10.11.0"
 }