testing 1

Author: nbsp

agents/src/pipeline/agent_output.ts

@@ -148,9 +148,10 @@ const stringSynthesisTask = (text: string, handle: SynthesisHandle): Cancellable
       if (cancelled || audio === SynthesizeStream.END_OF_STREAM) {
         break;
       }
-      writer.write(audio.frame);
+      await writer.write(audio.frame);
     }
-    writer.write(SynthesisHandle.FLUSH_SENTINEL);
+    await writer.write(SynthesisHandle.FLUSH_SENTINEL);
+    writer.releaseLock();
 
     resolve(text);
   });
@@ -177,9 +178,10 @@ const streamSynthesisTask = (
         if (audio === SynthesizeStream.END_OF_STREAM) {
           break;
         }
-        writer.write(audio.frame);
+        await writer.write(audio.frame);
       }
-      writer.write(SynthesisHandle.FLUSH_SENTINEL);
+      await writer.write(SynthesisHandle.FLUSH_SENTINEL);
+      writer.releaseLock();
     };
     readGeneratedAudio();
 

agents/src/stt/stt.ts

@@ -4,7 +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 } from 'node:stream/web';
+import type { ReadableStream, WritableStreamDefaultWriter } from 'node:stream/web';
 import { TransformStream } from 'node:stream/web';
 import type { STTMetrics } from '../metrics/base.js';
 import type { AudioBuffer } from '../utils.js';
@@ -148,12 +148,11 @@ export abstract class SpeechStream implements AsyncIterableIterator<SpeechEvent>
   protected inputClosed = false;
   #stt: STT;
   #outputReadable: ReadableStream<SpeechEvent>;
+  #writer: WritableStreamDefaultWriter<AudioFrame | typeof SpeechStream.FLUSH_SENTINEL>;
 
   constructor(stt: STT) {
     this.#stt = stt;
-    this.output.writable.close().then(() => {
-      this.inputClosed = true;
-    });
+    this.#writer = this.input.writable.getWriter();
     const [r1, r2] = this.output.readable.tee();
     this.#outputReadable = r1;
     this.monitorMetrics(r2);
@@ -185,7 +184,7 @@ export abstract class SpeechStream implements AsyncIterableIterator<SpeechEvent>
     if (this.closed) {
       throw new Error('Stream is closed');
     }
-    this.input.writable.getWriter().write(frame);
+    this.#writer.write(frame);
   }
 
   /** Flush the STT, causing it to process all pending text */
@@ -196,7 +195,7 @@ export abstract class SpeechStream implements AsyncIterableIterator<SpeechEvent>
     if (this.closed) {
       throw new Error('Stream is closed');
     }
-    this.input.writable.getWriter().write(SpeechStream.FLUSH_SENTINEL);
+    this.#writer.write(SpeechStream.FLUSH_SENTINEL);
   }
 
   /** Mark the input as ended and forbid additional pushes */
@@ -207,6 +206,7 @@ export abstract class SpeechStream implements AsyncIterableIterator<SpeechEvent>
     if (this.closed) {
       throw new Error('Stream is closed');
     }
+    this.inputClosed = true;
     this.input.writable.close();
   }
 
@@ -228,6 +228,7 @@ export abstract class SpeechStream implements AsyncIterableIterator<SpeechEvent>
     this.input.writable.close();
     this.output.writable.close();
     this.closed = true;
+    this.inputClosed = true;
   }
 
   [Symbol.asyncIterator](): SpeechStream {

agents/src/tts/tts.ts

@@ -4,7 +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 } from 'node:stream/web';
+import type { ReadableStream, WritableStreamDefaultWriter } from 'node:stream/web';
 import { TransformStream } from 'node:stream/web';
 import type { TTSMetrics } from '../metrics/base.js';
 import { mergeFrames } from '../utils.js';
@@ -122,12 +122,14 @@ export abstract class SynthesizeStream
   #metricsPendingTexts: string[] = [];
   #metricsText = '';
   #outputReadable: ReadableStream<SynthesizedAudio | typeof SynthesizeStream.END_OF_STREAM>;
+  #writer: WritableStreamDefaultWriter<string | typeof SynthesizeStream.FLUSH_SENTINEL>;
 
   constructor(tts: TTS) {
     this.#tts = tts;
     this.output.writable.close().then(() => {
       this.inputClosed = true;
     });
+    this.#writer = this.input.writable.getWriter();
     const [r1, r2] = this.output.readable.tee();
     this.#outputReadable = r1;
     this.monitorMetrics(r2);
@@ -187,7 +189,7 @@ export abstract class SynthesizeStream
     if (this.closed) {
       throw new Error('Stream is closed');
     }
-    this.input.writable.getWriter().write(text);
+    this.#writer.write(text);
   }
 
   /** Flush the TTS, causing it to process all pending text */
@@ -202,7 +204,7 @@ export abstract class SynthesizeStream
     if (this.closed) {
       throw new Error('Stream is closed');
     }
-    this.input.writable.getWriter().write(SynthesizeStream.FLUSH_SENTINEL);
+    this.#writer.write(SynthesizeStream.FLUSH_SENTINEL);
   }
 
   /** Mark the input as ended and forbid additional pushes */
@@ -213,7 +215,7 @@ export abstract class SynthesizeStream
     if (this.closed) {
       throw new Error('Stream is closed');
     }
-    this.input.writable.close();
+    this.#writer.close();
   }
 
   async next(): Promise<IteratorResult<SynthesizedAudio | typeof SynthesizeStream.END_OF_STREAM>> {
@@ -231,6 +233,7 @@ export abstract class SynthesizeStream
 
   /** Close both the input and output of the TTS stream */
   close() {
+    this.#writer.close();
     this.input.writable.close();
     this.output.writable.close();
     this.closed = true;

agents/src/vad.ts

@@ -4,7 +4,11 @@
 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 type {
+  ReadableStream,
+  ReadableStreamDefaultReader,
+  WritableStreamDefaultWriter,
+} from 'node:stream/web';
 import { TransformStream } from 'node:stream/web';
 import type { VADMetrics } from './metrics/base.js';
 
@@ -87,15 +91,14 @@ export abstract class VADStream implements AsyncIterableIterator<VADEvent> {
   protected inputClosed = false;
   #vad: VAD;
   #lastActivityTime = BigInt(0);
-  #outputReadable: ReadableStream<VADEvent>;
+  #writer: WritableStreamDefaultWriter<AudioFrame | typeof VADStream.FLUSH_SENTINEL>;
+  #reader: ReadableStreamDefaultReader<VADEvent>;
 
   constructor(vad: VAD) {
     this.#vad = vad;
-    this.output.writable.close().then(() => {
-      this.inputClosed = true;
-    });
     const [r1, r2] = this.output.readable.tee();
-    this.#outputReadable = r1;
+    this.#reader = r1.getReader();
+    this.#writer = this.input.writable.getWriter();
     this.monitorMetrics(r2);
   }
 
@@ -137,7 +140,7 @@ export abstract class VADStream implements AsyncIterableIterator<VADEvent> {
     if (this.closed) {
       throw new Error('Stream is closed');
     }
-    this.input.writable.getWriter().write(frame);
+    this.#writer.write(frame);
   }
 
   flush() {
@@ -147,7 +150,8 @@ export abstract class VADStream implements AsyncIterableIterator<VADEvent> {
     if (this.closed) {
       throw new Error('Stream is closed');
     }
-    this.input.writable.getWriter().write(VADStream.FLUSH_SENTINEL);
+    this.inputClosed = true;
+    this.#writer.write(VADStream.FLUSH_SENTINEL);
   }
 
   endInput() {
@@ -157,20 +161,18 @@ export abstract class VADStream implements AsyncIterableIterator<VADEvent> {
     if (this.closed) {
       throw new Error('Stream is closed');
     }
+    this.inputClosed = true;
     this.input.writable.close();
   }
 
   async next(): Promise<IteratorResult<VADEvent>> {
-    return this.#outputReadable
-      .getReader()
-      .read()
-      .then(({ value }) => {
-        if (value) {
-          return { value, done: false };
-        } else {
-          return { value: undefined, done: true };
-        }
-      });
+    return this.#reader.read().then(({ value }) => {
+      if (value) {
+        return { value, done: false };
+      } else {
+        return { value: undefined, done: true };
+      }
+    });
   }
 
   close() {