livekit-plugins-fishaudio: drop startup prebuffer

.changeset/fishaudio-drop-startup-prebuffer.md

@@ -0,0 +1,5 @@
+---
+'@livekit/agents-plugin-fishaudio': patch
+---
+
+Drop Fish Audio's startup audio prebuffer so streaming audio starts from the opening chunk.

plugins/fishaudio/src/tts.test.ts

@@ -1,19 +1,123 @@
 // SPDX-FileCopyrightText: 2026 LiveKit, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
+import { tts } from '@livekit/agents';
 import { STT } from '@livekit/agents-plugin-openai';
-import { tts } from '@livekit/agents-plugins-test';
-import { describe, it } from 'vitest';
+import { tts as testTts } from '@livekit/agents-plugins-test';
+import { decode, encode } from '@msgpack/msgpack';
+import { once } from 'node:events';
+import type { AddressInfo } from 'node:net';
+import { describe, expect, it } from 'vitest';
+import { type WebSocket, WebSocketServer } from 'ws';
 import { TTS } from './tts.js';
 
+async function startWebSocketServer() {
+  const wss = new WebSocketServer({ host: '127.0.0.1', port: 0 });
+  await once(wss, 'listening');
+  const address = wss.address() as AddressInfo;
+  return { wss, baseURL: `http://127.0.0.1:${address.port}` };
+}
+
+async function closeWebSocketServer(wss: WebSocketServer): Promise<void> {
+  for (const client of wss.clients) {
+    client.close();
+  }
+  await new Promise<void>((resolve) => wss.close(() => resolve()));
+}
+
+async function waitFor<T>(promise: Promise<T>, timeoutMs = 1000): Promise<T> {
+  let timeout: ReturnType<typeof setTimeout> | undefined;
+  try {
+    return await Promise.race([
+      promise,
+      new Promise<never>((_, reject) => {
+        timeout = setTimeout(() => reject(new Error('timed out waiting for promise')), timeoutMs);
+      }),
+    ]);
+  } finally {
+    if (timeout) clearTimeout(timeout);
+  }
+}
+
+async function startSynthesis(wss: WebSocketServer) {
+  const stopReceived = new Promise<WebSocket>((resolve) => {
+    wss.on('connection', (ws) => {
+      ws.on('message', (raw) => {
+        const message = decode(Buffer.from(raw as ArrayBuffer)) as Record<string, unknown>;
+        if (message.event === 'stop') resolve(ws);
+      });
+    });
+  });
+
+  const fishAudio = new TTS({
+    apiKey: 'test-key',
+    baseURL: `http://127.0.0.1:${(wss.address() as AddressInfo).port}`,
+  });
+  const stream = fishAudio.stream();
+  stream.pushText('hello world.');
+  stream.endInput();
+
+  return { fishAudio, stream, ws: await waitFor(stopReceived) };
+}
+
 const hasFishAudioConfig = Boolean(process.env.FISH_API_KEY && process.env.OPENAI_API_KEY);
 
 if (hasFishAudioConfig) {
   describe('FishAudio', async () => {
-    await tts(new TTS(), new STT());
+    await testTts(new TTS(), new STT({ useRealtime: false }));
   });
 } else {
   describe('FishAudio', () => {
     it.skip('requires FISH_API_KEY and OPENAI_API_KEY', () => {});
   });
 }
+
+describe('FishAudio streaming', () => {
+  it('emits the first complete frame before the next provider event', async () => {
+    const { wss } = await startWebSocketServer();
+    const { fishAudio, stream, ws } = await startSynthesis(wss);
+
+    try {
+      ws.send(encode({ event: 'audio', audio: Buffer.alloc(4800) }));
+
+      const first = await waitFor(stream.next());
+      expect(first.done).toBe(false);
+      expect(first.value).not.toBe(tts.SynthesizeStream.END_OF_STREAM);
+      if (first.value !== tts.SynthesizeStream.END_OF_STREAM) {
+        expect(first.value.frame.samplesPerChannel).toBe(2400);
+        expect(first.value.final).toBe(false);
+      }
+
+      ws.send(encode({ event: 'finish', reason: 'stop' }));
+      for await (const _event of stream) {
+        // Drain the stream so its tasks finish before cleanup.
+      }
+    } finally {
+      stream.close();
+      await fishAudio.close();
+      await closeWebSocketServer(wss);
+    }
+  });
+
+  it('marks the terminal frame final before ending the stream', async () => {
+    const { wss } = await startWebSocketServer();
+    const { fishAudio, stream, ws } = await startSynthesis(wss);
+
+    try {
+      ws.send(encode({ event: 'audio', audio: Buffer.alloc(4800) }));
+      ws.send(encode({ event: 'finish', reason: 'stop' }));
+
+      const events: tts.SynthesizedAudio[] = [];
+      for await (const event of stream) {
+        if (event !== tts.SynthesizeStream.END_OF_STREAM) events.push(event);
+      }
+
+      expect(events.length).toBeGreaterThan(0);
+      expect(events.at(-1)?.final).toBe(true);
+    } finally {
+      stream.close();
+      await fishAudio.close();
+      await closeWebSocketServer(wss);
+    }
+  });
+});

plugins/fishaudio/src/tts.ts

@@ -12,7 +12,6 @@ import {
   tokenize,
   tts,
 } from '@livekit/agents';
-import type { AudioFrame } from '@livekit/rtc-node';
 import { decode, encode } from '@msgpack/msgpack';
 import { request } from 'node:https';
 import { type RawData, WebSocket } from 'ws';
@@ -399,14 +398,6 @@ export class SynthesizeStream extends tts.SynthesizeStream {
       }
     };
 
-    let lastFrame: AudioFrame | undefined;
-    const sendLastFrame = (final: boolean) => {
-      if (lastFrame) {
-        this.queue.put({ requestId, segmentId: requestId, frame: lastFrame, final });
-        lastFrame = undefined;
-      }
-    };
-
     const recvTask = async () => {
       // No per-receive timeout: Fish has natural inter-sentence gaps that can
       // exceed connOptions.timeoutMs when the LLM is slow.
@@ -432,9 +423,8 @@ export class SynthesizeStream extends tts.SynthesizeStream {
         if (event === 'audio') {
           const audio = parsed.audio as Uint8Array | undefined;
           if (audio && audio.byteLength > 0) {
-            for (const f of bstream.write(audio)) {
-              sendLastFrame(false);
-              lastFrame = f;
+            for (const frame of bstream.write(audio)) {
+              this.queue.put({ requestId, segmentId: requestId, frame, final: false });
             }
           }
         } else if (event === 'finish') {
@@ -448,11 +438,15 @@ export class SynthesizeStream extends tts.SynthesizeStream {
             );
             return;
           }
-          for (const f of bstream.flush()) {
-            sendLastFrame(false);
-            lastFrame = f;
+          const remainingFrames = [...bstream.flush()];
+          for (const [idx, frame] of remainingFrames.entries()) {
+            this.queue.put({
+              requestId,
+              segmentId: requestId,
+              frame,
+              final: idx === remainingFrames.length - 1,
+            });
           }
-          sendLastFrame(true);
           if (!this.queue.closed) {
             this.queue.put(SynthesizeStream.END_OF_STREAM);
           }