diff --git a/packages/server/src/server/speech/providers/openai/test-utils/prefetched-playback.ts b/packages/server/src/server/speech/providers/openai/test-utils/prefetched-playback.ts new file mode 100644 index 0000000000..406d3056f1 --- /dev/null +++ b/packages/server/src/server/speech/providers/openai/test-utils/prefetched-playback.ts @@ -0,0 +1,70 @@ +import { createServer } from "node:http"; +import { setTimeout as delay } from "node:timers/promises"; +import pino from "pino"; +import { OpenAITTS } from "../tts.js"; +import { TTSManager } from "../../../../agent/tts-manager.js"; + +if (!global.gc) throw new Error("Run with --expose-gc"); + +const sentences = ["First sentence.", "Second sentence.", "Third sentence."]; +const audio = sentences.map((_, index) => Buffer.alloc(192_000, index + 1)); +const requests: string[] = []; +const server = createServer(async (request, response) => { + const chunks: Buffer[] = []; + for await (const chunk of request) chunks.push(Buffer.from(chunk)); + const { input } = JSON.parse(Buffer.concat(chunks).toString()) as { input: string }; + requests.push(input); + const body = audio[sentences.indexOf(input)]; + response.writeHead(200, { "Content-Type": "audio/pcm", "Content-Length": body.length }); + response.end(body); +}); + +await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); +const address = server.address(); +if (!address || typeof address === "string") throw new Error("Missing server address"); +const logger = pino({ level: "silent" }); +const provider = new OpenAITTS( + { apiKey: "test", baseUrl: `http://127.0.0.1:${address.port}/v1` }, + logger, +); +const manager = new TTSManager("prefetched-playback", logger, provider); +const emitted: { bytes: number; value: number; format: string }[] = []; + +try { + await manager.generateAndWaitForPlayback( + sentences.join(" "), + (message) => { + if (message.type !== "audio_output") return; + const buffer = Buffer.from(message.payload.audio, "base64"); + emitted.push({ bytes: buffer.length, value: buffer[0], format: message.payload.format }); + if (emitted.length === 1) { + // Hold the playback acknowledgement while the other responses are prefetched. + void (async () => { + while (requests.length < sentences.length) await delay(10); + for (let cycle = 0; cycle < 20; cycle++) { + await delay(25); + global.gc!(); + } + manager.confirmAudioPlayed(message.payload.id); + })(); + } else { + setImmediate(() => manager.confirmAudioPlayed(message.payload.id)); + } + }, + new AbortController().signal, + true, + ); + console.log(JSON.stringify({ requests, emitted })); +} finally { + manager.cleanup(); + server.closeAllConnections(); + await new Promise((resolve, reject) => { + server.close((error) => { + if (error) { + reject(error); + return; + } + resolve(); + }); + }); +} diff --git a/packages/server/src/server/speech/providers/openai/tts-playback.test.ts b/packages/server/src/server/speech/providers/openai/tts-playback.test.ts new file mode 100644 index 0000000000..b7b2224144 --- /dev/null +++ b/packages/server/src/server/speech/providers/openai/tts-playback.test.ts @@ -0,0 +1,21 @@ +import { execFileSync } from "node:child_process"; +import { fileURLToPath } from "node:url"; +import { expect, test } from "vitest"; + +test("prefetched speech survives garbage collection until its playback turn", () => { + const fixture = fileURLToPath(new URL("./test-utils/prefetched-playback.ts", import.meta.url)); + const output = execFileSync(process.execPath, ["--expose-gc", "--import", "tsx", fixture], { + encoding: "utf8", + timeout: 15_000, + }); + + const result = JSON.parse(output); + expect(result.requests.sort()).toEqual( + ["First sentence.", "Second sentence.", "Third sentence."].sort(), + ); + expect(result.emitted).toEqual([ + { bytes: 192_000, value: 1, format: "pcm" }, + { bytes: 192_000, value: 2, format: "pcm" }, + { bytes: 192_000, value: 3, format: "pcm" }, + ]); +}); diff --git a/packages/server/src/server/speech/providers/openai/tts.ts b/packages/server/src/server/speech/providers/openai/tts.ts index 31f7b5b263..49f43137ca 100644 --- a/packages/server/src/server/speech/providers/openai/tts.ts +++ b/packages/server/src/server/speech/providers/openai/tts.ts @@ -67,7 +67,9 @@ export class OpenAITTS implements TextToSpeechProvider { | "pcm", }); - const audioStream = response.body as unknown as Readable; + // Prefetched speech can wait through playback of an earlier sentence. Consume + // the response here so fetch cannot collect it and cancel its unread body. + const audioStream = Readable.from([Buffer.from(await response.arrayBuffer())]); const duration = Date.now() - startTime; this.logger.debug({ duration }, "Speech synthesis stream ready");