fix(deepgram): make Flux TTS retries and batch failures behave - #2524
tinalenguyen wants to merge 1 commit into
Conversation
Three faults in the Flux TTS (/v2/speak) client, each turning a recoverable
failure into silence or a wrong result.
A streaming retry lost the utterance. SynthesizeStream.input is created once
and never reset between attempts, so text consumed by attempt 1 was gone by
attempt 2: the retry sent nothing and the stream ended successfully having
spoken nothing. Each flushed segment's words are now buffered in a
ReplaySegment and replayed from the start, and the reader of `input` runs
once for the life of the stream rather than once per attempt, so a retry can
no longer race a leftover reader for the caller's text. A segment that had
already begun playing is not replayed, mirroring the Python base class's
refusal to retry once output_emitter.pushed_duration() > 0.
Python does this buffering in its base SynthesizeStream, which hands each
attempt a fresh channel replayed from _input_buffer; the JS base class has no
equivalent, so this port carries it in the plugin. Every other JS streaming
TTS plugin has the same exposure and is left alone here.
The batch path always dialled TLS. `request` came from node:https
unconditionally, so any non-https baseUrl -- the form the streaming path
already accepts, and the one the tests use -- became a TLS handshake against
a plain HTTP port. The transport now follows the URL scheme, with ws:// and
wss:// normalized, matching what _to_deepgram_url does in Python.
A truncated batch response was reported as success. Node signals a severed
body as Error('aborted'), the same message a caller-side abort produces; that
was discarded, and the following `close` flushed the partial buffer and
resolved. Completion is now driven by `end`, and a body that stops early
raises a connection error -- as aiohttp does in Python, where a short read
surfaces as ClientPayloadError. It is retryable only when no audio has been
emitted yet, since the base ChunkedStream forwards a failed attempt's frames
to the consumer and retrying after partial audio would splice two synthesis
runs together.
Each of the five new tests was checked against the unfixed file and fails
there. Also stops the `abandoned queue reads` tests resolving @livekit/agents
inside a test body, which cost ~3s cold and pushed whichever test ran first
past vitest's 5s default under load.
🦋 Changeset detectedLatest commit: dc53492 The changes in this PR will be included in the next version bump. This PR includes changesets to release 39 packages
Not sure what this means? Click here to learn what changesets are. Click here if you're a maintainer who wants to add another changeset to this PR |
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
| if (this.#segmentEmitted > 0 && e instanceof APIError && e.retryable) { | ||
| throw new APIConnectionError({ | ||
| message: `${asError(e).message} (not retried: part of this segment had already played)`, | ||
| options: { retryable: false }, | ||
| }); | ||
| } |
There was a problem hiding this comment.
🟡 WebSocket failures replay partial speech
After audio emission, a raw WebSocket error bypasses #segmentEmitted because the guard accepts only APIError. The outer handler makes it retryable, so the next attempt repeats the segment.
Learn more
The receive task rejects with the raw error from the WebSocket error event. The inner catch only applies the partial-audio guard to APIError, so raw transport errors pass through. The outer catch then converts every non-API error into a retryable APIConnectionError. The base retry loop re-enters run(), where the pending segment is replayed from its first word.
Example: A segment emits 200 ms of audio, then the socket raises ECONNRESET. #segmentEmitted is positive, but the raw error bypasses this condition. The retry sends the whole segment again, repeating its opening audio.
Recommended fix: Apply the non-retryable wrapper to raw errors when #segmentEmitted > 0, while preserving already non-retryable APIError instances.
| if (this.#segmentEmitted > 0 && e instanceof APIError && e.retryable) { | |
| throw new APIConnectionError({ | |
| message: `${asError(e).message} (not retried: part of this segment had already played)`, | |
| options: { retryable: false }, | |
| }); | |
| } | |
| if ( | |
| this.#segmentEmitted > 0 && | |
| (!(e instanceof APIError) || e.retryable) | |
| ) { | |
| throw new APIConnectionError({ | |
| message: `${asError(e).message} (not retried: part of this segment had already played)`, | |
| options: { retryable: false }, | |
| }); | |
| } |
Was this helpful? React with 👍 or 👎 to provide feedback.
| let emitted = 0; | ||
| let bodyComplete = false; | ||
| const truncatedError = () => | ||
| new APIConnectionError({ | ||
| message: 'Deepgram Flux TTS response ended before the full body arrived', | ||
| options: { retryable: emitted === 0 }, | ||
| }); |
There was a problem hiding this comment.
🟡 Batch failures splice duplicate audio
After audio emission, emitted disables retries only for truncatedError; timeouts and other transport errors remain retryable. The base retry loop forwards both attempts, splicing partial and complete synthesis.
Learn more
ChunkedStream creates a fresh queue for each attempt but forwards every queue to the consumer, including failed attempts. This method makes only the synthesized truncated-body error conditional on emitted. APITimeoutError remains retryable, and ordinary request or response errors are wrapped in a retryable APIConnectionError. Any of those failures after at least one frame has escaped can therefore start a complete second synthesis after the partial first synthesis.
Example: Deepgram returns 300 ms of audio and then stalls for 30 seconds. The timeout destroys the request with a retryable APITimeoutError. The consumer receives the 300 ms already forwarded, followed by the entire utterance from the retry.
Recommended fix: Centralize batch transport-error conversion so every timeout, request error, response error, and premature close sets retryable: emitted === 0. Preserve non-retryable status errors and caller cancellation.
Was this helpful? React with 👍 or 👎 to provide feedback.
1 participant
Description
Three faults in the Flux TTS (
/v2/speak) client, each of which turns a recoverable failure into silence or a wrong result. All three are onmaintoday and all three are invisible to the current tests, because the batch path has no coverage at all and the streaming tests never exercisemaxRetry > 0.I found these while reviewing #2514 and reproduced each one before fixing it.
1. A streaming retry lost the utterance
SynthesizeStream.inputis a singleAsyncIterableQueuecreated once in the base-class constructor and never reset between retry attempts (agents/src/tts/tts.ts:300), while the retry loop simply callsrun()again (:386). Text consumed by attempt 1 is therefore gone by attempt 2:tokenizeInputfound a drained, closed queue, exited immediately, and the stream ended successfully having spoken nothing — no audio, noEND_OF_STREAM, no error, and no second connection.Reproduced with a server that returns a retryable
Erroronce:Each flushed segment's words are now buffered in a
ReplaySegmentand replayed from the start, and the reader ofinputruns once for the life of the stream rather than once per attempt, so a retry can no longer race a leftover reader for the caller's text.A segment that had already begun playing is not replayed — repeating its opening words to the listener is worse than failing the turn.
2. The batch path always dialled TLS
requestwas imported fromnode:httpsunconditionally, so any non-httpsbaseUrl— the form the streaming path already accepts (baseUrl.replace(/^http/, 'ws')) and the one the plugin's own tests use — became a TLS handshake against a plain HTTP port. The transport now follows the URL scheme, withws:///wss://normalized.3. A truncated batch response was reported as success
Node signals a severed response body as
Error('aborted')with codeECONNRESET— the same message a caller-side abort produces — so theerr.message === 'aborted'filter discarded a real transport failure, and the followingclosehandler flushed the partial buffer and resolved.Observed event order for a body cut short, and the result:
Completion is now driven by
end, and a body that stops early raises a connection error. It is retryable only when no audio has been emitted yet: the baseChunkedStreamforwards a failed attempt's frames to the consumer (drainAttemptQueue), so retrying after partial audio would splice two synthesis runs together. Worth knowing that the retry machinery itself was healthy — a clean 503 already recovered correctly — so this bug was purely about bypassing it.Parity with Python
Checked against
livekit-plugins/livekit-plugins-deepgram/.../tts_v2.pyandlivekit-agents/.../tts/tts.pyonlivekit/agents@origin/main:_to_deepgram_url(..., websocket=False)rewritesws→httpand aiohttp picks the transport from the scheme. Matches.resp.content.iter_chunks()raisesClientPayloadErroron a short read, which becomes anAPIConnectionError. Python never reports a truncated body as success either. Matches.SynthesizeStream—_input_bufferrecords every pushed event, and on retry a fresh_input_chis replayed from it (tts.py:654-659). That is why the Python plugin's per-attempt_tokenize_inputis correct. The JS base class has no equivalent, so this port carries the buffering in the plugin.should_retryrequirespushed_duration == 0.0and logs "TTS failed after partial audio was already sent to the user, skip retrying." Mirrored here, but scoped to the current segment rather than the whole stream — narrower than Python's check, and safe because completed segments have already been dropped.Known gap, deliberately out of scope
The root cause of #1 is the JS base class, not this plugin:
SynthesizeStreamnever resetsinputbetween attempts.plugins/cartesia/src/tts.tshas the same exposure — I confirmed it loses the caller's text on retry too, sending only the empty terminator on every attempt:Against the real Cartesia service, which rejects an empty transcript 4xx (treated as non-fatal at
tts.ts:566), that lands in the same place: a silent, empty, apparently-successful turn. Fixing the JSSynthesizeStreamretry contract to match Python's_input_bufferwould fix every streaming plugin at once. I've kept this PR plugin-local rather than changing shared code every TTS plugin depends on; happy to open a follow-up issue with the repro.Changes Made
plugins/deepgram/src/tts_v2.ts—ReplaySegmentplus a stream-lifetime input task; scheme-aware batch transport;end-driven batch completion with conditional retryability; partial-audio guard.plugins/deepgram/src/tts_v2.test.ts— five new tests: segment replay after a retryable failure, no replay once audio has played, plain-httpbaseUrl, truncation surfaced as an error, and truncation-before-any-audio retried.plugins/deepgram/src/_utils.test.ts— theabandoned queue readstests resolved@livekit/agentsinside the test body, which cost ~3s on a cold cache and pushed whichever test ran first past vitest's 5s default under load. It fails onmainin a full-suite run and passes in isolation. Moved into abeforeAllwith its own timeout: 3441ms → 1ms. It stays a dynamic import because a top-level one resolves down a path this package cannot satisfy.patch).Testing
Each of the five new tests was run against the unfixed
tts_v2.tsand fails there, so they detect the regressions they exist to catch rather than passing vacuously.pnpm vitest run plugins/deepgram— 31 passed, 4 skipped (the skips are the API-key-gated live blocks)pnpm build— 40/40pnpm --filter @livekit/agents-plugin-deepgram api:check— clean; nothing new is exported,ReplaySegmentis module-localpnpm lint,pnpm format:check— clean on the touched files (the 11 remaining warnings are pre-existing, in_utils.tsandstt.ts)Note for anyone running the deepgram tests locally:
mainadded@opentelemetry/exporter-metrics-otlp-prototoagents/package.json, and without a freshpnpm installevery test in this plugin fails at module load. That is likely what was meant by "the tests in this plugin do not run in my environment" in #2470.restaurant_agent.ts/realtime_agent.ts— not run; neither imports@livekit/agents-plugin-deepgram, so neither exercises this change.