Skip to content

fix: report speech end timestamps across stt plugins #7796

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
chenghao-mou wants to merge 29 commits into from
+2,754 −227
Open

fix: report speech end timestamps across stt plugins #7796

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
29 commits
Select commit Hold shift + click to select a range
05f3300
fix(stt): anchor transcript timestamps on the session input start
chenghao-mou Oct 13, 2026
fe33192
fix(stt): anchor the offset on consumed audio and keep custom nodes f…
chenghao-mou Oct 13, 2026
26820a3
fix(stt): stamp resampled frames with the capture time of their audio
chenghao-mou Oct 13, 2026
2cdf6bc
fix(stt): keep capture gaps when stamping resampled frames
chenghao-mou Oct 13, 2026
953bdfa
fix(stt): report speech end timestamps on endpoint events
chenghao-mou Oct 13, 2026
b0d5b5e
fix(stt): anchor the offset on push and drop the resampler span tracking
chenghao-mou Oct 13, 2026
90a815b
fix(stt): preserve endpoint boundaries and valid timing
chenghao-mou Oct 13, 2026
95b9e55
fix(stt): retain verified duration-only endpoint timing
chenghao-mou Oct 13, 2026
15ba66f
docs(stt): state the timestamp contract on stt_node and rename the of…
chenghao-mou Oct 13, 2026
b19759c
refactor(stt): keep one offset helper and write the setter inline
chenghao-mou Oct 13, 2026
d316acd
refactor(stt): rename the offset helper to _capture_start_offset_from
chenghao-mou Oct 13, 2026
668586e
refactor(stt): seed the fallback leg with a walrus
chenghao-mou Oct 13, 2026
1383597
chore(stt): assign the queued-frame TODO
chenghao-mou Oct 13, 2026
0d948d1
fix: preserve EOS timing across streaming STT plugins
chenghao-mou Oct 13, 2026
7137157
fix(openai): re-anchor STT endpoints after internal reconnects
chenghao-mou Oct 14, 2026
44af029
refactor(elevenlabs): share speech endpoint emission
chenghao-mou Oct 14, 2026
9386a25
refactor(stt): keep the stream timeline in the input channel
chenghao-mou Oct 14, 2026
2a48ebf
test(voice): restore the input_started_at stamping check
chenghao-mou Oct 14, 2026
b0ea418
Merge updated STT timestamp anchor into EOS timing fixes
chenghao-mou Oct 14, 2026
5ba998d
fix(stt): anchor queued audio on its oldest capture across input pauses
chenghao-mou Oct 14, 2026
8161e25
fix(stt): seed a fallback leg that starts before the timeline
chenghao-mou Oct 14, 2026
4e59347
test(stt): describe the first-audio anchor for custom nodes
chenghao-mou Oct 14, 2026
fda46fa
Merge updated STT timestamp anchor into EOS timing fixes
chenghao-mou Oct 14, 2026
10dfa08
Merge rebased STT timestamp anchor into EOS fixes
chenghao-mou Oct 14, 2026
b605fe1
test(stt): bound skewed timestamp clamp by arrival time
chenghao-mou Oct 14, 2026
9c585c3
fix(stt): re-anchor provider clocks on internal reconnects
chenghao-mou Oct 20, 2026
42de4f1
chore: isolate elevenlabs speech endpoint timing
chenghao-mou Oct 20, 2026
5f5b302
fix: report speech end times across other stt plugins
chenghao-mou Oct 20, 2026
7820a64
fix(xai): preserve speech timing on wordless completion
chenghao-mou Oct 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 31 additions & 1 deletion livekit-agents/livekit/agents/inference/stt.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -321,6 +321,15 @@ def _supports_agent_context_carryover(model: NotGivenOr[STTModels | str]) -> boo
}
)

_UTTERANCE_DURATION_MODELS = frozenset(
{
"soniox/stt-rt",
"soniox/stt-rt-v4",
"soniox/stt-rt-v5",
"speechmatics/linden-1",
}
)


def _aligned_transcript_for_model(
model: NotGivenOr[STTModels | str],
Expand Down Expand Up @@ -1444,7 +1453,28 @@ def _process_transcript(self, data: dict, is_final: bool) -> None:

if self._speaking:
self._speaking = False
end_event = stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH)
words = data.get("words") or []
end_alternatives = []
if words and words[-1].get("end", 0) > 0:
end_alternatives = [
replace(speech_data, end_time=self.start_time_offset + words[-1]["end"])
]
elif (
data.get("duration", 0) > 0
and self._opts.model in _UTTERANCE_DURATION_MODELS
and (
not is_given(self._opts.fallback)
or all(
item["model"] in _UTTERANCE_DURATION_MODELS
for item in self._opts.fallback
)
)
):
end_alternatives = [speech_data]
end_event = stt.SpeechEvent(
type=stt.SpeechEventType.END_OF_SPEECH,
alternatives=end_alternatives,
)
self._event_ch.send_nowait(end_event)
self._on_end_of_speech()
else:
Expand Down
7 changes: 7 additions & 0 deletions livekit-agents/livekit/agents/stt/multi_speaker_adapter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -281,6 +281,13 @@ def push_audio(self, frame: rtc.AudioFrame) -> None:
self._rms_buffer = self._rms_buffer[-self._max_rms_size :]

def on_stt_event(self, ev: SpeechEvent) -> SpeechEvent | None:
if ev.type not in (
SpeechEventType.INTERIM_TRANSCRIPT,
SpeechEventType.PREFLIGHT_TRANSCRIPT,
SpeechEventType.FINAL_TRANSCRIPT,
):
return ev

if not ev.alternatives:
return ev

Expand Down
9 changes: 8 additions & 1 deletion livekit-plugins/livekit-plugins-assemblyai/livekit/plugins/assemblyai/stt.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1113,7 +1113,14 @@ def _process_stream_event(self, data: dict) -> None:
self._session_id,
)

self._event_ch.send_nowait(stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH))
self._event_ch.send_nowait(
stt.SpeechEvent(
type=stt.SpeechEventType.END_OF_SPEECH,
alternatives=final_event.alternatives
if words and words[-1].get("end", 0) > 0
else [],
)
)

if self._speech_duration > 0.0:
usage_event = stt.SpeechEvent(
Expand Down
13 changes: 10 additions & 3 deletions livekit-plugins/livekit-plugins-aws/livekit/plugins/aws/stt.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -447,30 +447,37 @@ def _process_transcript_event(self, transcript_event: TranscriptEvent) -> None:

stream = transcript_event.transcript.results
for resp in stream:
speech_data: stt.SpeechData | None = None
if resp.start_time is not None and resp.start_time == 0.0:
self._event_ch.send_nowait(
stt.SpeechEvent(type=stt.SpeechEventType.START_OF_SPEECH)
)

if resp.end_time is not None and resp.end_time > 0.0:
speech_data = self._streaming_recognize_response_to_speech_data(resp)
if resp.is_partial:
self._event_ch.send_nowait(
stt.SpeechEvent(
type=stt.SpeechEventType.INTERIM_TRANSCRIPT,
alternatives=[self._streaming_recognize_response_to_speech_data(resp)],
alternatives=[speech_data],
)
)

else:
self._event_ch.send_nowait(
stt.SpeechEvent(
type=stt.SpeechEventType.FINAL_TRANSCRIPT,
alternatives=[self._streaming_recognize_response_to_speech_data(resp)],
alternatives=[speech_data],
)
)

if not resp.is_partial:
self._event_ch.send_nowait(stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH))
self._event_ch.send_nowait(
stt.SpeechEvent(
type=stt.SpeechEventType.END_OF_SPEECH,
alternatives=[speech_data] if speech_data is not None else [],
)
)

def _streaming_recognize_response_to_speech_data(self, resp: Result) -> stt.SpeechData:
confidence = 0.0
Expand Down
17 changes: 15 additions & 2 deletions livekit-plugins/livekit-plugins-azure/livekit/plugins/azure/stt.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -282,6 +282,8 @@ async def _run(self) -> None:
self._session_started_event.wait(), self._conn_options.timeout
)

self._input_ch.restart_provider_clock()

async def process_input() -> None:
async for input in self._input_ch:
if isinstance(input, rtc.AudioFrame):
Expand Down Expand Up @@ -424,7 +426,7 @@ def _on_speech_start(self, evt: speechsdk.SpeechRecognitionEventArgs) -> None:
stt.SpeechEvent(type=stt.SpeechEventType.START_OF_SPEECH),
)

def _on_speech_end(self, evt: speechsdk.SpeechRecognitionEventArgs) -> None:
def _on_speech_end(self, evt: speechsdk.RecognitionEventArgs) -> None:
if not self._speaking:
return

Expand All @@ -433,7 +435,18 @@ def _on_speech_end(self, evt: speechsdk.SpeechRecognitionEventArgs) -> None:
with contextlib.suppress(RuntimeError):
self._loop.call_soon_threadsafe(
self._event_ch.send_nowait,
stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH),
stt.SpeechEvent(
type=stt.SpeechEventType.END_OF_SPEECH,
alternatives=[
stt.SpeechData(
language=LanguageCode(""),
text="",
end_time=evt.offset / 10**7 + self.start_time_offset,
)
]
if evt.offset > 0
else [],
),
)

def _on_session_started(self, evt: speechsdk.SpeechRecognitionEventArgs) -> None:
Expand Down
7 changes: 6 additions & 1 deletion livekit-plugins/livekit-plugins-baseten/livekit/plugins/baseten/qwen3_stt.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -378,7 +378,12 @@ async def _recv() -> None:
if is_final:
if speaking:
self._event_ch.send_nowait(
stt_module.SpeechEvent(type=stt_module.SpeechEventType.END_OF_SPEECH)
stt_module.SpeechEvent(
type=stt_module.SpeechEventType.END_OF_SPEECH,
alternatives=[self._speech_data(event, segments, text)]
if segments and float(segments[-1].get("end_time", 0)) > 0
else [],
Comment on lines +383 to +385

@devin-ai-integration devin-ai-integration Bot Oct 20, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Baseten empty finals discard speech timing

When Baseten sends timed segments before an empty final, segments is empty at the endpoint. The endpoint loses the preceding segment's end time and falls back to arrival timing.

Learn more

The receiver keeps speaking across non-final messages, but it only reads segments from the current message. An empty commit-triggered final can therefore close a turn after an earlier timed transcription without carrying that earlier timestamp. The new endpoint alternatives stay empty.

Example: A non-final transcription supplies a segment ending at 0.4 seconds; the subsequent final has segments=[]. END_OF_SPEECH has no timestamp, although the preceding message supplied one.

Recommended fix: Keep the last valid segment end while the turn is open and use it for an empty final. Clear it when the turn closes so a later turn cannot inherit it.

Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

)
)
speaking = False
# Only the commit-triggered final ends the session; VAD
Expand Down
1 change: 1 addition & 0 deletions livekit-plugins/livekit-plugins-baseten/livekit/plugins/baseten/stt.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -595,6 +595,7 @@ async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
while True:
try:
ws = await self._connect_ws()
self._input_ch.restart_provider_clock()
tasks = [
asyncio.create_task(send_task(ws)),
asyncio.create_task(recv_task(ws)),
Expand Down
21 changes: 10 additions & 11 deletions ...t-plugins-cartesia/livekit/plugins/cartesia/_recognize_streams/legacy_recognize_stream.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -166,7 +166,6 @@ def __init__(
self._reconnect_event = asyncio.Event()
self._speaking = False
self._speech_duration: float = 0
self._last_speech_end_time: float = 0

async def _run(self) -> None:
if self._input_ch.closed:
Expand All @@ -177,7 +176,6 @@ async def _run(self) -> None:
# invocation by the base class) starts fresh.
self._speaking = False
self._speech_duration = 0
self._last_speech_end_time = 0

async def keepalive_task(ws: aiohttp.ClientWebSocketResponse) -> None:
try:
Expand Down Expand Up @@ -265,6 +263,7 @@ async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
while True:
try:
ws = await self._connect_ws()
self._input_ch.restart_provider_clock()

send = asyncio.create_task(send_task(ws))
recv = asyncio.create_task(recv_task(ws))
Expand Down Expand Up @@ -367,12 +366,7 @@ def _process_stream_event(self, data: STTEventMessage) -> None:
)
for word in words
]
# word timestamps are often within the audio window, so we track time separately
if self._last_speech_end_time == 0.0:
self._last_speech_end_time = self.start_time_offset
start_time = self._last_speech_end_time
end_time = start_time + data.get("duration", 0)
self._last_speech_end_time = end_time
end_time = words[-1].get("end", 0) if words else 0.0
is_final = data["is_final"]

if not text and not is_final:
Expand All @@ -388,8 +382,10 @@ def _process_stream_event(self, data: STTEventMessage) -> None:

speech_data = stt.SpeechData(
language=self._language or LanguageCode("en"),
start_time=start_time,
end_time=end_time,
start_time=words[0].get("start", 0) + self.start_time_offset
if end_time > 0
else 0.0,
end_time=end_time + self.start_time_offset if end_time > 0 else 0.0,
text=text,
words=timed_words,
)
Expand All @@ -416,7 +412,10 @@ def _process_stream_event(self, data: STTEventMessage) -> None:

if self._speaking:
self._speaking = False
end_event = stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH)
end_event = stt.SpeechEvent(
type=stt.SpeechEventType.END_OF_SPEECH,
alternatives=[speech_data] if end_time > 0 else [],
)
self._event_ch.send_nowait(end_event)
else:
event = stt.SpeechEvent(
Expand Down
33 changes: 31 additions & 2 deletions livekit-plugins/livekit-plugins-deepgram/livekit/plugins/deepgram/stt.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -458,6 +458,7 @@ class SpeechStream(stt.SpeechStream):
_KEEPALIVE_MSG: str = json.dumps({"type": "KeepAlive"})
_CLOSE_MSG: str = json.dumps({"type": "CloseStream"})
_FINALIZE_MSG: str = json.dumps({"type": "Finalize"})
_final_alternatives: list[stt.SpeechData]

def __init__(
self,
Expand All @@ -481,6 +482,7 @@ def __init__(
self._session = http_session
self._opts.endpoint_url = base_url
self._speaking = False
self._final_alternatives = []
self._audio_duration_collector = PeriodicCollector(
callback=self._on_audio_duration_report,
duration=5.0,
Expand Down Expand Up @@ -584,6 +586,7 @@ def update_options(
self._reconnect_event.set()

def _on_end_of_speech(self) -> None:
self._final_alternatives = []
if self._pending_keyterm is not None:
self.update_options(keyterm=self._pending_keyterm)
self._pending_keyterm = None
Expand Down Expand Up @@ -701,6 +704,8 @@ async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
conn_start_time = 0.0
try:
ws = await self._connect_ws()
self._input_ch.restart_provider_clock()
self._final_alternatives = []
conn_start_time = time.perf_counter()
tasks = [
asyncio.create_task(send_task(ws)),
Expand Down Expand Up @@ -873,6 +878,8 @@ def _process_stream_event(self, data: dict) -> None:
self._event_ch.send_nowait(start_event)

if is_final_transcript:
words = data["channel"]["alternatives"][0]["words"]
self._final_alternatives = alts if words and words[-1].get("end", 0) > 0 else []
final_event = stt.SpeechEvent(
type=stt.SpeechEventType.FINAL_TRANSCRIPT,
request_id=request_id,
Expand All @@ -892,15 +899,37 @@ def _process_stream_event(self, data: dict) -> None:
# a non-empty transcript (deepgram doesn't have a SpeechEnded event)
if is_endpoint and self._speaking:
self._speaking = False
self._event_ch.send_nowait(stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH))
# An empty endpoint follows the last non-empty final on some turns.
self._event_ch.send_nowait(
stt.SpeechEvent(
type=stt.SpeechEventType.END_OF_SPEECH,
alternatives=self._final_alternatives,
)
)
self._on_end_of_speech()

elif data["type"] == "UtteranceEnd":
# Fired when utterance_end_ms is set and the configured silence duration has elapsed.
# https://developers.deepgram.com/docs/understand-endpointing-interim-results
last_word_end = data.get("last_word_end", 0)
if last_word_end < 0:
return # The utterance was already finalized.
if self._speaking:
self._speaking = False
self._event_ch.send_nowait(stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH))
self._event_ch.send_nowait(
stt.SpeechEvent(
type=stt.SpeechEventType.END_OF_SPEECH,
alternatives=[
stt.SpeechData(
language=LanguageCode(self._opts.language),
text="",
end_time=self.start_time_offset + last_word_end,
)
]
if last_word_end > 0
else self._final_alternatives,
)
)
self._on_end_of_speech()
elif data["type"] == "Metadata":
pass # metadata is too noisy
Expand Down
24 changes: 19 additions & 5 deletions livekit-plugins/livekit-plugins-deepgram/livekit/plugins/deepgram/stt_v2.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
import os
import weakref
from collections.abc import Sequence
from dataclasses import dataclass
from dataclasses import dataclass, replace
from typing import Any, Literal

import aiohttp
Expand Down Expand Up @@ -563,6 +563,7 @@ async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
while True:
try:
ws = await self._connect_ws()
self._input_ch.restart_provider_clock()
# expose the connection for in-band Configure updates
self._ws = ws
tasks = [
Expand Down Expand Up @@ -678,7 +679,9 @@ def _on_audio_duration_report(self, duration: float) -> None:
)
self._event_ch.send_nowait(usage_event)

def _send_transcript_event(self, event_type: stt.SpeechEventType, data: dict) -> None:
def _send_transcript_event(
self, event_type: stt.SpeechEventType, data: dict
) -> list[stt.SpeechData]:
alts = _parse_transcription(self._opts.language, data, self.start_time_offset)
if alts:
event = stt.SpeechEvent(
Expand All @@ -687,6 +690,7 @@ def _send_transcript_event(self, event_type: stt.SpeechEventType, data: dict) ->
alternatives=alts,
)
self._event_ch.send_nowait(event)
return alts

def _process_stream_event(self, data: dict) -> None:
assert self._opts.language is not None
Expand Down Expand Up @@ -732,9 +736,19 @@ def _process_stream_event(self, data: dict) -> None:

self._speaking = False

self._send_transcript_event(stt.SpeechEventType.FINAL_TRANSCRIPT, data)

end_event = stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH)
alts = self._send_transcript_event(stt.SpeechEventType.FINAL_TRANSCRIPT, data)
words = data.get("words") or []
last_word_end = words[-1].get("end", 0) if words else 0

end_event = stt.SpeechEvent(
type=stt.SpeechEventType.END_OF_SPEECH,
alternatives=[
replace(alt, end_time=self.start_time_offset + last_word_end)
for alt in alts
]
if last_word_end > 0
else [],
)
self._event_ch.send_nowait(end_event)

elif data["type"] == "ConfigureSuccess":
Expand Down
Loading
Loading