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>

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<void>((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<void>((resolve, reject) => {
+    server.close((error) => {
+      if (error) {
+        reject(error);
+        return;
+      }
+      resolve();
+    });
+  });
+}

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" },
+  ]);
+});

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");