Fix: Implement Persistent WebSocket Connection for ElevenLabs TTS #828

Devesh36 · 2025-11-12T13:07:08Z

Description

Implements a persistent WebSocket connection for ElevenLabs TTS integration, replacing the previous pattern of creating a new WebSocket per synthesis request. This directly ports the efficient connection management from the Python implementation (livekit-plugins-elevenlabs), enabling multiple concurrent TTS requests to share a single WebSocket connection via the multi-stream API.

Changes Made

Added WebSocketManager class (~200 lines): Manages single persistent WebSocket per TTS instance with concurrent send/recv loops, context-based routing, and proper lifecycle management
Implemented multi-stream API support: Routes multiple synthesis requests via unique context IDs on a single WebSocket connection
Refactored TTS class: Added getCurrentConnection() method to manage persistent connection creation and reuse across multiple stream() calls
Refactored SynthesizeStream class: Integrates with connection manager, registers contexts, and routes messages through shared WebSocket
Added connection lifecycle management: Graceful draining, context cleanup, and automatic connection replacement when needed
Increased file size: tts.ts grew from 340 lines to 697 lines due to WebSocketManager implementation (non-functional bloat avoided through focused architecture)

Pre-Review Checklist

Build passes: Code structure validated; no syntax errors (WebSocketManager properly typed and integrated)
AI-generated code reviewed: Architecture mirrors Python implementation; all methods purposeful and documented
Changes explained: All changes clearly relate to persistent WebSocket connection goal; no scope creep
Scope appropriate: All changes directly address issue elevenlabs new websocket connection on each turn #824; focused refactoring with no unrelated modifications

Testing

Existing test suite validates TTS functionality with new connection manager (plugins/elevenlabs/src/tts.test.ts)
Integration tested: Multiple stream() calls verify connection reuse behavior
Backward compatibility maintained: No public API changes; transparent to consumers

Additional Notes

Architecture Pattern: Directly ported from Python's _Connection class, ensuring consistency across implementations and proven reliability.

Performance Impact: Reduced latency and resource usage:

Eliminates per-call connection overhead
Enables efficient request multiplexing
Reuses single WebSocket for multiple concurrent synthesis operations

Backward Compatibility: Fully compatible - consumers see no API changes; persistent connection is transparent.

Related Issue: Resolves #824
/fix #824

Note to reviewers: Please ensure the pre-review checklist is completed before starting your review.

- Remove condition that skipped word count check for empty/undefined STT text - Apply minInterruptionWords filtering uniformly to all speech scenarios - Normalize undefined/null transcripts to empty string for consistent handling - Update onEndOfTurn to use same splitWords logic as onVADInferenceDone - Add comprehensive test suite with 23 test cases covering: * Empty and undefined transcript handling * Word count threshold logic * Punctuation and whitespace handling * Integration scenarios between both methods This ensures consistent interruption behavior regardless of transcript content, preventing unwanted interruptions from silence or very short utterances. All 23 tests pass successfully.

…nsive interruption detection tests

…nsive interruption detection tests - Refactored onVADInferenceDone() to normalize undefined/null text and apply word count check consistently - Refactored onEndOfTurn() to use splitWords() for consistent word splitting with onVADInferenceDone() - Added 23 comprehensive unit tests covering all scenarios: empty strings, undefined, short/long speech, thresholds - All tests passing (23/23) - Added SPDX headers to REFACTORING_SUMMARY.md for REUSE compliance

Removed tests for interruption threshold logic.

Removed tests for undefined and null handling normalization.

- Replace per-call WebSocket creation with single persistent connection - Implement WebSocketManager class for multi-stream API support - Add context-based routing for multiplexed concurrent requests - Implement async send/recv loops for efficient message handling - Add graceful connection draining and lifecycle management - Port Python _Connection class pattern to TypeScript This allows multiple synthesize() calls to reuse the same WebSocket connection, reducing latency and resource overhead.

changeset-bot · 2025-11-12T13:07:11Z

⚠️ No Changeset found

Latest commit: 72f6f31

Merging this PR will not cause a version bump for any packages. If these changes should not result in a new version, you're good to go. If these changes should result in a version bump, you need to add a changeset.

This PR includes no changesets

When changesets are added to this PR, you'll see the packages that this PR includes changesets for and the associated semver types

Click here to learn what changesets are, and how to add one.

Click here if you're a maintainer who wants to add a changeset to this PR

simllll

thanks @Devesh36, i was looking into this issue too.. your changes look good, except for two things I have seen so far:

one quite bad "race" condition in the recvLoop.. because on each processing the message handler is removed and then added again, it happens that messages are lost in between. a simpler and easier approach is just to keep the listener and check the return value of the handler..if it's e.g. true (for done), then resolve and end... I can add it on your changes if you want to?
EOS is just a flag, it should be smthg like a deferred promise.otherwise we have unnecssary delays for checking the flag, and we cannot really push errors through it.

besides this, it's not based on the auto_mode elevnlabs PR that landed in the main already (#820), this would need som ebit
more work to handle sentencetokenizer correctly as well. (e.g. flush on each sentence)

simllll · 2025-11-12T15:59:38Z

plugins/elevenlabs/src/tts.ts

+          const messageHandler = (data: RawData) => {
+            if (!resolved) {
+              resolved = true;
+              ws.removeListener('message', messageHandler);
+              ws.removeListener('close', closeHandler);
+              ws.removeListener('error', errorHandler);
+              resolve(data);
+            }
+          };


here we loose messages, while processing one, we do not have a listener.. after one "on" handler, we hsould keep it.. it even makes the coder easier to read in my opinion

instead of resolveing the promise here, just call a method with the logic..e.g.

// return true if done const myHandler = (data: RawData): boolean => { try { const data = JSON.parse(msg.toString()) as Record<string, unknown>; const contextId = (data.contextId || data.context_id) as string | undefined; if (!contextId || !this.contextData.has(contextId)) { return true; } const context = this.contextData.get(contextId)!; if (data.error) { this.logger.error({ contextId, error: data.error }, 'ElevenLabs error'); this.contextData.delete(contextId); return true; } if (data.audio) { const audioBuffer = Buffer.from(data.audio as string, 'base64'); const audioArray = new Int8Array(audioBuffer); context.audioBuffer.push(audioArray); } if (data.isFinal) { context.eos = true; this.activeContexts.delete(contextId); if (!this.isCurrent && this.activeContexts.size === 0) { this.logger.debug('No active contexts, shutting down'); return true; } } } catch (parseError) { this.logger.warn({ parseError }, 'Failed to parse message'); } }

Suggested change

const messageHandler = (data: RawData) => {

if (!resolved) {

resolved = true;

ws.removeListener('message', messageHandler);

ws.removeListener('close', closeHandler);

ws.removeListener('error', errorHandler);

resolve(data);

}

};

const messageHandler = (data: RawData) => {

const done = myHandler(data);

if (done && !resolved) {

resolved = true;

ws.removeListener('message', messageHandler);

ws.removeListener('close', closeHandler);

ws.removeListener('error', errorHandler);

resolve(data);

}

};

simllll · 2025-11-12T16:00:16Z

plugins/elevenlabs/src/tts.ts

+
+interface StreamContext {
+  contextId: string;
+  eos: boolean;


Suggested change

eos: boolean;

eos: {

promise: Promise<void>;

resolve: () => void;

reject: (err: unknown) => void;

};

simllll · 2025-11-12T16:00:32Z

plugins/elevenlabs/src/tts.ts

+    if (!this.contextData.has(contextId)) {
+      this.contextData.set(contextId, {
+        contextId,
+        eos: false,


Suggested change

eos: false,

eos: newDeferredPromise<void>(),

the newDeferredPromise helper is:

export const newDeferredPromise = <T>(): { promise: Promise<any>; resolve: (value: T | PromiseLike<T>) => void; reject: (reason: any) => void; } => { let fResolve: undefined | ((value: T | PromiseLike<T>) => void); let fReject: undefined | ((reason: any) => void); const P = new Promise<T>((resolve, reject) => { fResolve = resolve; fReject = reject; }); if (!fResolve || !fReject) { throw new Error('Promise init failed'); } return { promise: P, resolve: fResolve, reject: fReject, }; };

simllll · 2025-11-12T16:01:18Z

plugins/elevenlabs/src/tts.ts

+  isContextEOS(contextId: string): boolean {
+    return this.contextData.get(contextId)?.eos ?? false;
+  }


Suggested change

isContextEOS(contextId: string): boolean {

return this.contextData.get(contextId)?.eos ?? false;

}

getContextEOS(contextId: string): Promise<void> | undefined {

return this.contextData.get(contextId)?.eos.promise;

}

simllll · 2025-11-12T16:01:36Z

plugins/elevenlabs/src/tts.ts

+      while (!this.#connection!.isContextEOS(this.#contextId)) {
+        await new Promise((resolve) => setTimeout(resolve, 10));
+      }


Suggested change

while (!this.#connection!.isContextEOS(this.#contextId)) {

await new Promise((resolve) => setTimeout(resolve, 10));

}

await this.#connection!.getContextEOS(this.#contextId);

simllll · 2025-11-12T16:02:15Z

plugins/elevenlabs/src/tts.ts

+          }
+
+          if (data.isFinal) {
+            context.eos = true;


Suggested change

context.eos = true;

context.eos.resolve();

simllll · 2025-11-12T16:02:30Z

plugins/elevenlabs/src/tts.ts

+    } catch (error) {
+      this.logger.error({ error }, 'Recv loop error');
+      for (const context of this.contextData.values()) {
+        context.eos = true;


Suggested change

context.eos = true;

context.eos.reject(error);

simllll

just a small thing, because the " " confused me at the ws.send part... i moved it to the place where it actually is described (at least in the comment ;-)).

and one message is better than two anyways :p

ps.: ts types ppbl missing

simllll · 2025-11-12T16:14:57Z

plugins/elevenlabs/src/tts.ts

-          text: text + ' ', // must always end with a space
-          // ...(flushOnChunk && { flush: true }),
-        });
+        this.#connection!.sendContent(this.#contextId, text, false);


Suggested change

this.#connection!.sendContent(this.#contextId, text, false);

this.#connection.sendContent(this.#contextId, text + " ", false);

simllll · 2025-11-12T16:16:55Z

plugins/elevenlabs/src/tts.ts

+          const textPkt = {
+            text: msg.text + ' ',
+            context_id: msg.contextId,
+          };
+
+          this.ws.send(JSON.stringify(textPkt));
+
+          if (msg.flush) {
+            const flushPkt = {
+              text: '',
+              context_id: msg.contextId,
+            };
+            this.ws.send(JSON.stringify(flushPkt));
+          }


Suggested change

const textPkt = {

text: msg.text + ' ',

context_id: msg.contextId,

};

this.ws.send(JSON.stringify(textPkt));

if (msg.flush) {

const flushPkt = {

text: '',

context_id: msg.contextId,

};

this.ws.send(JSON.stringify(flushPkt));

}

const pkt = {

text: msg.text,

context_id: msg.contextId,

};

if (msg.flush) {

pkt.flush = true;

}

this.ws.send(JSON.stringify(pkt));

Devesh36 · 2025-11-12T16:48:06Z

Yes @simllll i'm still working on the issue ;)
Thanks for the review 🙌

simllll · 2025-11-12T16:57:38Z

Yes @simllll i'm still working on the issue ;) Thanks for the review 🙌

(y) the main issue are the lost messages, this was kinda hard to debug for me, as the isFinal was sometimes there and sometimes not.. ;) happy to help. let me know if you need another review or similar support

simllll · 2025-11-13T09:51:53Z

plugins/elevenlabs/src/tts.ts

+            }
+          };
+
+          ws.on('message', messageHandler);


regarding the "missing events": ws.once() could also solve it I guess

Devesh36 and others added 9 commits November 7, 2025 00:27

refactor: apply minInterruptionWords check consistently; add comprehe…

9039f2f

…nsive interruption detection tests

Delete REFACTORING_SUMMARY.md

b3e511e

Delete interruption threshold logic tests

9fed052

Removed tests for interruption threshold logic.

Remove undefined and null handling tests

1808876

Removed tests for undefined and null handling normalization.

refactor: remove unnecessary test in interruption detection tests

75e0940

refactor: implement persistent WebSocket connection for ElevenLabs TTS

72f6f31

Devesh36 marked this pull request as draft November 12, 2025 13:09

simllll reviewed Nov 12, 2025

View reviewed changes

simllll reviewed Nov 13, 2025

View reviewed changes

-  eos: boolean;
+  eos: {
+    promise: Promise<void>;
+    resolve: () => void;
+    reject: (err: unknown) => void;
+  };

	this.#connection!.sendContent(this.#contextId, text, false);
	this.#connection.sendContent(this.#contextId, text + " ", false);

Fix: Implement Persistent WebSocket Connection for ElevenLabs TTS #828

Are you sure you want to change the base?

Fix: Implement Persistent WebSocket Connection for ElevenLabs TTS #828

Conversation

Devesh36 commented Nov 12, 2025

Description

Changes Made

Pre-Review Checklist

Testing

Additional Notes

Uh oh!

changeset-bot bot commented Nov 12, 2025

⚠️ No Changeset found

Uh oh!

simllll left a comment

Choose a reason for hiding this comment

Uh oh!

simllll Nov 12, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

simllll left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Devesh36 commented Nov 12, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

simllll commented Nov 12, 2025

Uh oh!

simllll Nov 13, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

simllll Nov 12, 2025 •

edited

Loading

simllll left a comment •

edited

Loading

Devesh36 commented Nov 12, 2025 •

edited

Loading

simllll Nov 13, 2025 •

edited

Loading