fix(stt): preserve provider speech end timing

livekit-agents/livekit/agents/stt/stt.py

@@ -121,6 +121,9 @@ class SpeechEvent:
     speech_start_time: float | None = None
     """server-reported wall-clock time of speech onset, when the provider sends
     a separate speech-start signal carrying onset timing."""
+    speech_end_time: float | None = None
+    """server-reported wall-clock time of speech end, when the provider sends
+    a separate speech-end signal carrying end timing."""
 
 
 @dataclass

livekit-agents/livekit/agents/voice/audio_recognition.py

@@ -1190,17 +1190,19 @@ async def _on_stt_event(self, ev: stt.SpeechEvent) -> None:
                 await self._flush_held_transcripts(cooldown=end_cooldown)
                 # no return here to allow the new event to be processed normally
 
-        has_stt_end_time = bool(
+        has_stt_speech_end_time = ev.speech_end_time is not None and ev.speech_end_time > 0
+        has_stt_transcript_end_time = bool(
             len(ev.alternatives) > 0
             and ev.alternatives[0].end_time > 0
             and self._input_started_at is not None
         )
         now = time.time()
-        stt_last_speaking_time = (
-            min(ev.alternatives[0].end_time + self._input_started_at, now)
-            if has_stt_end_time and self._input_started_at is not None
-            else now
-        )
+        if has_stt_speech_end_time and ev.speech_end_time is not None:
+            stt_last_speaking_time = min(ev.speech_end_time, now)
+        elif has_stt_transcript_end_time and self._input_started_at is not None:
+            stt_last_speaking_time = min(ev.alternatives[0].end_time + self._input_started_at, now)
+        else:
+            stt_last_speaking_time = now
         if ev.type == stt.SpeechEventType.FINAL_TRANSCRIPT:
             transcript = ev.alternatives[0].text
             language = ev.alternatives[0].language

livekit-plugins/livekit-plugins-deepgram/livekit/plugins/deepgram/stt_v2.py

@@ -17,6 +17,7 @@
 import asyncio
 import json
 import os
+import time
 import weakref
 from collections.abc import Sequence
 from dataclasses import dataclass
@@ -429,6 +430,7 @@ async def send_task(ws: aiohttp.ClientWebSocketResponse) -> None:
             nonlocal closing_ws
 
             # forward audio to deepgram in chunks of 50ms
+            anchored = False
             samples_50ms = self._opts.sample_rate // 20
             audio_bstream = utils.audio.AudioByteStream(
                 sample_rate=self._opts.sample_rate,
@@ -447,6 +449,11 @@ async def send_task(ws: aiohttp.ClientWebSocketResponse) -> None:
                         has_ended = True
 
                     for frame in frames:
+                        if not anchored:
+                            # Flux reports speech timing relative to the start of the first
+                            # audio frame, which has already elapsed when it reaches this loop.
+                            self.start_time = time.time() - frame.duration
+                            anchored = True
                         self._audio_duration_collector.push(frame.duration)
                         await ws.send_bytes(frame.data.tobytes())
 
@@ -649,7 +656,21 @@ def _process_stream_event(self, data: dict) -> None:
 
                 self._send_transcript_event(stt.SpeechEventType.FINAL_TRANSCRIPT, data)
 
-                end_event = stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH)
+                speech_end_offset: float | None = None
+                for word in data.get("words") or []:
+                    word_end = word.get("end")
+                    if isinstance(word_end, (int, float)):
+                        word_end = float(word_end)
+                        if speech_end_offset is None or word_end > speech_end_offset:
+                            speech_end_offset = word_end
+
+                speech_end_time = (
+                    self.start_time + speech_end_offset if speech_end_offset is not None else None
+                )
+                end_event = stt.SpeechEvent(
+                    type=stt.SpeechEventType.END_OF_SPEECH,
+                    speech_end_time=speech_end_time,
+                )
                 self._event_ch.send_nowait(end_event)
 
         elif data["type"] == "ConfigureSuccess":

tests/test_agent_session.py

@@ -1127,11 +1127,11 @@ async def test_backchannel_boundary_suppresses_start_boundary_backchannel() -> N
         await _close_test_session(session)
 
 
-async def _make_stt_eos_recognition() -> AudioRecognition:
+async def _make_stt_eos_recognition(*, min_delay: float = 0.0) -> AudioRecognition:
     return AudioRecognition(
         create_session(FakeActions()),
         hooks=_TestRecognitionHooks(),
-        endpointing=BaseEndpointing(min_delay=0.0, max_delay=0.0),
+        endpointing=BaseEndpointing(min_delay=min_delay, max_delay=min_delay),
         stt=None,
         vad=None,
         using_default_vad=False,
@@ -1140,6 +1140,46 @@ async def _make_stt_eos_recognition() -> AudioRecognition:
     )
 
 
+@pytest.mark.parametrize("explicit_vad", [False, True])
+async def test_stt_eos_endpointing_uses_actual_speech_end_time(explicit_vad: bool) -> None:
+    recognition = await _make_stt_eos_recognition(min_delay=0.5)
+    signal_received_at = 100.7
+    speech_end_time = 100.4
+    endpointing_timeouts: list[float] = []
+
+    if explicit_vad:
+        recognition._vad = MagicMock()
+        recognition._last_speaking_time = speech_end_time
+
+    async def capture_endpointing_timeout(awaitable: object, timeout: float) -> None:
+        close = getattr(awaitable, "close", None)
+        if close is not None:
+            close()
+        endpointing_timeouts.append(timeout)
+        raise asyncio.TimeoutError
+
+    try:
+        with (
+            patch(
+                "livekit.agents.voice.audio_recognition.time.time",
+                return_value=signal_received_at,
+            ),
+            patch.object(asyncio, "wait_for", side_effect=capture_endpointing_timeout),
+        ):
+            await recognition._on_stt_event(
+                SpeechEvent(
+                    type=SpeechEventType.END_OF_SPEECH,
+                    speech_end_time=speech_end_time,
+                )
+            )
+            assert recognition._end_of_turn_task is not None
+            await recognition._end_of_turn_task
+
+        assert endpointing_timeouts == [pytest.approx(0.2)]
+    finally:
+        await _close_test_session(recognition._session)
+
+
 async def test_stt_eos_resets_active_vad_stream_without_restarting_vad() -> None:
     recognition = await _make_stt_eos_recognition()
     recognition._speaking = True

tests/test_plugin_deepgram_stt.py

@@ -44,6 +44,46 @@ def _make_flux_stream(*, ws=None, **opts_kwargs):
     return stream
 
 
+def test_flux_end_of_turn_emits_server_speech_end_time():
+    from livekit.agents import stt
+    from livekit.plugins.deepgram.stt_v2 import SpeechStreamv2
+
+    events: list[stt.SpeechEvent] = []
+    stream = _make_flux_stream()
+    stream._speaking = True
+    stream._request_id = "request-id"
+    stream.start_time = 100.0
+    stream.start_time_offset = 0.0
+    stream._event_ch = SimpleNamespace(send_nowait=events.append)
+    stream._send_transcript_event = SpeechStreamv2._send_transcript_event.__get__(stream)
+
+    SpeechStreamv2._process_stream_event(
+        stream,
+        {
+            "type": "TurnInfo",
+            "event": "EndOfTurn",
+            "request_id": "request-id",
+            "transcript": "hello",
+            "audio_window_start": 1.0,
+            "audio_window_end": 3.0,
+            "words": [
+                {
+                    "word": "hello",
+                    "start": 1.0,
+                    "end": 2.5,
+                    "confidence": 0.9,
+                }
+            ],
+        },
+    )
+
+    assert [event.type for event in events] == [
+        stt.SpeechEventType.FINAL_TRANSCRIPT,
+        stt.SpeechEventType.END_OF_SPEECH,
+    ]
+    assert events[-1].speech_end_time == pytest.approx(102.5)
+
+
 async def test_update_options_uses_stored_language_for_model_validation():
     from livekit.plugins.deepgram import STT
 
@@ -208,6 +248,31 @@ async def _fake_connect() -> Any:
     return stream
 
 
+def _live_flux_stream(ws: _LiveWS):
+    """A real Flux SpeechStream running its send loop against a fake socket."""
+    import dataclasses
+    from typing import Any, cast
+
+    from livekit.agents import DEFAULT_API_CONNECT_OPTIONS
+    from livekit.plugins.deepgram.stt_v2 import SpeechStreamv2, STTv2
+
+    instance = STTv2(api_key="test-key", sample_rate=16000)
+    stream = SpeechStreamv2(
+        stt=instance,
+        opts=dataclasses.replace(instance._opts, sample_rate=16000),
+        conn_options=DEFAULT_API_CONNECT_OPTIONS,
+        api_key="test-key",
+        http_session=cast(Any, SimpleNamespace(closed=False)),
+        base_url="wss://api.deepgram.com/v2/listen",
+    )
+
+    async def _fake_connect() -> Any:
+        return ws
+
+    stream._connect_ws = _fake_connect
+    return stream
+
+
 def _frame(ms: int, sample_rate: int = 16000):
     from livekit import rtc
 
@@ -229,6 +294,54 @@ async def _wait_until(predicate, *, timeout: float = 5.0) -> None:
         await asyncio.sleep(0.01)
 
 
+async def test_flux_anchors_server_timestamps_to_first_audio_frame():
+    import time
+
+    ws = _LiveWS()
+    stream = _live_flux_stream(ws)
+    try:
+        await asyncio.sleep(0.01)
+        before_send = time.time()
+        stream.push_frame(_frame(50))
+        await _wait_until(lambda: ws.sent() == ["audio"])
+        after_send = time.time()
+
+        assert before_send - 0.05 <= stream.start_time <= after_send - 0.05
+    finally:
+        await stream.aclose()
+
+
+async def test_flux_reanchors_server_timestamps_after_reconnect():
+    import time
+
+    first_ws = _LiveWS()
+    second_ws = _LiveWS()
+    connections = iter([first_ws, second_ws])
+    stream = _live_flux_stream(first_ws)
+
+    async def _next_connection():
+        return next(connections)
+
+    stream._connect_ws = _next_connection
+    try:
+        stream.push_frame(_frame(50))
+        await _wait_until(lambda: first_ws.sent() == ["audio"])
+        first_anchor = stream.start_time
+
+        await asyncio.sleep(0.01)
+        stream._reconnect_event.set()
+        await _wait_until(lambda: stream._ws is second_ws)
+
+        before_second_send = time.time()
+        stream.push_frame(_frame(50))
+        await _wait_until(lambda: second_ws.sent() == ["audio"])
+
+        assert stream.start_time > first_anchor
+        assert before_second_send - 0.05 <= stream.start_time <= time.time() - 0.05
+    finally:
+        await stream.aclose()
+
+
 async def test_flush_finalizes_the_turn_when_no_audio_is_left_to_send():
     # 50ms is exactly one repack chunk, so AudioByteStream.flush() returns no frames.
     # Finalize must still go out, otherwise the turn waits on Deepgram's own endpointing