From 21e00a1409f4da39fa9bc795f0217708bc2e594f Mon Sep 17 00:00:00 2001 From: Toubat Date: Mon, 27 Jul 2026 22:57:35 -0700 Subject: [PATCH] fix(inference): treat a TTS gateway `done` as a generation boundary MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Several gateway providers answer one `session.flush` with more than one generation, splitting at roughly 1kB of text or 35s of audio and sending a `done` after each. A probe of `inworld/inworld-tts-2` saw 6 `done` events for a single flush, the first covering 39.4s of the 201.0s the session went on to produce. Stopping at the first `done` cut replies off after 20-80% of their audio while the transcript committed all of it, and released the websocket into the ConnectionPool mid-synthesis so the next reply spoke the previous one's leftover audio. The FIFO session is never reset, so each reply queued behind the last one's unsynthesized text and the lag compounded. Keep reading after `done` and end the flush only once the session has stayed quiet. The wait is capped by how much synthesized audio is still unplayed, so it is hidden behind playout and collapses to near zero on replies too short for the gateway to split. A terminal event that arrives during that wait is judged on what it actually proves. `session.closed` still ends the reply normally, but its socket is evicted instead of pooled: `session.create` is only ever sent when a socket is opened, so a pooled socket would keep the closed session and the next reply would stall inside it. An `error` no longer resolves the reply, because a `done` is only a candidate end and a provider that fails while preparing the next generation has left the reply unfinished; a genuinely finished reply is protected from that retry by the idle timeout, which resolves the run before a later error can be read. The cap on the wait is a known limitation: a gateway streaming at roughly playback speed leaves nothing buffered, so the wait collapses to about zero and the first `done` ends the flush. It is documented and pinned by a test rather than lengthened — the bound is what keeps this off the turn latency path, and the providers that do split a flush run far faster than realtime. Co-authored-by: Cursor --- .changeset/tts-multi-generation-flush.md | 24 ++ agents/src/inference/tts.ts | 156 ++++++-- .../inference/tts_multi_generation.test.ts | 352 ++++++++++++++++++ agents/src/inference/tts_pool_reuse.test.ts | 2 +- 4 files changed, 507 insertions(+), 27 deletions(-) create mode 100644 .changeset/tts-multi-generation-flush.md create mode 100644 agents/src/inference/tts_multi_generation.test.ts diff --git a/.changeset/tts-multi-generation-flush.md b/.changeset/tts-multi-generation-flush.md new file mode 100644 index 000000000..b44c732d9 --- /dev/null +++ b/.changeset/tts-multi-generation-flush.md @@ -0,0 +1,24 @@ +--- +'@livekit/agents': patch +--- + +fix(inference): don't stop TTS synthesis at the first gateway `done` + +Several inference gateway providers answer one `session.flush` with more than one generation, +splitting at roughly 1kB of text or 35s of audio and sending a `done` after each. A probe of +`inworld/inworld-tts-2` saw 6 `done` events for a single flush, the first covering 39.4s of +the 201.0s the session went on to produce. + +The client treated the first `done` as the end of synthesis, so a reply longer than one +generation was cut off after 20-80% of its audio while the transcript committed all of it. It +also released the websocket back into the `ConnectionPool` while the gateway was still +streaming, so the next reply picked up that socket and spoke the previous reply's leftover +audio. Because the gateway session is FIFO and never reset, each reply then queued behind the +last one's unsynthesized text and the lag compounded — in the trace this came from, replies +fell up to 113s behind and 9 of 11 were never audible. + +`done` is now treated as a generation boundary. The client keeps reading and only ends the +flush once the session has stayed quiet, which both delivers the whole reply and means a +socket is only ever recycled after it has been observed to be idle. The wait for silence is +capped by how much synthesized audio is still unplayed, so it costs nothing while audio is +playing out and collapses to near zero on replies too short for the gateway to split. diff --git a/agents/src/inference/tts.ts b/agents/src/inference/tts.ts index dcd972e16..8db8b6863 100644 --- a/agents/src/inference/tts.ts +++ b/agents/src/inference/tts.ts @@ -286,6 +286,29 @@ const DEFAULT_SAMPLE_RATE = 16000; const NUM_CHANNELS = 1; const DEFAULT_LANGUAGE = 'en'; +/** + * Longest a gateway session has to stay silent after a `done` before that `done` is accepted + * as the end of the flush. + * + * Several providers answer one `session.flush` with several generations, split at roughly 1kB + * of text or 35s of audio, and send a `done` after each. A probe of `inworld/inworld-tts-2` + * saw 6 `done` events for a single flush, the first covering 39.4s of the 201.0s the session + * went on to produce. Gaps between consecutive generations reached 1.26s, so the grace has to + * comfortably exceed that. + * + * Known limitation: this is only an upper bound. The wait actually taken is + * {@link SynthesizeStream.run}'s `drainTimeoutMs`, which caps it at how much synthesized audio + * is still unplayed, so a gateway that streams a generation at roughly playback speed leaves + * nothing buffered, the wait collapses to about zero, and the first `done` ends the flush — + * later generations are dropped and the socket can be recycled while the gateway is still + * streaming. The bound is deliberate: it is what keeps this off the turn latency path, and the + * providers observed to split a flush produce their audio far faster than realtime (~201s of + * audio per flush), so their buffer is deep and the protection does apply. Waiting longer than + * the buffer would cost every short reply real silence to cover a case those providers do not + * produce. `tts_multi_generation.test.ts` pins both cadences. + */ +const DRAIN_IDLE_TIMEOUT = 2000; + export interface InferenceTTSOptions { model?: TModel; voice?: string; @@ -581,14 +604,14 @@ export class SynthesizeStream extends BaseSynthesizeSt protected async run(): Promise { let closing = false; let lastFrame: AudioFrame | undefined; - // Only a `done` from the gateway proves the session owes us no more audio, and a socket - // recycled before that hands the leftover audio to whichever SynthesizeStream picks it - // up next. `session.closed` is the exit that reaches the pool: it returns from this run - // normally, so nothing else evicts the socket. The remaining non-`done` exits — a closed - // event channel, a swallowed abort — are only ever reached after `onClose` / `onAbort` - // has already removed the socket, so gating reuse on `done` is what keeps reuse tied to - // the one event that proves the session is drained rather than to each exit remembering. - let sessionDrained = false; + // Set only when this run ends on a session that has gone quiet after a `done` and is + // still open, which is the one state in which the socket can serve another reply: the + // session owes no more audio, and it still exists. A socket recycled any earlier hands + // the leftover audio to whichever SynthesizeStream picks it up next. The exits that + // return from this run normally are the ones that reach the pool, so nothing else evicts + // the socket; the remaining exits — a closed event channel, a swallowed abort — are only + // ever reached after `onClose` / `onAbort` has already removed it. + let socketReusable = false; // Timestamps are delivered in their own WS message; buffer them and attach // to the next audio frame that we forward to the output emitter. This // mirrors the semantics of `output_emitter.push_timed_transcript` on the @@ -623,8 +646,16 @@ export class SynthesizeStream extends BaseSynthesizeSt ws.send(JSON.stringify(validatedEvent)); }; + // Duration of audio already handed downstream, and when the first frame went out. + // Downstream plays out in real time, so the difference bounds how long this run can stall + // before the listener would hear silence. See `drainTimeoutMs` below. + let emittedAudioMs = 0; + let firstEmittedAt: number | undefined; + const sendLastFrame = (segmentId: string, final: boolean) => { if (lastFrame) { + firstEmittedAt ??= Date.now(); + emittedAudioMs += (lastFrame.samplesPerChannel / lastFrame.sampleRate) * 1000; this.queue.put({ requestId, segmentId, @@ -781,15 +812,72 @@ export class SynthesizeStream extends BaseSynthesizeSt const serverEventStream = eventChannel.stream(); const reader = serverEventStream.getReader(); + // Whether a `done` has arrived with nothing after it yet. `done` only marks the end of + // one gateway generation, and several providers split a single flushed utterance into + // more than one, so it cannot be taken as the end of the flush on its own. + let draining = false; + + /** + * How long to keep reading before accepting a `done` as the end of the flush. + * + * Bounded by the audio already handed downstream and not yet played, because that is + * exactly how long this run can stall for free: downstream finishes a segment at + * `max(END_OF_STREAM, playout end)`, so a wait no longer than the remaining playout adds + * nothing to when the turn completes. Short replies therefore keep their old latency, + * and only replies with seconds of audio behind them — the only ones the gateway ever + * splits — wait out the full grace. + * + * The bound is also the known limitation of this whole mechanism: at roughly playback + * speed `emittedAudioMs` grows as fast as wall time, so this returns about zero and the + * first `done` ends the flush. See {@link DRAIN_IDLE_TIMEOUT}. + */ + const drainTimeoutMs = () => { + const unplayedMs = + firstEmittedAt === undefined + ? 0 + : Math.max(0, emittedAudioMs - (Date.now() - firstEmittedAt)); + return Math.min(DRAIN_IDLE_TIMEOUT, unplayedMs); + }; + + /** + * End the flush: hand over the trailing audio, close the segment, release the socket. + * + * `reusable` is whether the session behind the socket can still serve another reply. + * Only the caller knows: a session that went quiet is idle and reusable, one the gateway + * has closed is finished with this reply but dead. + */ + const finalize = async ({ reusable }: { reusable: boolean }) => { + for (const frame of bstream.flush()) { + sendLastFrame(currentSessionId!, false); + lastFrame = frame; + } + sendLastFrame(currentSessionId!, true); + this.queue.put(SynthesizeStream.END_OF_STREAM); + socketReusable = reusable; + await resourceCleanup(); + completionFuture.resolve(); + }; + try { await inputSentEvent.wait(); while (!this.closed && !signal.aborted) { - const result = await waitUntilTimeout( - reader.read(), - recvTimeoutMs, - () => new APITimeoutError({ message: 'TTS recv idle timeout' }), - ); + let result: Awaited>; + try { + result = await waitUntilTimeout( + reader.read(), + draining ? drainTimeoutMs() : recvTimeoutMs, + () => new APITimeoutError({ message: 'TTS recv idle timeout' }), + ); + } catch (e) { + if (draining && e instanceof APITimeoutError) { + // The session produced nothing after its `done`: the flush really is finished, + // and the session is idle rather than gone, so the socket can be reused. + await finalize({ reusable: true }); + return; + } + throw e; + } if (signal.aborted) return; if (result.done) return; @@ -805,6 +893,11 @@ export class SynthesizeStream extends BaseSynthesizeSt currentSessionId = sessionIdFromEvent; } + const wasDraining = draining; + if (serverEvent.type !== 'done') { + draining = false; + } + switch (serverEvent.type) { case 'session.created': currentSessionId = serverEvent.session_id; @@ -840,19 +933,24 @@ export class SynthesizeStream extends BaseSynthesizeSt } break; case 'done': - for (const frame of bstream.flush()) { - sendLastFrame(currentSessionId!, false); - lastFrame = frame; - } - sendLastFrame(currentSessionId!, true); - this.queue.put(SynthesizeStream.END_OF_STREAM); - sessionDrained = true; - await resourceCleanup(); - completionFuture.resolve(); - return; + // Only a candidate end of the flush. Keep reading: if the gateway is merely + // between generations it resumes within a gap measured at up to 1.26s, and the + // read above finalizes once it has stayed quiet instead. + draining = true; + break; case 'session.closed': - // The gateway dropped the session before it finished the reply. Hand over - // the audio it did produce, then fail the attempt: Python has no + // After a `done` the flush is complete, so this is the gateway tearing down a + // session it has nothing left to say on. Finish normally — failing here would + // retry a reply the listener has already heard. The socket, though, is spent: + // `session.create` is only ever sent when a socket is opened, so a pooled + // socket keeps this closed session forever and the next reply would write its + // transcript and flush into it and stall until the receive timeout. + if (wasDraining) { + await finalize({ reusable: false }); + return; + } + // Otherwise the gateway dropped the session before it finished the reply. Hand + // over the audio it did produce, then fail the attempt: Python has no // `session.closed` branch at all, so the dropped session surfaces there as // a read timeout or a closed socket, i.e. an error that evicts the socket // and lets the retry machinery resynthesize what is left. Resolving here @@ -875,6 +973,12 @@ export class SynthesizeStream extends BaseSynthesizeSt { serverEvent }, 'Received error message from LiveKit TTS WebSocket', ); + // Unlike `session.closed`, an error is not itself evidence that the flush is + // over: a `done` is only a candidate end, so a provider that fails while + // preparing the next generation has left the reply unfinished. Fail the attempt + // and let the retry finish it. What guards a genuinely finished reply from + // being retried is the drain timeout above, which resolves this run and closes + // the event channel before a later error can be read at all. await resourceCleanup(); completionFuture.reject( new APIError(`LiveKit TTS returned error: ${serverEvent.message}`), @@ -953,7 +1057,7 @@ export class SynthesizeStream extends BaseSynthesizeSt await resourceCleanup(); await cancelAndWait(tasks, 5000); this.abortController.signal.removeEventListener('abort', onStreamAbort); - if (!sessionDrained) { + if (!socketReusable) { this.tts.pool.remove(ws); } } diff --git a/agents/src/inference/tts_multi_generation.test.ts b/agents/src/inference/tts_multi_generation.test.ts new file mode 100644 index 000000000..a40f62722 --- /dev/null +++ b/agents/src/inference/tts_multi_generation.test.ts @@ -0,0 +1,352 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import type { AddressInfo } from 'node:net'; +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { WebSocketServer } from 'ws'; +import type { WebSocket as WsSocket } from 'ws'; +import { initializeLogger } from '../log.js'; +import { TTS } from './tts.js'; + +/** + * Some inference gateway providers split a single flushed utterance into several generations + * and send a `done` after each one, all under the same session id. A websocket probe against + * `inworld/inworld-tts-2` answered one `session.flush` of 40 sentences with 6 `done` events, + * the first covering 39.4s of the 201.0s the session went on to produce. + * + * These tests hold the gateway to that shape and pin what the client has to do with it. + */ + +initializeLogger({ pretty: false }); + +const SAMPLE_RATE = 16000; +/** `AudioByteStream` frames at 100ms; keep the gateway's chunks frame-aligned. */ +const CHUNK_MS = 100; +const SAMPLES_PER_CHUNK = (SAMPLE_RATE * CHUNK_MS) / 1000; + +/** Each generation carries a distinct constant sample so a frame can be attributed to it. */ +const FIRST_REPLY_SAMPLES = [1000, 1001, 1002]; +const SECOND_REPLY_SAMPLE = 2000; + +const CHUNKS_PER_GENERATION = 20; +const GENERATION_AUDIO_MS = CHUNKS_PER_GENERATION * CHUNK_MS; + +/** Gap the gateway leaves between generations; probed inter-generation gaps reached 1.26s. */ +const GENERATION_GAP_MS = 300; + +/** + * Gap used by the playback-speed case. A gateway streaming at playback speed leaves at most a + * chunk or two unplayed, so the drain wait is only ever a few hundred milliseconds there; the + * gap has to sit clearly outside that to keep the test off the boundary. + */ +const REALTIME_GAP_MS = 1000; + +/** + * How long the gateway takes to answer the second reply's flush. It has to be longer than + * {@link GENERATION_GAP_MS} so that a client which released the socket at the first `done` + * reads the first reply's leftover generations while waiting for its own audio — which is + * exactly what happened in production. + */ +const SECOND_REPLY_START_DELAY_MS = 400; + +function audioEvent(sessionId: string, sample: number): string { + const pcm = Buffer.alloc(SAMPLES_PER_CHUNK * 2); + for (let i = 0; i < SAMPLES_PER_CHUNK; i++) { + pcm.writeInt16LE(sample, i * 2); + } + return JSON.stringify({ + type: 'output_audio', + session_id: sessionId, + audio: pcm.toString('base64'), + }); +} + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +/** + * What the gateway sends right after the last generation's `done`, while the client is still + * deciding whether that `done` ended the flush. + */ +type FlushTerminator = 'session.closed' | 'error'; + +interface FlushPlan { + /** One sample value per generation the gateway splits this flush into. */ + generations: number[]; + /** Delay before the first generation of this flush starts producing. */ + startDelayMs?: number; + /** Terminal event to send once the last generation's `done` is out. */ + after?: FlushTerminator; + /** Emit each chunk at playback speed instead of as a single burst. */ + paced?: boolean; + /** Gap between this flush's generations. Defaults to {@link GENERATION_GAP_MS}. */ + gapMs?: number; +} + +interface GatewayOptions { + /** What the gateway does with each successive `session.flush`, in order. */ + flushes: FlushPlan[]; + chunksPerGeneration?: number; +} + +/** + * Gateway stand-in that answers each `session.flush` with several generations on one session, + * emitting `done` after every generation and never resetting the session. + */ +async function startFakeGateway(options: GatewayOptions) { + const chunks = options.chunksPerGeneration ?? CHUNKS_PER_GENERATION; + const wss = new WebSocketServer({ port: 0, host: '127.0.0.1' }); + await new Promise((resolve) => wss.once('listening', () => resolve())); + + const sockets: WsSocket[] = []; + let connections = 0; + let flushCount = 0; + let staleFlushes = 0; + const flushedAt: number[] = []; + + wss.on('connection', (ws: WsSocket) => { + sockets.push(ws); + connections++; + const sessionId = `session-${connections}`; + let sessionClosed = false; + + const send = (payload: string) => { + if (ws.readyState === ws.OPEN) ws.send(payload); + }; + + ws.on('message', (raw: Buffer) => { + const event = JSON.parse(raw.toString()) as { type: string }; + + if (event.type === 'session.create') { + send(JSON.stringify({ type: 'session.created', session_id: sessionId })); + return; + } + + if (event.type !== 'session.flush') return; + + // A flush on a session the gateway already closed. The client only ever sends + // `session.create` when it opens a socket, so this can only be a pooled socket being + // reused: the transcript and the flush go into a session that is gone. + if (sessionClosed) staleFlushes++; + + const plan = options.flushes[flushCount++]; + flushedAt.push(Date.now()); + if (!plan) return; + + void (async () => { + if (plan.startDelayMs) await sleep(plan.startDelayMs); + for (const [index, sample] of plan.generations.entries()) { + if (index > 0) await sleep(plan.gapMs ?? GENERATION_GAP_MS); + for (let i = 0; i < chunks; i++) { + send(audioEvent(sessionId, sample)); + if (plan.paced) await sleep(CHUNK_MS); + } + // The session id never changes: every generation reports the same one, which is + // why the client cannot tell a boundary from an end by session id alone. + send(JSON.stringify({ type: 'done', session_id: sessionId })); + } + if (plan.after === 'session.closed') { + sessionClosed = true; + send(JSON.stringify({ type: 'session.closed', session_id: sessionId })); + } else if (plan.after === 'error') { + send(JSON.stringify({ type: 'error', message: 'provider failed mid-reply' })); + } + })(); + }); + }); + + const { port } = wss.address() as AddressInfo; + return { + baseURL: `http://127.0.0.1:${port}/v1`, + flushedAt, + get connections() { + return connections; + }, + get staleFlushes() { + return staleFlushes; + }, + close: () => { + for (const socket of sockets) socket.terminate(); + return new Promise((resolve) => wss.close(() => resolve())); + }, + }; +} + +function createTTS(baseURL: string) { + return new TTS({ + model: 'inworld/inworld-tts-2', + voice: 'Sarah', + sampleRate: SAMPLE_RATE, + baseURL, + apiKey: 'devkey', + apiSecret: 'secret'.padEnd(32, 'x'), + }); +} + +async function synthesize(tts: TTS, text: string) { + const stream = tts.stream(); + stream.pushText(text); + stream.endInput(); + + const samples: number[] = []; + let audioMs = 0; + for await (const event of stream) { + if (typeof event === 'symbol' || event.frame.samplesPerChannel === 0) continue; + samples.push(event.frame.data[0]!); + audioMs += (event.frame.samplesPerChannel / event.frame.sampleRate) * 1000; + } + await stream.close(); + return { samples: new Set(samples), audioMs, endedAt: Date.now() }; +} + +describe('inference TTS multi-generation flush', () => { + let gateway: Awaited>; + + afterEach(async () => { + await gateway.close(); + }); + + describe('with several generations per flush', () => { + beforeEach(async () => { + gateway = await startFakeGateway({ + flushes: [ + { generations: FIRST_REPLY_SAMPLES }, + { generations: [SECOND_REPLY_SAMPLE], startDelayMs: SECOND_REPLY_START_DELAY_MS }, + ], + }); + }); + + it('speaks every generation of the reply, not just the first', async () => { + const tts = createTTS(gateway.baseURL); + + const reply = await synthesize(tts, 'Tell me a long story about the lighthouse.'); + + // The reply is one utterance the gateway chose to synthesize in three passes. Stopping + // at the first `done` commits the whole transcript but speaks only a third of it. + expect([...reply.samples].sort()).toEqual(FIRST_REPLY_SAMPLES); + expect(reply.audioMs).toBe(FIRST_REPLY_SAMPLES.length * GENERATION_AUDIO_MS); + + await tts.close(); + }, 30_000); + + it('does not hand a later generation to the next reply', async () => { + const tts = createTTS(gateway.baseURL); + + await synthesize(tts, 'Tell me a long story about the lighthouse.'); + const second = await synthesize(tts, 'Now tell me a joke.'); + + // Releasing the socket at the first `done` puts it back in the pool while the gateway + // is still streaming, and the next reply reads that audio as its own. + expect(second.samples).toEqual(new Set([SECOND_REPLY_SAMPLE])); + expect(second.audioMs).toBe(GENERATION_AUDIO_MS); + + await tts.close(); + }, 30_000); + }); + + describe('with a reply short enough to finish in one generation', () => { + beforeEach(async () => { + gateway = await startFakeGateway({ + flushes: [{ generations: [FIRST_REPLY_SAMPLES[0]!] }], + chunksPerGeneration: 2, + }); + }); + + it('finalizes without waiting out the full idle grace', async () => { + const tts = createTTS(gateway.baseURL); + + const reply = await synthesize(tts, 'Sure.'); + + expect(reply.samples).toEqual(new Set([FIRST_REPLY_SAMPLES[0]])); + // Waiting for silence is only free while there is buffered audio left to play. With + // 200ms of audio behind it, the wait has to collapse to about that, not to the full + // idle grace, or the agent goes quiet at the end of every short reply. + expect(reply.endedAt - gateway.flushedAt[0]!).toBeLessThan(1000); + + await tts.close(); + }, 30_000); + }); + + describe('when the gateway closes the session after a `done`', () => { + beforeEach(async () => { + gateway = await startFakeGateway({ + flushes: [ + { generations: [FIRST_REPLY_SAMPLES[0]!], after: 'session.closed' }, + { generations: [SECOND_REPLY_SAMPLE] }, + ], + }); + }); + + it('finishes the reply but never returns the closed session to the pool', async () => { + const tts = createTTS(gateway.baseURL); + + // The `done` really was the end of the flush, so this reply is complete and must not + // be failed and retried. + const first = await synthesize(tts, 'Tell me a long story about the lighthouse.'); + expect(first.samples).toEqual(new Set([FIRST_REPLY_SAMPLES[0]])); + expect(first.audioMs).toBe(GENERATION_AUDIO_MS); + + const second = await synthesize(tts, 'Now tell me a joke.'); + expect(second.samples).toEqual(new Set([SECOND_REPLY_SAMPLE])); + + // `session.create` is only sent when a socket is opened, so a pooled socket keeps the + // session the gateway has already closed. Recycling it writes the next reply's + // transcript and flush into a session that is gone, and that reply stalls until the + // receive timeout. + expect(gateway.staleFlushes).toBe(0); + expect(gateway.connections).toBe(2); + + await tts.close(); + }, 30_000); + }); + + describe('when the gateway errors after a `done`', () => { + beforeEach(async () => { + gateway = await startFakeGateway({ + flushes: [ + { generations: [FIRST_REPLY_SAMPLES[0]!], after: 'error' }, + { generations: [SECOND_REPLY_SAMPLE] }, + ], + }); + }); + + it('fails the attempt instead of ending the reply at the boundary', async () => { + const tts = createTTS(gateway.baseURL); + + const reply = await synthesize(tts, 'Tell me a long story about the lighthouse.'); + + // A `done` is only a candidate end of the flush, so a provider that fails while + // preparing the next generation has left the reply unfinished. Treating that error as + // a completion truncates the reply silently and suppresses the retry that would + // finish it. + expect([...reply.samples].sort()).toEqual([FIRST_REPLY_SAMPLES[0], SECOND_REPLY_SAMPLE]); + expect(gateway.connections).toBe(2); + + await tts.close(); + }, 30_000); + }); + + describe('with generations streamed at playback speed', () => { + beforeEach(async () => { + gateway = await startFakeGateway({ + flushes: [{ generations: FIRST_REPLY_SAMPLES, paced: true, gapMs: REALTIME_GAP_MS }], + }); + }); + + it('ends the flush at the first `done` (known limitation)', async () => { + const tts = createTTS(gateway.baseURL); + + const reply = await synthesize(tts, 'Tell me a long story about the lighthouse.'); + + // Pinned, not desired: see `DRAIN_IDLE_TIMEOUT`. A gateway producing audio at roughly + // playback speed leaves nothing buffered, so the drain wait collapses to about zero + // and the first `done` ends the flush — the later generations are dropped. Lengthening + // the wait is not the answer: the bound is what keeps this fix off the turn latency + // path, and the providers that actually split a flush run far faster than realtime, so + // their buffer is deep and the protection does apply. + expect(reply.samples).toEqual(new Set([FIRST_REPLY_SAMPLES[0]])); + expect(reply.audioMs).toBe(GENERATION_AUDIO_MS); + + await tts.close(); + }, 30_000); + }); +}); diff --git a/agents/src/inference/tts_pool_reuse.test.ts b/agents/src/inference/tts_pool_reuse.test.ts index 93f9e5ced..cd181099a 100644 --- a/agents/src/inference/tts_pool_reuse.test.ts +++ b/agents/src/inference/tts_pool_reuse.test.ts @@ -15,7 +15,7 @@ import { TTS } from './tts.js'; * `session.closed` is the only route that reaches that leak. Once the `session.closed` * handler rejects the attempt instead of resolving it, the exception path in * `ConnectionPool.withConnection` evicts the socket by itself and this test passes with or - * without the `sessionDrained` eviction in `SynthesizeStream.run`. Read it as coverage of + * without the `socketReusable` eviction in `SynthesizeStream.run`. Read it as coverage of * the behaviour, not of that eviction. */