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';
 
 export class DeferredReadableStream<T> {

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,62 @@
+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();
+  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;
+  }
+}

agents/src/stt/stt.ts

@@ -4,7 +4,6 @@
 import type { AudioFrame } from '@livekit/rtc-node';
 import type { TypedEventEmitter as TypedEmitter } from '@livekit/typed-emitter';
 import { EventEmitter } from 'node:events';
-import type { ReadableStream } from 'node:stream/web';
 import { log } from '../log.js';
 import type { STTMetrics } from '../metrics/base.js';
 import { DeferredReadableStream } from '../stream/deferred_stream.js';

agents/src/tts/tts.ts

@@ -4,7 +4,6 @@
 import type { AudioFrame } from '@livekit/rtc-node';
 import type { TypedEventEmitter as TypedEmitter } from '@livekit/typed-emitter';
 import { EventEmitter } from 'node:events';
-import type { ReadableStream } from 'node:stream/web';
 import { log } from '../log.js';
 import type { TTSMetrics } from '../metrics/base.js';
 import { DeferredReadableStream } from '../stream/deferred_stream.js';

agents/src/vad.ts

@@ -4,11 +4,7 @@
 import type { AudioFrame } from '@livekit/rtc-node';
 import type { TypedEventEmitter as TypedEmitter } from '@livekit/typed-emitter';
 import { EventEmitter } from 'node:events';
-import type {
-  ReadableStream,
-  ReadableStreamDefaultReader,
-  WritableStreamDefaultWriter,
-} from 'node:stream/web';
+
 import { log } from './log.js';
 import type { VADMetrics } from './metrics/base.js';
 import { DeferredReadableStream } from './stream/deferred_stream.js';

agents/src/voice/agent.ts

@@ -6,11 +6,10 @@
 
 /* eslint-disable @typescript-eslint/no-explicit-any */
 import type { AudioFrame } from '@livekit/rtc-node';
-import { ReadableStream } from 'node:stream/web';
 import type { ChatChunk, ChatMessage, LLM } from '../llm/index.js';
 import { ChatContext } from '../llm/index.js';
-import { StreamAdapter as STTStreamAdapter } from '../stt/index.js';
 import type { STT, SpeechEvent } from '../stt/index.js';
+import { StreamAdapter as STTStreamAdapter } from '../stt/index.js';
 import { SentenceTokenizer as BasicSentenceTokenizer } from '../tokenize/basic/index.js';
 import type { TTS } from '../tts/index.js';
 import { SynthesizeStream, StreamAdapter as TTSStreamAdapter } from '../tts/index.js';

agents/src/voice/agent_activity.ts

@@ -3,7 +3,6 @@
 // SPDX-License-Identifier: Apache-2.0
 import type { AudioFrame } from '@livekit/rtc-node';
 import { Heap } from 'heap-js';
-import type { ReadableStream } from 'node:stream/web';
 import { type ChatContext, ChatMessage, ChatRole } from '../llm/chat_context.js';
 import type { LLM } from '../llm/index.js';
 import { log } from '../log.js';
@@ -12,8 +11,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

@@ -2,9 +2,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/audio_recognition.ts

@@ -3,7 +3,6 @@
 // SPDX-License-Identifier: Apache-2.0
 import type { AudioFrame } from '@livekit/rtc-node';
 import { delay } from '@std/async';
-import { ReadableStream } from 'node:stream/web';
 import { type ChatContext, ChatRole } from '../llm/chat_context.js';
 import { log } from '../log.js';
 import { DeferredReadableStream } from '../stream/deferred_stream.js';

agents/src/voice/generation.ts

@@ -3,7 +3,6 @@
 // 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 { ChatContext } from '../llm/chat_context.js';
 import { IdentityTransform } from '../stream/identity_transform.js';
 import { Future } from '../utils.js';

agents/src/voice/io.ts

@@ -2,7 +2,6 @@
 //
 // SPDX-License-Identifier: Apache-2.0
 import type { AudioFrame } from '@livekit/rtc-node';
-import type { ReadableStream } from 'node:stream/web';
 import type { ChatContext } from '../llm/chat_context.js';
 import type { ChatChunk } from '../llm/llm.js';
 import type { SpeechEvent } from '../stt/stt.js';

agents/src/voice/room_io.ts

@@ -13,7 +13,6 @@ import {
   TrackPublishOptions,
   TrackSource,
 } from '@livekit/rtc-node';
-import type { ReadableStream } from 'node:stream/web';
 import { log } from '../log.js';
 import { DeferredReadableStream } from '../stream/deferred_stream.js';
 import type { AgentSession } from './agent_session.js';
@@ -57,7 +56,7 @@ export class RoomIO {
           // TODO(AJS-41) remove hardcoded sample rate
           sampleRate: 16000,
           numChannels: 1,
-        }),
+        }) as ReadableStream<AudioFrame>,
       );
     }
   };