fix(deepgram): keep the pooled Flux TTS websocket alive between turnsj

livekit-plugins/livekit-plugins-deepgram/livekit/plugins/deepgram/tts_v2.py

@@ -35,6 +35,9 @@
 # https://developers.deepgram.com/docs/flux-tts/overview
 BASE_URL_V2 = "https://api.deepgram.com/v2/speak"
 NUM_CHANNELS = 1
+# Deepgram closes a Flux TTS session after 60s without a client message (NET-0004);
+# a WebSocket ping resets that timer, so ping at half the window.
+DEFAULT_KEEPALIVE_INTERVAL = 30.0
 
 # Encodings the LiveKit pipeline can decode, mapped to their mime type. Deepgram Flux
 # also offers mulaw/alaw, but those aren't playable through the pipeline yet, so they're
@@ -68,6 +71,7 @@ class _TTSOptionsV2:
     api_key: str
     mip_opt_out: bool = False
     bit_rate: int | None = None
+    keepalive_interval: float | None = DEFAULT_KEEPALIVE_INTERVAL
 
 
 class TTSv2(tts.TTS):
@@ -83,6 +87,7 @@ def __init__(
         word_tokenizer: NotGivenOr[tokenize.WordTokenizer] = NOT_GIVEN,
         http_session: aiohttp.ClientSession | None = None,
         mip_opt_out: bool = False,
+        keepalive_interval: float | None = DEFAULT_KEEPALIVE_INTERVAL,
     ) -> None:
         """
         Create a new instance of Deepgram Flux TTS (the /v2/speak endpoint).
@@ -105,6 +110,13 @@ def __init__(
             http_session (aiohttp.ClientSession): Optional aiohttp session to use for requests.
             mip_opt_out (bool): Opt out of the Deepgram Model Improvement Program. Defaults to
                 False (requests may be used to improve models). See https://dpgr.am/deepgram-mip
+            keepalive_interval (float | None): Seconds between WebSocket ping frames sent on an
+                idle streaming connection. Defaults to 30. Deepgram closes a Flux TTS session
+                that receives no client message for 60 seconds (error NET-0004), and the
+                streaming connection is pooled and reused across turns, so without pings a
+                pause longer than that (the user thinking, a long tool call) leaves the next
+                turn on a closed socket. A ping or pong resets Deepgram's timer. None disables
+                the pings.
 
         """  # noqa: E501
         super().__init__(
@@ -129,6 +141,7 @@ def __init__(
             base_url=base_url,
             api_key=api_key,
             mip_opt_out=mip_opt_out,
+            keepalive_interval=keepalive_interval,
         )
         self._session = http_session
         self._streams = weakref.WeakSet[SynthesizeStreamv2]()
@@ -162,6 +175,8 @@ async def _connect_ws(self, timeout: float) -> aiohttp.ClientWebSocketResponse:
             session.ws_connect(
                 _to_deepgram_url(config, self._opts.base_url, websocket=True),
                 headers={"Authorization": f"Token {self._opts.api_key}"},
+                # keeps the pooled connection inside Deepgram's 60s inactivity window
+                heartbeat=self._opts.keepalive_interval,
             ),
             timeout,
         )

tests/test_plugin_deepgram_tts.py

@@ -248,3 +248,49 @@ async def test_stream_run_rejects_non_linear16_encoding():
 
     with pytest.raises(ValueError, match="linear16"):
         await stream._run(_FakeEmitter())  # type: ignore[arg-type]
+
+
+# --- pooled connection keepalive -----------------------------------------------------
+
+
+class _FakeSession:
+    """Records the ws_connect call and hands back a socket with the headers the
+    connect path logs."""
+
+    def __init__(self) -> None:
+        self.calls: list[dict] = []
+
+    async def ws_connect(self, url: str, **kwargs):  # noqa: ANN201
+        self.calls.append({"url": url, **kwargs})
+        return SimpleNamespace(_response=SimpleNamespace(headers={}))
+
+
+async def test_connect_ws_pings_the_pooled_connection_by_default():
+    # Deepgram closes a Flux session that gets no client message for 60s (NET-0004).
+    # The streaming connection is pooled across turns, so a ping keeps a pause longer
+    # than that from leaving the next turn on a closed socket.
+    from livekit.plugins.deepgram import TTSv2
+    from livekit.plugins.deepgram.tts_v2 import DEFAULT_KEEPALIVE_INTERVAL
+
+    tts = TTSv2(api_key="test-key")
+    session = _FakeSession()
+    tts._ensure_session = lambda: session  # type: ignore[method-assign]
+
+    await tts._connect_ws(timeout=5.0)
+
+    [call] = session.calls
+    assert call["heartbeat"] == DEFAULT_KEEPALIVE_INTERVAL
+    assert 0 < DEFAULT_KEEPALIVE_INTERVAL < 60
+
+
+async def test_keepalive_interval_none_disables_the_ping():
+    from livekit.plugins.deepgram import TTSv2
+
+    tts = TTSv2(api_key="test-key", keepalive_interval=None)
+    session = _FakeSession()
+    tts._ensure_session = lambda: session  # type: ignore[method-assign]
+
+    await tts._connect_ws(timeout=5.0)
+
+    [call] = session.calls
+    assert call["heartbeat"] is None