fix(stt): apply speech timestamp fixes onto the connection refactor by chenghao-mou · Pull Request #8333 · livekit/agents

chenghao-mou · 2026-11-02T21:12:06Z

Ports #7795 and #7796 onto #8147, plus #7739, which they build on. Nothing else from main is included. Your coordinator stays; each provider connection restarts the input clock.

Where ours wins:

RTZR detects idle in the send loop (no watchdog), so no audio is read while a segment drains.
Gladia and Reson8 re-anchor per session instead of summing audio or shifting start_time_offset; Gladia's queued-audio test now expects capture time.

Initial prompt and agent context

Model: Claude Opus 5.5

I am going to merge my two PRs, but DZ has a huge PR for STT refactoring too, can we open a PR for the merge conflicts targeting his PR once I merged mine?

we should take his approach whenever possible

so we included all commits in main? how about we just limit the changes to from my PRs

Chenghao chose our approach for Gladia session timing and RTZR idle handling after reviewing the tradeoff.

🤖 Generated with Claude Code

CLAassistant · 2026-11-02T21:12:19Z

All committers have signed the CLA.

Port #7795 and #7796 onto the shared STT connection coordinator, with #7739 that they build on. Each provider connection restarts the input clock. RTZR detects idle in the send loop, and Gladia and Reson8 re-anchor each session on its first frame. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

github-actions · 2026-11-02T21:20:47Z

STT Test Results

Status: ✗ Some tests failed

Metric	Count
✓ Passed	41
✗ Failed	13
× Errors	0
→ Skipped	46
▣ Total	100
⏱ Duration	475.5s

Failed Tests

tests.test_stt::test_stream[livekit.plugins.fireworksai]

livekit.agents._exceptions.APIStatusError: message='Invalid response status', status_code=503, retryable=True

The above exception was the direct cause of the following exception:

stt_factory = <function parameter_factory.<locals>.<lambda> at 0x70fd82b48220>
request = <FixtureRequest for <Coroutine test_stream[livekit.plugins.fireworksai]>>

    @pytest.mark.usefixtures("job_process")
    @pytest.mark.parametrize("stt_factory", STTs + STREAM_ONLY_STTs)
    async def test_stream(stt_factory: Callable[[], STT], request):
        plugin_id = request.node.callspec.id
        frames, transcript, _ = await make_test_speech(
            chunk_duration_ms=STREAM_CHUNK_DURATION_MS, sample_rate=SAMPLE_RATE
        )
        silence = rtc.AudioFrame.create(
            sample_rate=SAMPLE_RATE,
            num_channels=1,
            samples_per_channel=SAMPLE_RATE * STREAM_CHUNK_DURATION_MS // 1000,
        )
        frames.extend([silence] * (TRAILING_SILENCE_DURATION_MS // STREAM_CHUNK_DURATION_MS))
  
        # TODO: differentiate missing key vs other errors
        try:
            stt_instance: STT = stt_factory()
        except ValueError as e:
            pytest.skip(f"{plugin_id}: {e}")
  
        async with stt_instance as stt:
            label = f"{stt.model}@{stt.provider}"
            if not stt.capabilities.streaming:
                pytest.skip(f"{label} does not support streaming")
  
            for attempt in range(MAX_RETRIES):
                try:
                    started_at = time.monotonic()
                    async with stt.stream() as stream:
                        async with asyncio.timeout(120):
>                           events, audio_started_at = await collect_stream_events(stream, frames)
                                                       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

tests/test_stt.py:294: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

stream = <livekit.plugins.fireworksai.stt.Spe

tests.test_stt::test_stream[livekit.plugins.mistralai.realtime_v3]

self = <mistralai.extra.realtime.transcription.RealtimeTranscription object at 0x72fbd24e0500>
model = 'voxtral-mini-transcribe-realtime-3', audio_format = None
target_streaming_delay_ms = None, server_url = None, timeout_ms = None
http_headers = {'accept': '*/*', 'accept-encoding': 'gzip, deflate', 'connection': 'keep-alive', 'user-agent': 'python-httpx2/2.13.1'}

    async def connect(
        self,
        model: str,
        audio_format: Optional[AudioFormat] = None,
        target_streaming_delay_ms: Optional[int] = None,
        server_url: Optional[str] = None,
        timeout_ms: Optional[int] = None,
        http_headers: Optional[Mapping[str, str]] = None,
    ) -> RealtimeConnection:
        if timeout_ms is None:
            timeout_ms = self._sdk_config.timeout_ms
  
        security = self._sdk_config.security
        if security is not None and callable(security):
            security = security()
  
        resolved_security = get_security_from_env(security, models.Security)
  
        headers: dict[str, str] = {}
        query_params: dict[str, str] = {}
  
        if resolved_security is not None:
            security_headers, security_query = get_security(resolved_security)
            headers |= security_headers
            for key, values in security_query.items():
                if values:
                    query_params[key] = values[-1]
  
        # The websocket handshake bypasses the SDK's before-request hooks, so service-account
        # auth has to be applied here too, with the same precedence.
        caller_authorization = any(
            name.lower() == "authorization" for name in (http_headers or {})
        )
        if self._sdk_config.security is None and not caller_authorization:
            token = read_service_account_token()
            if token is not None:
                headers["Authorization"] = bearer_header(token)
  
        if http_headers is not None:
            headers |= dict(http_headers)

tests.test_stt::test_stream[livekit.plugins.nvidia]

self = <livekit.plugins.nvidia.stt.SpeechStream object at 0x70fd912af020>
config = config {
  encoding: LINEAR_PCM
  sample_rate_hertz: 16000
  language_code: "en-US"
  max_alternatives: 1
  audio_chan...automatic_punctuation: true
  model: "parakeet-1.1b-en-US-asr-streaming-silero-vad-sortformer"
}
interim_results: true

audio_queue = <queue.Queue object at 0x70fd912f4ce0>
done = <Future finished result=<_MultiThread...tion failed"
>>
service = <riva.client.asr.ASRService object at 0x70fd912f4740>
drain_started = <asyncio.locks.Event object at 0x70fd912f4da0 [unset]>

    def _recognition_worker(self, config, audio_queue, done, service, drain_started) -> None:
        def audio():
            yield from iter(audio_queue.get, None)
            self._event_loop.call_soon_threadsafe(drain_started.set)
  
        error = None
        try:
            responses = service.streaming_response_generator(audio(), config)
>           for response in responses:
                            ^^^^^^^^^

livekit-plugins/livekit-plugins-nvidia/livekit/plugins/nvidia/stt.py:241: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <riva.client.asr.ASRService object at 0x70fd912f4740>
audio_chunks = <generator object SpeechStream._recognition_worker.<locals>.audio at 0x70fd912ede00>
streaming_config = config {
  encoding: LINEAR_PCM
  sample_rate_hertz: 16000
  language_code: "en-US"
  max_alternatives: 1
  audio_chan...automatic_punctuation: true
  model: "parakeet-1.1b-en-US-asr-streaming-silero-vad-sortformer"
}
interim_results: true


    def streaming_response_generator(
        self, audio_chunks: Iterable[bytes], streaming_config: rasr.StreamingRecognitionConfig
    ) -> Generator[rasr.StreamingRecognizeResponse, None, None]:
        """
        Generates speech recognition responses for fragments of speech audio in :param:`audio_chunks`.
        The purpose of the method is to perform speech recognition "online" - as soon as
        audio

tests.test_stt::test_stream[livekit.plugins.mistralai.realtime]

self = <mistralai.extra.realtime.transcription.RealtimeTranscription object at 0x7c8320a42e40>
model = 'voxtral-mini-transcribe-realtime-2602', audio_format = None
target_streaming_delay_ms = None, server_url = None, timeout_ms = None
http_headers = {'accept': '*/*', 'accept-encoding': 'gzip, deflate', 'connection': 'keep-alive', 'user-agent': 'python-httpx2/2.13.1'}

    async def connect(
        self,
        model: str,
        audio_format: Optional[AudioFormat] = None,
        target_streaming_delay_ms: Optional[int] = None,
        server_url: Optional[str] = None,
        timeout_ms: Optional[int] = None,
        http_headers: Optional[Mapping[str, str]] = None,
    ) -> RealtimeConnection:
        if timeout_ms is None:
            timeout_ms = self._sdk_config.timeout_ms
  
        security = self._sdk_config.security
        if security is not None and callable(security):
            security = security()
  
        resolved_security = get_security_from_env(security, models.Security)
  
        headers: dict[str, str] = {}
        query_params: dict[str, str] = {}
  
        if resolved_security is not None:
            security_headers, security_query = get_security(resolved_security)
            headers |= security_headers
            for key, values in security_query.items():
                if values:
                    query_params[key] = values[-1]
  
        # The websocket handshake bypasses the SDK's before-request hooks, so service-account
        # auth has to be applied here too, with the same precedence.
        caller_authorization = any(
            name.lower() == "authorization" for name in (http_headers or {})
        )
        if self._sdk_config.security is None and not caller_authorization:
            token = read_service_account_token()
            if token is not None:
                headers["Authorization"] = bearer_header(token)
  
        if http_headers is not None:
            headers |= dict(http_headers)

tests.test_stt::test_stream[livekit.plugins.mistralai.realtime_v3_vad]

self = <mistralai.extra.realtime.transcription.RealtimeTranscription object at 0x7c8320ac79e0>
model = 'voxtral-mini-transcribe-realtime-3', audio_format = None
target_streaming_delay_ms = None, server_url = None, timeout_ms = None
http_headers = {'accept': '*/*', 'accept-encoding': 'gzip, deflate', 'connection': 'keep-alive', 'user-agent': 'python-httpx2/2.13.1'}

    async def connect(
        self,
        model: str,
        audio_format: Optional[AudioFormat] = None,
        target_streaming_delay_ms: Optional[int] = None,
        server_url: Optional[str] = None,
        timeout_ms: Optional[int] = None,
        http_headers: Optional[Mapping[str, str]] = None,
    ) -> RealtimeConnection:
        if timeout_ms is None:
            timeout_ms = self._sdk_config.timeout_ms
  
        security = self._sdk_config.security
        if security is not None and callable(security):
            security = security()
  
        resolved_security = get_security_from_env(security, models.Security)
  
        headers: dict[str, str] = {}
        query_params: dict[str, str] = {}
  
        if resolved_security is not None:
            security_headers, security_query = get_security(resolved_security)
            headers |= security_headers
            for key, values in security_query.items():
                if values:
                    query_params[key] = values[-1]
  
        # The websocket handshake bypasses the SDK's before-request hooks, so service-account
        # auth has to be applied here too, with the same precedence.
        caller_authorization = any(
            name.lower() == "authorization" for name in (http_headers or {})
        )
        if self._sdk_config.security is None and not caller_authorization:
            token = read_service_account_token()
            if token is not None:
                headers["Authorization"] = bearer_header(token)
  
        if http_headers is not None:
            headers |= dict(http_headers)

tests.test_stt_stream_live::test_live_stream_flush_idle_and_completion[gradium]

provider_config = Provider(live_constructor='livekit.plugins.gradium.stt:STT', required_env=('GRADIUM_API_KEY',), live_options={}, live_...otstrap=None, tls=False, expected_attempts_after_audio=3, progress_requires_final=False, reconnect_on_idle_close=False)

    @pytest.mark.usefixtures("job_process")
    @pytest.mark.parametrize(
        "provider_config", providers(), ids=lambda provider_config: provider_config.name
    )
    async def test_live_stream_flush_idle_and_completion(provider_config):
        if not provider_config.live_constructor:
            pytest.skip("live driver is not configured for this variant")
        load_dotenv()
        missing = [name for name in provider_config.required_env if not os.environ.get(name)]
        if missing:
            pytest.skip("Missing live test configuration: " + ", ".join(missing))
  
        async with provider_config.create_live_stt() as provider:
            stream = provider.stream(
                conn_options=APIConnectOptions(
                    max_retry=0,
                    timeout=15,
                    liveness_timeout=6,
                )
            )
            finals = asyncio.Queue()
            received = []
  
            async def collect():
                async for event in stream:
                    if event.type == stt.SpeechEventType.FINAL_TRANSCRIPT:
                        text = event.alternatives[0].text
                        received.append(text)
                        finals.put_nowait(text)
  
            collector = asyncio.create_task(collect())
            try:
                for index, (name, expected) in enumerate(
                    [
                        ("hello_world", {"hello", "world"}),
                        ("weather_question", {"weather", "paris"}),
                    ]
                ):
                    await _send_utterance(stream, name)
                    stream.flush()
  
                    async def wait_for_words(expected=expected):

tests.test_stt_stream_live::test_live_stream_flush_idle_and_completion[fireworksai]

provider_config = Provider(live_constructor='livekit.plugins.fireworksai.stt:STT', required_env=('FIREWORKS_API_KEY',), live_options={},...otstrap=None, tls=False, expected_attempts_after_audio=3, progress_requires_final=False, reconnect_on_idle_close=False)

    @pytest.mark.usefixtures("job_process")
    @pytest.mark.parametrize(
        "provider_config", providers(), ids=lambda provider_config: provider_config.name
    )
    async def test_live_stream_flush_idle_and_completion(provider_config):
        if not provider_config.live_constructor:
            pytest.skip("live driver is not configured for this variant")
        load_dotenv()
        missing = [name for name in provider_config.required_env if not os.environ.get(name)]
        if missing:
            pytest.skip("Missing live test configuration: " + ", ".join(missing))
  
        async with provider_config.create_live_stt() as provider:
            stream = provider.stream(
                conn_options=APIConnectOptions(
                    max_retry=0,
                    timeout=15,
                    liveness_timeout=6,
                )
            )
            finals = asyncio.Queue()
            received = []
  
            async def collect():
                async for event in stream:
                    if event.type == stt.SpeechEventType.FINAL_TRANSCRIPT:
                        text = event.alternatives[0].text
                        received.append(text)
                        finals.put_nowait(text)
  
            collector = asyncio.create_task(collect())
            try:
                for index, (name, expected) in enumerate(
                    [
                        ("hello_world", {"hello", "world"}),
                        ("weather_question", {"weather", "paris"}),
                    ]
                ):
                    await _send_utterance(stream, name)
                    stream.flush()
  
                    async def wait_for_words(expected=expected):

tests.test_stt_stream_live::test_live_stream_flush_idle_and_completion[mistralai]

self = <mistralai.extra.realtime.transcription.RealtimeTranscription object at 0x72fbd202bd70>
model = 'voxtral-mini-transcribe-realtime-2602', audio_format = None
target_streaming_delay_ms = None, server_url = None, timeout_ms = None
http_headers = {'accept': '*/*', 'accept-encoding': 'gzip, deflate', 'connection': 'keep-alive', 'user-agent': 'python-httpx2/2.13.1'}

    async def connect(
        self,
        model: str,
        audio_format: Optional[AudioFormat] = None,
        target_streaming_delay_ms: Optional[int] = None,
        server_url: Optional[str] = None,
        timeout_ms: Optional[int] = None,
        http_headers: Optional[Mapping[str, str]] = None,
    ) -> RealtimeConnection:
        if timeout_ms is None:
            timeout_ms = self._sdk_config.timeout_ms
  
        security = self._sdk_config.security
        if security is not None and callable(security):
            security = security()
  
        resolved_security = get_security_from_env(security, models.Security)
  
        headers: dict[str, str] = {}
        query_params: dict[str, str] = {}
  
        if resolved_security is not None:
            security_headers, security_query = get_security(resolved_security)
            headers |= security_headers
            for key, values in security_query.items():
                if values:
                    query_params[key] = values[-1]
  
        # The websocket handshake bypasses the SDK's before-request hooks, so service-account
        # auth has to be applied here too, with the same precedence.
        caller_authorization = any(
            name.lower() == "authorization" for name in (http_headers or {})
        )
        if self._sdk_config.security is None and not caller_authorization:
            token = read_service_account_token()
            if token is not None:
                headers["Authorization"] = bearer_header(token)
  
        if http_headers is not None:
            headers |= dict(http_headers)

tests.test_stt_stream_live::test_live_stream_flush_idle_and_completion[mistralai_voxtral3]

self = <mistralai.extra.realtime.transcription.RealtimeTranscription object at 0x72fbd204e750>
model = 'voxtral-mini-transcribe-realtime-3', audio_format = None
target_streaming_delay_ms = None, server_url = None, timeout_ms = None
http_headers = {'accept': '*/*', 'accept-encoding': 'gzip, deflate', 'connection': 'keep-alive', 'user-agent': 'python-httpx2/2.13.1'}

    async def connect(
        self,
        model: str,
        audio_format: Optional[AudioFormat] = None,
        target_streaming_delay_ms: Optional[int] = None,
        server_url: Optional[str] = None,
        timeout_ms: Optional[int] = None,
        http_headers: Optional[Mapping[str, str]] = None,
    ) -> RealtimeConnection:
        if timeout_ms is None:
            timeout_ms = self._sdk_config.timeout_ms
  
        security = self._sdk_config.security
        if security is not None and callable(security):
            security = security()
  
        resolved_security = get_security_from_env(security, models.Security)
  
        headers: dict[str, str] = {}
        query_params: dict[str, str] = {}
  
        if resolved_security is not None:
            security_headers, security_query = get_security(resolved_security)
            headers |= security_headers
            for key, values in security_query.items():
                if values:
                    query_params[key] = values[-1]
  
        # The websocket handshake bypasses the SDK's before-request hooks, so service-account
        # auth has to be applied here too, with the same precedence.
        caller_authorization = any(
            name.lower() == "authorization" for name in (http_headers or {})
        )
        if self._sdk_config.security is None and not caller_authorization:
            token = read_service_account_token()
            if token is not None:
                headers["Authorization"] = bearer_header(token)
  
        if http_headers is not None:
            headers |= dict(http_headers)

tests.test_stt_stream_live::test_live_stream_flush_idle_and_completion[mistralai_voxtral3_diarized]

self = <mistralai.extra.realtime.transcription.RealtimeTranscription object at 0x72fbd207d190>
model = 'voxtral-mini-transcribe-realtime-3', audio_format = None
target_streaming_delay_ms = None, server_url = None, timeout_ms = None
http_headers = {'accept': '*/*', 'accept-encoding': 'gzip, deflate', 'connection': 'keep-alive', 'user-agent': 'python-httpx2/2.13.1'}

    async def connect(
        self,
        model: str,
        audio_format: Optional[AudioFormat] = None,
        target_streaming_delay_ms: Optional[int] = None,
        server_url: Optional[str] = None,
        timeout_ms: Optional[int] = None,
        http_headers: Optional[Mapping[str, str]] = None,
    ) -> RealtimeConnection:
        if timeout_ms is None:
            timeout_ms = self._sdk_config.timeout_ms
  
        security = self._sdk_config.security
        if security is not None and callable(security):
            security = security()
  
        resolved_security = get_security_from_env(security, models.Security)
  
        headers: dict[str, str] = {}
        query_params: dict[str, str] = {}
  
        if resolved_security is not None:
            security_headers, security_query = get_security(resolved_security)
            headers |= security_headers
            for key, values in security_query.items():
                if values:
                    query_params[key] = values[-1]
  
        # The websocket handshake bypasses the SDK's before-request hooks, so service-account
        # auth has to be applied here too, with the same precedence.
        caller_authorization = any(
            name.lower() == "authorization" for name in (http_headers or {})
        )
        if self._sdk_config.security is None and not caller_authorization:
            token = read_service_account_token()
            if token is not None:
                headers["Authorization"] = bearer_header(token)
  
        if http_headers is not None:
            headers |= dict(http_headers)

tests.test_stt_stream_live::test_live_stream_flush_idle_and_completion[mistralai_voxtral3_turn_detection]

self = <mistralai.extra.realtime.transcription.RealtimeTranscription object at 0x72fbd1d68f50>
model = 'voxtral-mini-transcribe-realtime-3', audio_format = None
target_streaming_delay_ms = None, server_url = None, timeout_ms = None
http_headers = {'accept': '*/*', 'accept-encoding': 'gzip, deflate', 'connection': 'keep-alive', 'user-agent': 'python-httpx2/2.13.1'}

    async def connect(
        self,
        model: str,
        audio_format: Optional[AudioFormat] = None,
        target_streaming_delay_ms: Optional[int] = None,
        server_url: Optional[str] = None,
        timeout_ms: Optional[int] = None,
        http_headers: Optional[Mapping[str, str]] = None,
    ) -> RealtimeConnection:
        if timeout_ms is None:
            timeout_ms = self._sdk_config.timeout_ms
  
        security = self._sdk_config.security
        if security is not None and callable(security):
            security = security()
  
        resolved_security = get_security_from_env(security, models.Security)
  
        headers: dict[str, str] = {}
        query_params: dict[str, str] = {}
  
        if resolved_security is not None:
            security_headers, security_query = get_security(resolved_security)
            headers |= security_headers
            for key, values in security_query.items():
                if values:
                    query_params[key] = values[-1]
  
        # The websocket handshake bypasses the SDK's before-request hooks, so service-account
        # auth has to be applied here too, with the same precedence.
        caller_authorization = any(
            name.lower() == "authorization" for name in (http_headers or {})
        )
        if self._sdk_config.security is None and not caller_authorization:
            token = read_service_account_token()
            if token is not None:
                headers["Authorization"] = bearer_header(token)
  
        if http_headers is not None:
            headers |= dict(http_headers)

tests.test_stt_stream_live::test_live_stream_flush_idle_and_completion[assemblyai_universal36]

provider_config = Provider(live_constructor='livekit.plugins.assemblyai.stt:STT', required_env=('ASSEMBLYAI_API_KEY',), live_options={'m...otstrap=None, tls=False, expected_attempts_after_audio=3, progress_requires_final=False, reconnect_on_idle_close=False)

    @pytest.mark.usefixtures("job_process")
    @pytest.mark.parametrize(
        "provider_config", providers(), ids=lambda provider_config: provider_config.name
    )
    async def test_live_stream_flush_idle_and_completion(provider_config):
        if not provider_config.live_constructor:
            pytest.skip("live driver is not configured for this variant")
        load_dotenv()
        missing = [name for name in provider_config.required_env if not os.environ.get(name)]
        if missing:
            pytest.skip("Missing live test configuration: " + ", ".join(missing))
  
        async with provider_config.create_live_stt() as provider:
            stream = provider.stream(
                conn_options=APIConnectOptions(
                    max_retry=0,
                    timeout=15,
                    liveness_timeout=6,
                )
            )
            finals = asyncio.Queue()
            received = []
  
            async def collect():
                async for event in stream:
                    if event.type == stt.SpeechEventType.FINAL_TRANSCRIPT:
                        text = event.alternatives[0].text
                        received.append(text)
                        finals.put_nowait(text)
  
            collector = asyncio.create_task(collect())
            try:
                for index, (name, expected) in enumerate(
                    [
                        ("hello_world", {"hello", "world"}),
                        ("weather_question", {"weather", "paris"}),
                    ]
                ):
>                   await _send_utterance(stream, name)

tests/test_stt_stream_live.py:78: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

tests.test_stt_stream_live::test_live_stream_flush_idle_and_completion[soniox]

provider_config = Provider(live_constructor='livekit.plugins.soniox.stt:STT', required_env=('SONIOX_API_KEY',), live_options={}, live_op...otstrap=None, tls=False, expected_attempts_after_audio=3, progress_requires_final=False, reconnect_on_idle_close=False)

    @pytest.mark.usefixtures("job_process")
    @pytest.mark.parametrize(
        "provider_config", providers(), ids=lambda provider_config: provider_config.name
    )
    async def test_live_stream_flush_idle_and_completion(provider_config):
        if not provider_config.live_constructor:
            pytest.skip("live driver is not configured for this variant")
        load_dotenv()
        missing = [name for name in provider_config.required_env if not os.environ.get(name)]
        if missing:
            pytest.skip("Missing live test configuration: " + ", ".join(missing))
  
        async with provider_config.create_live_stt() as provider:
            stream = provider.stream(
                conn_options=APIConnectOptions(
                    max_retry=0,
                    timeout=15,
                    liveness_timeout=6,
                )
            )
            finals = asyncio.Queue()
            received = []
  
            async def collect():
                async for event in stream:
                    if event.type == stt.SpeechEventType.FINAL_TRANSCRIPT:
                        text = event.alternatives[0].text
                        received.append(text)
                        finals.put_nowait(text)
  
            collector = asyncio.create_task(collect())
            try:
                for index, (name, expected) in enumerate(
                    [
                        ("hello_world", {"hello", "world"}),
                        ("weather_question", {"weather", "paris"}),
                    ]
                ):
                    await _send_utterance(stream, name)
                    stream.flush()
  
                    async def wait_for_words(expected=expected):

Skipped Tests

Test	Reason
`tests.test_stt::test_recognize[livekit.plugins.assemblyai]`	universal-3-6-pro@AssemblyAI does not support batch recognition
`tests.test_stt::test_recognize[livekit.plugins.speechmatics]`	linden-1@Speechmatics does not support batch recognition
`tests.test_stt::test_recognize[livekit.plugins.soniox]`	stt-rt-v5@Soniox does not support batch recognition
`tests.test_stt::test_recognize[livekit.agents.inference]`	unknown@livekit does not support batch recognition
`tests.test_stt::test_recognize[livekit.plugins.azure]`	unknown@Azure STT does not support batch recognition
`tests.test_stt::test_recognize[livekit.plugins.fireworksai]`	unknown@FireworksAI does not support batch recognition
`tests.test_stt::test_recognize[livekit.plugins.aws]`	unknown@Amazon Transcribe does not support batch recognition
`tests.test_stt::test_recognize[livekit.plugins.reson8]`	livekit.plugins.reson8: Reson8 API key is required, either as argument or RESON8_API_KEY env var
`tests.test_stt::test_recognize[livekit.plugins.mistralai.realtime_v3]`	voxtral-mini-transcribe-realtime-3@MistralAI does not support batch recognition
`tests.test_stt::test_recognize[livekit.plugins.cartesia]`	ink-2@Cartesia does not support batch recognition
`tests.test_stt::test_stream[livekit.plugins.elevenlabs]`	scribe_v1@ElevenLabs does not support streaming
`tests.test_stt::test_recognize[livekit.plugins.cartesia._legacy]`	ink-whisper@Cartesia does not support batch recognition
`tests.test_stt::test_recognize[livekit.plugins.deepgram.STTv2]`	flux-general-en@Deepgram does not support batch recognition
`tests.test_stt::test_recognize[livekit.plugins.gradium.STT]`	default@Gradium does not support batch recognition
`tests.test_stt::test_recognize[livekit.plugins.nvidia]`	unknown@unknown does not support batch recognition
`tests.test_stt::test_stream[livekit.plugins.fal]`	Wizper@Fal does not support streaming
`tests.test_stt::test_stream[livekit.plugins.openai]`	gpt-4o-mini-transcribe@api.openai.com does not support streaming
`tests.test_stt::test_stream[livekit.plugins.rtzr]`	livekit.plugins.rtzr: RTZR_CLIENT_ID and RTZR_CLIENT_SECRET must be set
`tests.test_stt::test_stream[livekit.plugins.reson8]`	livekit.plugins.reson8: Reson8 API key is required, either as argument or RESON8_API_KEY env var
`tests.test_stt::test_stream[livekit.plugins.xai]`	livekit.plugins.xai: xAI API key is required
...	... and 26 more

Triggered by workflow run #5700

davidzhao

nice work.

devin-ai-integration

Devin Review found 1 potential issue.

1 flag not posted on this PR by your GitHub settings — view it in Devin Review. (Configure)

devin-ai-integration · 2026-11-10T06:40:38Z

remove_padding

+        while self._pending_padding:
+            start, duration = self._pending_padding[0]
+            if start + duration > provider_time:
+                break
+            self._settled_padding += duration
+            self._pending_padding.popleft()


🟡 Late word boundaries shift before speech

When a later timestamp precedes a delayed end_text, remove_padding subtracts settled silence from the earlier word boundary. The speech end moves backward, sometimes before the stream starts.

Learn more

Gradium's text messages map start_s through remove_padding, while end_text carries the previous word's stop_s for the final boundary here. A newer text timestamp can settle a synthetic silence interval before the previous word boundary arrives. The settled amount then gets subtracted from the earlier boundary even though that boundary predates the silence.

Example: Add one second of audio and 0.48 seconds of padding. Look up 1.6 seconds, then the delayed word boundary at 0.4 seconds. The second lookup returns -0.08 seconds instead of 0.4 seconds.

Recommended fix: Retain enough history to resolve earlier provider timestamps, or defer settling padding until no earlier word boundaries can arrive. Cover a text timestamp after padding followed by an earlier end_text boundary.

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

chenghao-mou added the review_effort:high Needs broad context and substantial validation label Nov 2, 2026

chenghao-mou force-pushed the chenghao/merge-main-into-8147 branch from 7f7337e to bab5a95 Compare November 2, 2026 21:20

chenghao-mou changed the title ~~chore: merge main into the stt connection refactor~~ fix(stt): apply speech timestamp fixes onto the connection refactor Nov 2, 2026

AboodFares mentioned this pull request Nov 7, 2026

ci(test-stt): propagate pytest exit code #8432

Closed

davidzhao approved these changes Nov 10, 2026

View reviewed changes

linear-code Bot marked this pull request as ready for review November 10, 2026 06:39

linear-code Bot requested a review from a team as a code owner November 10, 2026 06:39

devin-ai-integration Bot reviewed Nov 10, 2026

View reviewed changes

fix(stt): align multispeaker adapter output timestamps

cc9e8b9

Navigation Menu

Repository navigation

fix(stt): apply speech timestamp fixes onto the connection refactor - #8333

fix(stt): apply speech timestamp fixes onto the connection refactor#8333
chenghao-mou wants to merge 2 commits into
dz/consistent-ws-error-handlingfrom
chenghao/merge-main-into-8147

chenghao-mou commented Nov 2, 2026 •

edited

Loading

Uh oh!

CLAassistant commented Nov 2, 2026 •

edited

Loading

Uh oh!

github-actions Bot commented Nov 2, 2026 •

edited

Loading

Uh oh!

davidzhao left a comment

Uh oh!

devin-ai-integration Bot left a comment

Uh oh!

devin-ai-integration Bot Nov 10, 2026

Uh oh!

This branch has not been deployed

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Conversation

chenghao-mou commented Nov 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

CLAassistant commented Nov 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

github-actions Bot commented Nov 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

STT Test Results

Uh oh!

davidzhao left a comment

Choose a reason for hiding this comment

Uh oh!

devin-ai-integration Bot left a comment

Choose a reason for hiding this comment

Uh oh!

devin-ai-integration Bot Nov 10, 2026

Choose a reason for hiding this comment

Uh oh!

This branch has not been deployed

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

chenghao-mou commented Nov 2, 2026 •

edited

Loading

CLAassistant commented Nov 2, 2026 •

edited

Loading

github-actions Bot commented Nov 2, 2026 •

edited

Loading