Prototype streaming TTS chunks

2025-11-15 19:06:53 +01:00
parent 561e164b02
commit 7284efd0fe
5 changed files with 163 additions and 87 deletions
@@ -5,6 +5,8 @@ import type { SessionOutboundMessage } from "../messages.js";
 interface PendingPlayback {
  resolve: () => void;
  reject: (error: Error) => void;
+  pendingChunks: number;
+  streamEnded: boolean;
 }

 /**
@@ -36,8 +38,8 @@ export class TTSManager {
      return;
    }

-    // Generate TTS audio
-    const { audio, format } = await synthesizeSpeech(text);
+    // Generate TTS audio stream
+    const { stream, format } = await synthesizeSpeech(text);

    if (abortSignal.aborted) {
      console.log(
@@ -46,58 +48,129 @@ export class TTSManager {
      return;
    }

-    // Create unique ID for this audio segment
    const audioId = uuidv4();
+    let playbackResolve!: () => void;
+    let playbackReject!: (error: Error) => void;

-    // Store abort handler reference outside Promise constructor
-    let onAbort: (() => void) | undefined;
-
-    // Create promise that will be resolved when client confirms playback
    const playbackPromise = new Promise<void>((resolve, reject) => {
-      // Store handlers (no timeout - will resolve when client confirms or connection closes)
-      this.pendingPlaybacks.set(audioId, { resolve, reject });
-
-      // Handle abort signal
-      onAbort = () => {
-        console.log(
-          `[TTS-Manager ${this.sessionId}] Aborted while waiting for playback`
-        );
-        // Clean up pending playback
-        this.pendingPlaybacks.delete(audioId);
-        // Reject with abort error
-        resolve();
-      };
-
-      // Listen for abort (once: true for auto-cleanup if abort fires)
-      abortSignal.addEventListener("abort", onAbort, { once: true });
+      playbackResolve = resolve;
+      playbackReject = reject;
    });

-    // Clean up abort listener when promise settles (in case abort never fires)
-    if (onAbort) {
-      playbackPromise.finally(() => {
-        abortSignal!.removeEventListener("abort", onAbort!);
-      });
+    const pendingPlayback: PendingPlayback = {
+      resolve: playbackResolve,
+      reject: playbackReject,
+      pendingChunks: 0,
+      streamEnded: false,
+    };
+
+    this.pendingPlaybacks.set(audioId, pendingPlayback);
+
+    let onAbort: (() => void) | undefined;
+    const destroyStream = () => {
+      if (typeof stream.destroy === "function" && !stream.destroyed) {
+        stream.destroy();
+      }
+    };
+
+    onAbort = () => {
+      console.log(
+        `[TTS-Manager ${this.sessionId}] Aborted while waiting for playback`
+      );
+      pendingPlayback.streamEnded = true;
+      pendingPlayback.pendingChunks = 0;
+      this.pendingPlaybacks.delete(audioId);
+      playbackResolve();
+      destroyStream();
+    };
+
+    abortSignal.addEventListener("abort", onAbort, { once: true });
+
+    try {
+      const iterator = stream[Symbol.asyncIterator]();
+      let chunkIndex = 0;
+      let current = await iterator.next();
+
+      if (!current.done) {
+        let next = await iterator.next();
+
+        while (true) {
+          if (abortSignal.aborted) {
+            console.log(
+              `[TTS-Manager ${this.sessionId}] Aborted during stream emission`
+            );
+            break;
+          }
+
+          const chunkBuffer = Buffer.isBuffer(current.value)
+            ? current.value
+            : Buffer.from(current.value);
+
+          const chunkId = `${audioId}:${chunkIndex}`;
+          pendingPlayback.pendingChunks += 1;
+
+          emitMessage({
+            type: "audio_output",
+            payload: {
+              id: chunkId,
+              groupId: audioId,
+              chunkIndex,
+              isLastChunk: next.done,
+              audio: chunkBuffer.toString("base64"),
+              format,
+              isRealtimeMode,
+            },
+          });
+
+          console.log(
+            `[TTS-Manager ${this.sessionId}] ${new Date().toISOString()} Sent audio chunk ${chunkId}${
+              next.done ? " (last)" : ""
+            }`
+          );
+
+          chunkIndex += 1;
+
+          if (next.done) {
+            break;
+          }
+
+          current = next;
+          next = await iterator.next();
+        }
+      }
+
+      pendingPlayback.streamEnded = true;
+
+      if (pendingPlayback.pendingChunks === 0) {
+        this.pendingPlaybacks.delete(audioId);
+        playbackResolve();
+      }
+
+      await playbackPromise;
+    } catch (error) {
+      if (abortSignal.aborted) {
+        console.log(
+          `[TTS-Manager ${this.sessionId}] Audio stream closed after abort`
+        );
+      } else {
+        console.error(
+          `[TTS-Manager ${this.sessionId}] Error streaming audio`,
+          error
+        );
+        this.pendingPlaybacks.delete(audioId);
+        pendingPlayback.reject(error as Error);
+        throw error;
+      }
+    } finally {
+      if (onAbort) {
+        abortSignal.removeEventListener("abort", onAbort);
+      }
+      destroyStream();
    }

-    // Emit audio output message (include mode for drift protection)
-    emitMessage({
-      type: "audio_output",
-      payload: {
-        id: audioId,
-        audio: audio.toString("base64"),
-        format,
-        isRealtimeMode,
-      },
-    });
-
-    console.log(
-      `[TTS-Manager ${
-        this.sessionId
-      }] ${new Date().toISOString()} Sent audio ${audioId}, waiting for playback...`
-    );
-
-    // Wait for playback confirmation
-    await playbackPromise;
+    if (abortSignal.aborted) {
+      return;
+    }

    console.log(
      `[TTS-Manager ${
@@ -110,19 +183,25 @@ export class TTSManager {
   * Called when client confirms audio playback completed
   * Resolves the corresponding promise
   */
-  public confirmAudioPlayed(audioId: string): void {
+  public confirmAudioPlayed(chunkId: string): void {
+    const [audioId] = chunkId.includes(":")
+      ? chunkId.split(":")
+      : [chunkId];
    const pending = this.pendingPlaybacks.get(audioId);

    if (!pending) {
      console.warn(
-        `[TTS-Manager ${this.sessionId}] Received confirmation for unknown audio ID: ${audioId}`
+        `[TTS-Manager ${this.sessionId}] Received confirmation for unknown audio ID: ${chunkId}`
      );
      return;
    }

-    // Resolve promise and cleanup
-    pending.resolve();
-    this.pendingPlaybacks.delete(audioId);
+    pending.pendingChunks = Math.max(0, pending.pendingChunks - 1);
+
+    if (pending.pendingChunks === 0 && pending.streamEnded) {
+      pending.resolve();
+      this.pendingPlaybacks.delete(audioId);
+    }
  }

  /**
@@ -8,8 +8,8 @@ export interface TTSConfig {
  responseFormat?: "mp3" | "opus" | "aac" | "flac" | "wav" | "pcm";
 }

-export interface SpeechResult {
-  audio: Buffer;
+export interface SpeechStreamResult {
+  stream: Readable;
  format: string;
 }

@@ -31,7 +31,9 @@ export function initializeTTS(ttsConfig: TTSConfig): void {
  );
 }

-export async function synthesizeSpeech(text: string): Promise<SpeechResult> {
+export async function synthesizeSpeech(
+  text: string
+): Promise<SpeechStreamResult> {
  if (!openaiClient || !config) {
    throw new Error("TTS not initialized. Call initializeTTS() first.");
  }
@@ -58,18 +60,13 @@ export async function synthesizeSpeech(text: string): Promise<SpeechResult> {
      response_format: config.responseFormat as "mp3" | "opus" | "aac" | "flac" | "wav" | "pcm",
    });

-    // Response is a stream, convert to buffer
-    const audioBuffer = await streamToBuffer(
-      response.body as unknown as Readable
-    );
+    const audioStream = response.body as unknown as Readable;

    const duration = Date.now() - startTime;
-    console.log(
-      `[TTS] Speech synthesis complete in ${duration}ms (${audioBuffer.length} bytes)`
-    );
+    console.log(`[TTS] Speech synthesis stream ready in ${duration}ms`);

    return {
-      audio: audioBuffer,
+      stream: audioStream,
      format: config.responseFormat || "mp3",
    };
  } catch (error: any) {
@@ -78,24 +75,6 @@ export async function synthesizeSpeech(text: string): Promise<SpeechResult> {
  }
 }

-async function streamToBuffer(stream: Readable): Promise<Buffer> {
-  const chunks: Buffer[] = [];
-
-  return new Promise((resolve, reject) => {
-    stream.on("data", (chunk) => {
-      chunks.push(Buffer.from(chunk));
-    });
-
-    stream.on("end", () => {
-      resolve(Buffer.concat(chunks));
-    });
-
-    stream.on("error", (error) => {
-      reject(error);
-    });
-  });
-}
-
 export function isTTSInitialized(): boolean {
  return openaiClient !== null && config !== null;
 }
@@ -472,6 +472,9 @@ export const AudioOutputMessageSchema = z.object({
    format: z.string(),
    id: z.string(),
    isRealtimeMode: z.boolean(), // Mode when audio was generated (for drift protection)
+    groupId: z.string().optional(), // Logical utterance id
+    chunkIndex: z.number().int().nonnegative().optional(),
+    isLastChunk: z.boolean().optional(),
  }),
 });