From 497d48b6c8ae1b98077615908143916739c729f5 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= <aleix@daily.co>
Date: Thu, 17 Apr 2025 11:29:49 -0700
Subject: [PATCH 1/2] services: fix TTS websocket services disconnections

Fixes #1467
---
 CHANGELOG.md                              |  3 +++
 src/pipecat/services/cartesia/tts.py      | 10 +++++-----
 src/pipecat/services/elevenlabs/tts.py    | 10 +++++-----
 src/pipecat/services/fish/tts.py          | 11 ++++++-----
 src/pipecat/services/lmnt/tts.py          | 12 ++++++------
 src/pipecat/services/neuphonic/tts.py     | 11 +++++++----
 src/pipecat/services/playht/tts.py        |  8 ++++----
 src/pipecat/services/rime/tts.py          |  9 +++++----
 src/pipecat/services/websocket_service.py |  2 +-
 9 files changed, 42 insertions(+), 34 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 80b1369d5..491cbf24a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -50,6 +50,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed an issue that would cause TTS websocket-based services to not cleanup
+  resources properly when disconnecting.
+
 - Fixed a `TavusVideoService` issue that was causing audio choppiness.
 
 - Fixed an issue in `SmallWebRTCTransport` where an error was thrown if the
diff --git a/src/pipecat/services/cartesia/tts.py b/src/pipecat/services/cartesia/tts.py
index eb891693d..3fc70c5cd 100644
--- a/src/pipecat/services/cartesia/tts.py
+++ b/src/pipecat/services/cartesia/tts.py
@@ -197,7 +197,7 @@ class CartesiaTTSService(AudioContextWordTTSService):
 
     async def _connect_websocket(self):
         try:
-            if self._websocket:
+            if self._websocket and self._websocket.open:
                 return
             logger.debug("Connecting to Cartesia")
             self._websocket = await websockets.connect(
@@ -215,11 +215,11 @@ class CartesiaTTSService(AudioContextWordTTSService):
             if self._websocket:
                 logger.debug("Disconnecting from Cartesia")
                 await self._websocket.close()
-                self._websocket = None
-
-            self._context_id = None
         except Exception as e:
             logger.error(f"{self} error closing websocket: {e}")
+        finally:
+            self._context_id = None
+            self._websocket = None
 
     def _get_websocket(self):
         if self._websocket:
@@ -279,7 +279,7 @@ class CartesiaTTSService(AudioContextWordTTSService):
         logger.debug(f"{self}: Generating TTS [{text}]")
 
         try:
-            if not self._websocket:
+            if not self._websocket or self._websocket.closed:
                 await self._connect()
 
             if not self._context_id:
diff --git a/src/pipecat/services/elevenlabs/tts.py b/src/pipecat/services/elevenlabs/tts.py
index cc9a72889..54ba99fb7 100644
--- a/src/pipecat/services/elevenlabs/tts.py
+++ b/src/pipecat/services/elevenlabs/tts.py
@@ -327,7 +327,7 @@ class ElevenLabsTTSService(InterruptibleWordTTSService):
 
     async def _connect_websocket(self):
         try:
-            if self._websocket:
+            if self._websocket and self._websocket.open:
                 return
 
             logger.debug("Connecting to ElevenLabs")
@@ -374,11 +374,11 @@ class ElevenLabsTTSService(InterruptibleWordTTSService):
                 logger.debug("Disconnecting from ElevenLabs")
                 await self._websocket.send(json.dumps({"text": ""}))
                 await self._websocket.close()
-                self._websocket = None
-
-            self._started = False
         except Exception as e:
             logger.error(f"{self} error closing websocket: {e}")
+        finally:
+            self._started = False
+            self._websocket = None
 
     def _get_websocket(self):
         if self._websocket:
@@ -418,7 +418,7 @@ class ElevenLabsTTSService(InterruptibleWordTTSService):
         logger.debug(f"{self}: Generating TTS [{text}]")
 
         try:
-            if not self._websocket:
+            if not self._websocket or self._websocket.closed:
                 await self._connect()
 
             try:
diff --git a/src/pipecat/services/fish/tts.py b/src/pipecat/services/fish/tts.py
index c3ebba6e4..177444f35 100644
--- a/src/pipecat/services/fish/tts.py
+++ b/src/pipecat/services/fish/tts.py
@@ -116,7 +116,7 @@ class FishAudioTTSService(InterruptibleTTSService):
 
     async def _connect_websocket(self):
         try:
-            if self._websocket:
+            if self._websocket and self._websocket.open:
                 return
 
             logger.debug("Connecting to Fish Audio")
@@ -141,16 +141,17 @@ class FishAudioTTSService(InterruptibleTTSService):
                 stop_message = {"event": "stop"}
                 await self._websocket.send(ormsgpack.packb(stop_message))
                 await self._websocket.close()
-                self._websocket = None
-            self._request_id = None
-            self._started = False
         except Exception as e:
             logger.error(f"Error closing websocket: {e}")
+        finally:
+            self._request_id = None
+            self._started = False
+            self._websocket = None
 
     async def flush_audio(self):
         """Flush any buffered audio by sending a flush event to Fish Audio."""
         logger.trace(f"{self}: Flushing audio buffers")
-        if not self._websocket:
+        if not self._websocket or self._websocket.closed:
             return
         flush_message = {"event": "flush"}
         await self._get_websocket().send(ormsgpack.packb(flush_message))
diff --git a/src/pipecat/services/lmnt/tts.py b/src/pipecat/services/lmnt/tts.py
index 993834bb1..eb07c9e8a 100644
--- a/src/pipecat/services/lmnt/tts.py
+++ b/src/pipecat/services/lmnt/tts.py
@@ -122,7 +122,7 @@ class LmntTTSService(InterruptibleTTSService):
     async def _connect_websocket(self):
         """Connect to LMNT websocket."""
         try:
-            if self._websocket:
+            if self._websocket and self._websocket.open:
                 return
 
             logger.debug("Connecting to LMNT")
@@ -158,11 +158,11 @@ class LmntTTSService(InterruptibleTTSService):
                 # errors on the websocket, so we just skip it for now.
                 # await self._websocket.send(json.dumps({"eof": True}))
                 await self._websocket.close()
-                self._websocket = None
-
-            self._started = False
         except Exception as e:
             logger.error(f"{self} error closing websocket: {e}")
+        finally:
+            self._started = False
+            self._websocket = None
 
     def _get_websocket(self):
         if self._websocket:
@@ -170,7 +170,7 @@ class LmntTTSService(InterruptibleTTSService):
         raise Exception("Websocket not connected")
 
     async def flush_audio(self):
-        if not self._websocket:
+        if not self._websocket or self._websocket.closed:
             return
         await self._get_websocket().send(json.dumps({"flush": True}))
 
@@ -203,7 +203,7 @@ class LmntTTSService(InterruptibleTTSService):
         logger.debug(f"{self}: Generating TTS [{text}]")
 
         try:
-            if not self._websocket:
+            if not self._websocket or self._websocket.closed:
                 await self._connect()
 
             try:
diff --git a/src/pipecat/services/neuphonic/tts.py b/src/pipecat/services/neuphonic/tts.py
index 72fee0021..0d6a2de44 100644
--- a/src/pipecat/services/neuphonic/tts.py
+++ b/src/pipecat/services/neuphonic/tts.py
@@ -175,6 +175,9 @@ class NeuphonicTTSService(InterruptibleTTSService):
 
     async def _connect_websocket(self):
         try:
+            if self._websocket and self._websocket.open:
+                return
+
             logger.debug("Connecting to Neuphonic")
 
             tts_config = {
@@ -203,11 +206,11 @@ class NeuphonicTTSService(InterruptibleTTSService):
             if self._websocket:
                 logger.debug("Disconnecting from Neuphonic")
                 await self._websocket.close()
-                self._websocket = None
-
-            self._started = False
         except Exception as e:
             logger.error(f"{self} error closing websocket: {e}")
+        finally:
+            self._started = False
+            self._websocket = None
 
     async def _receive_messages(self):
         async for message in self._websocket:
@@ -235,7 +238,7 @@ class NeuphonicTTSService(InterruptibleTTSService):
         logger.debug(f"Generating TTS: [{text}]")
 
         try:
-            if not self._websocket:
+            if not self._websocket or self._websocket.closed:
                 await self._connect()
 
             try:
diff --git a/src/pipecat/services/playht/tts.py b/src/pipecat/services/playht/tts.py
index c2c14c88b..96bf1c75b 100644
--- a/src/pipecat/services/playht/tts.py
+++ b/src/pipecat/services/playht/tts.py
@@ -169,7 +169,7 @@ class PlayHTTTSService(InterruptibleTTSService):
 
     async def _connect_websocket(self):
         try:
-            if self._websocket:
+            if self._websocket and self._websocket.open:
                 return
 
             logger.debug("Connecting to PlayHT")
@@ -197,11 +197,11 @@ class PlayHTTTSService(InterruptibleTTSService):
             if self._websocket:
                 logger.debug("Disconnecting from PlayHT")
                 await self._websocket.close()
-                self._websocket = None
-
-            self._request_id = None
         except Exception as e:
             logger.error(f"{self} error closing websocket: {e}")
+        finally:
+            self._request_id = None
+            self._websocket = None
 
     async def _get_websocket_url(self):
         async with aiohttp.ClientSession() as session:
diff --git a/src/pipecat/services/rime/tts.py b/src/pipecat/services/rime/tts.py
index 0e1c7d239..abf08bbaa 100644
--- a/src/pipecat/services/rime/tts.py
+++ b/src/pipecat/services/rime/tts.py
@@ -182,7 +182,7 @@ class RimeTTSService(AudioContextWordTTSService):
     async def _connect_websocket(self):
         """Connect to Rime websocket API with configured settings."""
         try:
-            if self._websocket:
+            if self._websocket and self._websocket.open:
                 return
 
             params = "&".join(f"{k}={v}" for k, v in self._settings.items())
@@ -201,10 +201,11 @@ class RimeTTSService(AudioContextWordTTSService):
             if self._websocket:
                 await self._websocket.send(json.dumps(self._build_eos_msg()))
                 await self._websocket.close()
-                self._websocket = None
-            self._context_id = None
         except Exception as e:
             logger.error(f"{self} error closing websocket: {e}")
+        finally:
+            self._context_id = None
+            self._websocket = None
 
     def _get_websocket(self):
         """Get active websocket connection or raise exception."""
@@ -316,7 +317,7 @@ class RimeTTSService(AudioContextWordTTSService):
         """
         logger.debug(f"{self}: Generating TTS [{text}]")
         try:
-            if not self._websocket:
+            if not self._websocket or self._websocket.closed:
                 await self._connect()
 
             try:
diff --git a/src/pipecat/services/websocket_service.py b/src/pipecat/services/websocket_service.py
index 2e82ddaba..d757946f9 100644
--- a/src/pipecat/services/websocket_service.py
+++ b/src/pipecat/services/websocket_service.py
@@ -31,7 +31,7 @@ class WebsocketService(ABC):
             bool: True if connection is verified working, False otherwise
         """
         try:
-            if not self._websocket:
+            if not self._websocket or self._websocket.closed:
                 return False
             await self._websocket.ping()
             return True

From c0b3fe3985aba168a7f6b99f3741138ca53e70c7 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= <aleix@daily.co>
Date: Thu, 17 Apr 2025 11:54:07 -0700
Subject: [PATCH 2/2] services: only read from TTS websocket if websocket
 connection established

---
 src/pipecat/services/cartesia/tts.py   |  3 ++-
 src/pipecat/services/elevenlabs/tts.py |  4 ++--
 src/pipecat/services/fish/tts.py       |  3 ++-
 src/pipecat/services/gladia/stt.py     |  2 +-
 src/pipecat/services/lmnt/tts.py       |  2 +-
 src/pipecat/services/neuphonic/tts.py  | 11 ++++++++---
 src/pipecat/services/playht/tts.py     |  2 +-
 src/pipecat/services/rime/tts.py       |  2 +-
 8 files changed, 18 insertions(+), 11 deletions(-)

diff --git a/src/pipecat/services/cartesia/tts.py b/src/pipecat/services/cartesia/tts.py
index 3fc70c5cd..2acabd541 100644
--- a/src/pipecat/services/cartesia/tts.py
+++ b/src/pipecat/services/cartesia/tts.py
@@ -185,7 +185,8 @@ class CartesiaTTSService(AudioContextWordTTSService):
 
     async def _connect(self):
         await self._connect_websocket()
-        if not self._receive_task:
+
+        if self._websocket and not self._receive_task:
             self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
 
     async def _disconnect(self):
diff --git a/src/pipecat/services/elevenlabs/tts.py b/src/pipecat/services/elevenlabs/tts.py
index 54ba99fb7..4efaa8807 100644
--- a/src/pipecat/services/elevenlabs/tts.py
+++ b/src/pipecat/services/elevenlabs/tts.py
@@ -308,10 +308,10 @@ class ElevenLabsTTSService(InterruptibleWordTTSService):
     async def _connect(self):
         await self._connect_websocket()
 
-        if not self._receive_task:
+        if self._websocket and not self._receive_task:
             self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
 
-        if not self._keepalive_task:
+        if self._websocket and not self._keepalive_task:
             self._keepalive_task = self.create_task(self._keepalive_task_handler())
 
     async def _disconnect(self):
diff --git a/src/pipecat/services/fish/tts.py b/src/pipecat/services/fish/tts.py
index 177444f35..abf609177 100644
--- a/src/pipecat/services/fish/tts.py
+++ b/src/pipecat/services/fish/tts.py
@@ -104,7 +104,8 @@ class FishAudioTTSService(InterruptibleTTSService):
 
     async def _connect(self):
         await self._connect_websocket()
-        if not self._receive_task:
+
+        if self._websocket and not self._receive_task:
             self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
 
     async def _disconnect(self):
diff --git a/src/pipecat/services/gladia/stt.py b/src/pipecat/services/gladia/stt.py
index 03dd35ff2..55581b3e6 100644
--- a/src/pipecat/services/gladia/stt.py
+++ b/src/pipecat/services/gladia/stt.py
@@ -285,7 +285,7 @@ class GladiaSTTService(STTService):
         settings = self._prepare_settings()
         response = await self._setup_gladia(settings)
         self._websocket = await websockets.connect(response["url"])
-        if not self._receive_task:
+        if self._websocket and not self._receive_task:
             self._receive_task = self.create_task(self._receive_task_handler())
 
     async def stop(self, frame: EndFrame):
diff --git a/src/pipecat/services/lmnt/tts.py b/src/pipecat/services/lmnt/tts.py
index eb07c9e8a..b9e0bc79f 100644
--- a/src/pipecat/services/lmnt/tts.py
+++ b/src/pipecat/services/lmnt/tts.py
@@ -109,7 +109,7 @@ class LmntTTSService(InterruptibleTTSService):
     async def _connect(self):
         await self._connect_websocket()
 
-        if not self._receive_task:
+        if self._websocket and not self._receive_task:
             self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
 
     async def _disconnect(self):
diff --git a/src/pipecat/services/neuphonic/tts.py b/src/pipecat/services/neuphonic/tts.py
index 0d6a2de44..1819482b8 100644
--- a/src/pipecat/services/neuphonic/tts.py
+++ b/src/pipecat/services/neuphonic/tts.py
@@ -106,6 +106,9 @@ class NeuphonicTTSService(InterruptibleTTSService):
         self._started = False
         self._cumulative_time = 0
 
+        self._receive_task = None
+        self._keepalive_task = None
+
     def can_generate_metrics(self) -> bool:
         return True
 
@@ -159,8 +162,11 @@ class NeuphonicTTSService(InterruptibleTTSService):
     async def _connect(self):
         await self._connect_websocket()
 
-        self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
-        self._keepalive_task = self.create_task(self._keepalive_task_handler())
+        if self._websocket and not self._receive_task:
+            self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
+
+        if self._websocket and not self._keepalive_task:
+            self._keepalive_task = self.create_task(self._keepalive_task_handler())
 
     async def _disconnect(self):
         if self._receive_task:
@@ -193,7 +199,6 @@ class NeuphonicTTSService(InterruptibleTTSService):
             url = f"{self._url}/speak/{self._settings['lang_code']}?{'&'.join(query_params)}"
 
             self._websocket = await websockets.connect(url)
-
         except Exception as e:
             logger.error(f"{self} initialization error: {e}")
             self._websocket = None
diff --git a/src/pipecat/services/playht/tts.py b/src/pipecat/services/playht/tts.py
index 96bf1c75b..eb1b8d3f1 100644
--- a/src/pipecat/services/playht/tts.py
+++ b/src/pipecat/services/playht/tts.py
@@ -157,7 +157,7 @@ class PlayHTTTSService(InterruptibleTTSService):
     async def _connect(self):
         await self._connect_websocket()
 
-        if not self._receive_task:
+        if self._websocket and not self._receive_task:
             self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
 
     async def _disconnect(self):
diff --git a/src/pipecat/services/rime/tts.py b/src/pipecat/services/rime/tts.py
index abf08bbaa..d3e8e5adb 100644
--- a/src/pipecat/services/rime/tts.py
+++ b/src/pipecat/services/rime/tts.py
@@ -168,7 +168,7 @@ class RimeTTSService(AudioContextWordTTSService):
         """Establish websocket connection and start receive task."""
         await self._connect_websocket()
 
-        if not self._receive_task:
+        if self._websocket and not self._receive_task:
             self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
 
     async def _disconnect(self):