From ca894d49a5e22bd2f5e096fd56048f6bc8b4a85d Mon Sep 17 00:00:00 2001 From: "paseo-bot[bot]" <266920839+paseo-bot[bot]@users.noreply.github.com> Date: Fri, 23 Oct 2026 23:01:02 +0200 Subject: [PATCH] fix(voice): keep prefetched sentences available for playback (#7756) Unread speech bodies lost their owning HTTP response while earlier sentences played. Garbage collection then cancelled those bodies and the playback manager skipped their audio. Consume each response inside the provider before returning its audio stream, matching the existing complete-clip playback. Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com> Co-authored-by: grueneisfabian <174708694+grueneisfabian@users.noreply.github.com> --- .../openai/test-utils/prefetched-playback.ts | 70 +++++++++++++++++++ .../providers/openai/tts-playback.test.ts | 21 ++++++ .../src/server/speech/providers/openai/tts.ts | 4 +- 3 files changed, 94 insertions(+), 1 deletion(-) create mode 100644 packages/server/src/server/speech/providers/openai/test-utils/prefetched-playback.ts create mode 100644 packages/server/src/server/speech/providers/openai/tts-playback.test.ts 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");