diff --git a/.changeset/fishaudio-drop-startup-prebuffer.md b/.changeset/fishaudio-drop-startup-prebuffer.md new file mode 100644 index 000000000..e113846df --- /dev/null +++ b/.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. diff --git a/plugins/fishaudio/src/tts.test.ts b/plugins/fishaudio/src/tts.test.ts index b1149b8e5..126d19972 100644 --- a/plugins/fishaudio/src/tts.test.ts +++ b/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 { + for (const client of wss.clients) { + client.close(); + } + await new Promise((resolve) => wss.close(() => resolve())); +} + +async function waitFor(promise: Promise, timeoutMs = 1000): Promise { + let timeout: ReturnType | undefined; + try { + return await Promise.race([ + promise, + new Promise((_, 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((resolve) => { + wss.on('connection', (ws) => { + ws.on('message', (raw) => { + const message = decode(Buffer.from(raw as ArrayBuffer)) as Record; + 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); + } + }); +}); diff --git a/plugins/fishaudio/src/tts.ts b/plugins/fishaudio/src/tts.ts index 6bae25dd8..1b47914bf 100644 --- a/plugins/fishaudio/src/tts.ts +++ b/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); }