update keepalive times depending on watchdog timers

src/pipecat/services/elevenlabs/tts.py

@@ -428,9 +428,10 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
                 self._cumulative_time = word_times[-1][1]
 
     async def _keepalive_task_handler(self):
+        KEEPALIVE_SLEEP = 10 if self.watchdog_timers_enabled else 3
         while True:
             self.reset_watchdog()
-            await asyncio.sleep(4)
+            await asyncio.sleep(KEEPALIVE_SLEEP)
             try:
                 if self._websocket and self._websocket.open:
                     if self._context_id:

src/pipecat/services/gladia/stt.py

@@ -392,8 +392,8 @@ class GladiaSTTService(STTService):
                         await self._send_buffered_audio()
 
                         # Start tasks
-                        self._receive_task = asyncio.create_task(self._receive_task_handler())
-                        self._keepalive_task = asyncio.create_task(self._keepalive_task_handler())
+                        self._receive_task = self.create_task(self._receive_task_handler())
+                        self._keepalive_task = self.create_task(self._keepalive_task_handler())
 
                         # Wait for tasks to complete
                         await asyncio.gather(self._receive_task, self._keepalive_task)
@@ -404,9 +404,9 @@ class GladiaSTTService(STTService):
 
                         # Clean up tasks
                         if self._receive_task:
-                            self._receive_task.cancel()
+                            await self.cancel_task(self._receive_task)
                         if self._keepalive_task:
-                            self._keepalive_task.cancel()
+                            await self.cancel_task(self._keepalive_task)
 
                         # Attempt reconnect using helper
                         if not await self._maybe_reconnect():
@@ -485,9 +485,11 @@ class GladiaSTTService(STTService):
     async def _keepalive_task_handler(self):
         """Send periodic empty audio chunks to keep the connection alive."""
         try:
+            KEEPALIVE_SLEEP = 20 if self.watchdog_timers_enabled else 3
             while self._connection_active:
-                # Send keepalive every 20 seconds (Gladia times out after 30 seconds)
-                await asyncio.sleep(20)
+                self.reset_watchdog()
+                # Send keepalive (Gladia times out after 30 seconds)
+                await asyncio.sleep(KEEPALIVE_SLEEP)
                 if self._websocket and not self._websocket.closed:
                     # Send an empty audio chunk as keepalive
                     empty_audio = b""

src/pipecat/services/neuphonic/tts.py

@@ -30,6 +30,7 @@ from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.tts_service import InterruptibleTTSService, TTSService
 from pipecat.transcriptions.language import Language
 from pipecat.utils.tracing.service_decorators import traced_tts
+from pipecat.utils.watchdog_async_iterator import WatchdogAsyncIterator
 
 try:
     import websockets
@@ -221,7 +222,9 @@ class NeuphonicTTSService(InterruptibleTTSService):
             self._websocket = None
 
     async def _receive_messages(self):
-        async for message in self._websocket:
+        async for message in WatchdogAsyncIterator(
+            self._websocket, reseter=self, watchdog_enabled=self.watchdog_timers_enabled
+        ):
             if isinstance(message, str):
                 msg = json.loads(message)
                 if msg.get("data", {}).get("audio") is not None:
@@ -232,8 +235,10 @@ class NeuphonicTTSService(InterruptibleTTSService):
                     await self.push_frame(frame)
 
     async def _keepalive_task_handler(self):
+        KEEPALIVE_SLEEP = 10 if self.watchdog_timers_enabled else 3
         while True:
-            await asyncio.sleep(10)
+            self.reset_watchdog()
+            await asyncio.sleep(KEEPALIVE_SLEEP)
             await self._send_text("")
 
     async def _send_text(self, text: str):