deepgram-stt: report connection-lifetime remainder so usage matches billing

STTMetrics.audio_duration accumulated only the duration of audio frames
pushed to the Deepgram WebSocket via the PeriodicCollector. Deepgram
bills for the full WebSocket connection lifetime, so cost-tracking
users saw a consistent 4–5 s per-stream undercount against Deepgram's
usage reports (much worse on short calls).

Track the connection open timestamp and the total duration already
reported via _on_audio_duration_report, and on close emit any
remainder between wall-clock connection lifetime and reported frames.
The collector is also flushed before the remainder calc so pending
frame duration is accounted for first.

Fixes #5498

Author: joaquinhuigomez

livekit-plugins/livekit-plugins-deepgram/livekit/plugins/deepgram/stt.py

@@ -410,6 +410,11 @@ def __init__(
 
         self._request_id = ""
         self._reconnect_event = asyncio.Event()
+        # Track how much duration has already been reported so we can emit
+        # the connection-lifetime remainder on close, matching what Deepgram
+        # actually bills (which includes WebSocket open/teardown overhead
+        # beyond the pushed audio frames).
+        self._reported_duration: float = 0.0
 
     def update_options(
         self,
@@ -562,8 +567,10 @@ async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
         ws: aiohttp.ClientWebSocketResponse | None = None
 
         while True:
+            conn_start_time = 0.0
             try:
                 ws = await self._connect_ws()
+                conn_start_time = time.perf_counter()
                 tasks = [
                     asyncio.create_task(send_task(ws)),
                     asyncio.create_task(recv_task(ws)),
@@ -593,6 +600,18 @@ async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
             finally:
                 if ws is not None:
                     await ws.close()
+                    # Deepgram bills WebSocket lifetime, not just audio
+                    # frames pushed.  Emit the remainder between the
+                    # connection's wall-clock lifetime and the frame
+                    # durations we've already reported so usage reflects
+                    # what the provider actually charges for.
+                    if conn_start_time:
+                        self._audio_duration_collector.flush()
+                        lifetime = time.perf_counter() - conn_start_time
+                        remainder = lifetime - self._reported_duration
+                        if remainder > 0:
+                            self._on_audio_duration_report(remainder)
+                        self._reported_duration = 0.0
 
     async def _connect_ws(self) -> aiohttp.ClientWebSocketResponse:
         live_config: dict[str, Any] = {
@@ -646,6 +665,7 @@ async def _connect_ws(self) -> aiohttp.ClientWebSocketResponse:
         return ws
 
     def _on_audio_duration_report(self, duration: float) -> None:
+        self._reported_duration += duration
         usage_event = stt.SpeechEvent(
             type=stt.SpeechEventType.RECOGNITION_USAGE,
             request_id=self._request_id,