fix(voice): make STT the user_state source when turn_detection is stt

livekit-agents/livekit/agents/voice/agent_activity.py

@@ -188,6 +188,11 @@ def __init__(self, agent: Agent, sess: AgentSession) -> None:
         self._cancel_speech_pause_task: asyncio.Task[None] | None = None
 
         self._stt_eos_received: bool = False
+        # True while an STT-driven speech segment is open (paired ev=None
+        # start/end hook calls); lets the end-of-speech gate below tell an
+        # STT-authored "speaking" state apart from one written by another
+        # source (e.g. claim_user_turn), which the STT will never clear
+        self._stt_user_speaking: bool = False
 
         # fired when a speech_task finishes or when a new speech_handle is scheduled
         # this is used to wake up the main task when the scheduling state changes
@@ -1972,7 +1977,14 @@ def on_start_of_speech(
         ev: vad.VADEvent | None,
         speech_start_time: float,
     ) -> None:
-        self._session._update_user_state("speaking", last_speaking_time=speech_start_time)
+        # with STT-driven turn detection, STT speech events (ev is None) are the
+        # authoritative user_state source: VAD stays active for interruption and
+        # endpointing below, but background noise it picks up must not flip
+        # user_state to "speaking" when the STT hears no speech (#5580)
+        if ev is None:
+            self._stt_user_speaking = True
+        if ev is None or self._turn_detection != "stt":
+            self._session._update_user_state("speaking", last_speaking_time=speech_start_time)
         if self._audio_recognition:
             self._audio_recognition._on_start_of_speech(
                 started_at=speech_start_time,
@@ -2019,10 +2031,23 @@ def on_end_of_speech(self, ev: vad.VADEvent | None) -> None:
                 else NOT_GIVEN,
             )
 
-        self._session._update_user_state(
-            "listening",
-            last_speaking_time=speech_end_time,
-        )
+        if ev is None:
+            self._stt_user_speaking = False
+        # in stt mode the VAD end must not clear an STT-authored "speaking"
+        # (the STT end-of-speech will), but "speaking" can also be entered by
+        # writers the STT will never clear - claim_user_turn re-deriving from
+        # VAD silence, or a turn_detection switch mid-speech - so when no
+        # STT-driven segment is open, let the VAD end recover the state
+        # instead of leaving it stuck at "speaking"
+        if (
+            ev is None
+            or self._turn_detection != "stt"
+            or (not self._stt_user_speaking and self._session.user_state == "speaking")
+        ):
+            self._session._update_user_state(
+                "listening",
+                last_speaking_time=speech_end_time,
+            )
         self._user_silence_event.set()
 
         if self._paused_speech:

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

@@ -155,11 +155,18 @@ class _STTPipeline:
     """
 
     def __init__(
-        self, stt_node: io.STTNode, *, is_closing: Callable[[], bool] | None = None
+        self,
+        stt_node: io.STTNode,
+        *,
+        is_closing: Callable[[], bool] | None = None,
+        on_stream_reset: Callable[[], None] | None = None,
     ) -> None:
         self._stt_node = stt_node
         # don't recreate the stream while the session is closing
         self._is_closing = is_closing or (lambda: False)
+        # notified when the stream is recreated mid-flight: events of the old
+        # stream (e.g. a pending END_OF_SPEECH) are lost with it
+        self._on_stream_reset = on_stream_reset
         self._audio_ch = aio.Chan[rtc.AudioFrame]()
         self._event_ch = aio.Chan[stt.SpeechEvent]()
         self._pump_task = asyncio.create_task(self._stt_pump())
@@ -209,6 +216,8 @@ async def _stt_pump(self) -> None:
                     "STT stream ended on an unrecoverable error, recreating",
                     exc_info=True,
                 )
+                if self._on_stream_reset is not None:
+                    self._on_stream_reset()
                 await asyncio.sleep(_STT_RECONNECT_INTERVAL)
                 # the session may have started closing during the backoff
                 if self._is_closing():
@@ -218,11 +227,15 @@ async def _stt_pump(self) -> None:
             # node ended without error (audio input closed): stop
             return
 
-    def _rebind_node(self, stt_node: io.STTNode) -> None:
+    def _rebind_node(
+        self, stt_node: io.STTNode, *, on_stream_reset: Callable[[], None] | None = None
+    ) -> None:
         # the pipeline outlives the agent that created it (reused across handoff);
         # recreation must call the currently-active node, not the previous agent's
-        # bound node whose activity is torn down (would raise, stopping the pump)
+        # bound node whose activity is torn down (would raise, stopping the pump).
+        # The reset callback is rebound for the same reason.
         self._stt_node = stt_node
+        self._on_stream_reset = on_stream_reset
 
     async def aclose(self) -> None:
         await aio.cancel_and_wait(self._pump_task)
@@ -816,11 +829,15 @@ def _update_stt(
         if reset_context:
             self.stt_context = None
         if pipeline is None and stt is not None:
-            pipeline = _STTPipeline(stt, is_closing=self._session._is_closing)
+            pipeline = _STTPipeline(
+                stt,
+                is_closing=self._session._is_closing,
+                on_stream_reset=self._on_stt_stream_reset,
+            )
         elif pipeline is not None and stt is not None:
             # reused pipeline: rebind to this activity's node so a recreation
             # after an error doesn't call into the previous (torn-down) agent
-            pipeline._rebind_node(stt)
+            pipeline._rebind_node(stt, on_stream_reset=self._on_stt_stream_reset)
 
         if pipeline is not None:
             self._stt_consumer_atask = asyncio.create_task(
@@ -972,7 +989,21 @@ def _detach_turn_detector(self) -> _StreamingTurnDetectorStream | None:
         self._turn_detector_prediction_fut = None
         return stream
 
+    def _on_stt_stream_reset(self) -> None:
+        # the STT stream is being torn down mid-utterance (recreated after a
+        # connection failure, or the user turn was cleared): its pending
+        # END_OF_SPEECH will never arrive. In stt turn-detection mode the STT
+        # drives user_state, so close the open segment like the vad-task
+        # teardown does - otherwise the user stays "speaking" until a later
+        # utterance is fully transcribed
+        if self._turn_detection_mode != "stt" or not self._speaking:
+            return
+        with trace.use_span(self._ensure_user_turn_span()):
+            self._hooks.on_end_of_speech(None)
+        self._speaking = False
+
     def _clear_user_turn(self) -> None:
+        self._on_stt_stream_reset()
         self._audio_transcript = ""
         self._audio_interim_transcript = ""
         self._audio_preflight_transcript = ""

tests/test_stt_user_state_source.py

@@ -0,0 +1,210 @@
+"""Regression tests for #5580: with STT-driven turn detection, VAD must not drive user_state.
+
+When ``turn_detection="stt"``, both VAD and STT used to write ``user_state``.
+In noisy environments VAD flips it to "speaking" on background noise even when
+the STT hears nothing, breaking everything keyed on user state (away timeouts,
+filler triggering) — and the only workaround, ``vad=None``, also gave up
+VAD-based interruption sensing. STT is now the authoritative ``user_state``
+source in that mode, while VAD keeps its interruption/endpointing roles.
+"""
+
+import asyncio
+import time
+from collections.abc import AsyncIterable, AsyncIterator
+
+import pytest
+
+from livekit.agents import APIError, vad
+from livekit.agents.voice.agent_activity import AgentActivity
+from livekit.agents.voice.audio_recognition import AudioRecognition, _STTPipeline
+from livekit.agents.voice.endpointing import BaseEndpointing
+
+from .fake_session import FakeActions, create_session
+from .test_agent_session import MyAgent, _close_test_session
+
+pytestmark = [pytest.mark.unit, pytest.mark.virtual_time, pytest.mark.no_concurrent]
+
+
+def _vad_event(type_: vad.VADEventType) -> vad.VADEvent:
+    return vad.VADEvent(
+        type=type_,
+        samples_index=0,
+        timestamp=time.time(),
+        speech_duration=0.5,
+        silence_duration=0.0,
+    )
+
+
+def _make_activity(turn_detection: str | None) -> AgentActivity:
+    session = create_session(FakeActions(), turn_handling={"turn_detection": turn_detection})
+    return AgentActivity(MyAgent(), session)
+
+
+def _make_recognition(activity: AgentActivity, turn_detection: str) -> AudioRecognition:
+    return AudioRecognition(
+        activity._session,
+        hooks=activity,
+        endpointing=BaseEndpointing(min_delay=0.0, max_delay=0.0),
+        stt=None,
+        vad=None,
+        using_default_vad=False,
+        interruption_detection=None,
+        turn_detection=turn_detection,
+    )
+
+
+class TestSttDrivenUserState:
+    async def test_vad_noise_does_not_flip_user_state_in_stt_mode(self) -> None:
+        activity = _make_activity("stt")
+        session = activity._session
+        try:
+            assert session.user_state == "listening"
+
+            # background noise: VAD fires but the STT hears no speech
+            activity.on_start_of_speech(_vad_event(vad.VADEventType.START_OF_SPEECH), time.time())
+            assert session.user_state == "listening"
+
+            activity.on_end_of_speech(_vad_event(vad.VADEventType.END_OF_SPEECH))
+            assert session.user_state == "listening"
+        finally:
+            await _close_test_session(session)
+
+    async def test_stt_speech_drives_user_state_in_stt_mode(self) -> None:
+        activity = _make_activity("stt")
+        session = activity._session
+        try:
+            # STT-sourced hook calls pass ev=None
+            activity.on_start_of_speech(None, time.time())
+            assert session.user_state == "speaking"
+
+            activity.on_end_of_speech(None)
+            assert session.user_state == "listening"
+        finally:
+            await _close_test_session(session)
+
+    async def test_vad_drives_user_state_in_default_mode(self) -> None:
+        activity = _make_activity(None)
+        session = activity._session
+        try:
+            activity.on_start_of_speech(_vad_event(vad.VADEventType.START_OF_SPEECH), time.time())
+            assert session.user_state == "speaking"
+
+            activity.on_end_of_speech(_vad_event(vad.VADEventType.END_OF_SPEECH))
+            assert session.user_state == "listening"
+        finally:
+            await _close_test_session(session)
+
+    async def test_vad_end_recovers_speaking_written_by_non_stt_source(self) -> None:
+        # "speaking" can be entered by writers the STT will never clear
+        # (claim_user_turn re-derivation, a turn_detection switch mid-speech);
+        # a VAD end-of-speech must recover the state instead of leaving it
+        # stuck at "speaking" with no STT end-of-speech ever coming
+        activity = _make_activity("stt")
+        session = activity._session
+        try:
+            session._update_user_state("speaking", last_speaking_time=time.time())
+            assert session.user_state == "speaking"
+
+            activity.on_end_of_speech(_vad_event(vad.VADEventType.END_OF_SPEECH))
+            assert session.user_state == "listening"
+        finally:
+            await _close_test_session(session)
+
+    async def test_vad_end_does_not_clear_stt_authored_speaking(self) -> None:
+        # the VAD usually endpoints before the STT: its end-of-speech must not
+        # cut short a "speaking" state the STT opened and will close itself
+        activity = _make_activity("stt")
+        session = activity._session
+        try:
+            activity.on_start_of_speech(None, time.time())
+            assert session.user_state == "speaking"
+
+            activity.on_end_of_speech(_vad_event(vad.VADEventType.END_OF_SPEECH))
+            assert session.user_state == "speaking"
+
+            activity.on_end_of_speech(None)
+            assert session.user_state == "listening"
+        finally:
+            await _close_test_session(session)
+
+    async def test_claimed_turn_with_vad_noise_recovers(self) -> None:
+        # background noise trips the VAD during a programmatic (text) turn:
+        # the release re-derives "speaking" from the VAD-driven silence event,
+        # and only the later VAD end-of-speech can clear it (the STT heard
+        # nothing, so no STT end-of-speech will ever arrive)
+        activity = _make_activity("stt")
+        session = activity._session
+        session._activity = activity
+        try:
+            async with session._claim_user_turn():
+                activity.on_start_of_speech(
+                    _vad_event(vad.VADEventType.START_OF_SPEECH), time.time()
+                )
+            assert session.user_state == "speaking"
+
+            activity.on_end_of_speech(_vad_event(vad.VADEventType.END_OF_SPEECH))
+            assert session.user_state == "listening"
+        finally:
+            await _close_test_session(session)
+
+    async def test_clear_user_turn_closes_open_stt_segment(self) -> None:
+        # clear_user_turn() tears down and recreates the STT stream, so a
+        # pending STT end-of-speech will never arrive; the open segment must
+        # be closed on the way out or user_state stays "speaking" forever
+        activity = _make_activity("stt")
+        session = activity._session
+        try:
+            recognition = _make_recognition(activity, "stt")
+            activity.on_start_of_speech(None, time.time())
+            recognition._speaking = True  # as the STT START branch would have set
+            assert session.user_state == "speaking"
+
+            recognition._clear_user_turn()
+
+            assert session.user_state == "listening"
+            assert activity._stt_user_speaking is False
+            assert recognition._speaking is False
+        finally:
+            await _close_test_session(session)
+
+    async def test_clear_user_turn_leaves_state_to_vad_outside_stt_mode(self) -> None:
+        # outside stt mode the VAD end-of-speech still owns the transition;
+        # clearing the turn must not touch user_state
+        activity = _make_activity(None)
+        session = activity._session
+        try:
+            recognition = _make_recognition(activity, "vad")
+            activity.on_start_of_speech(_vad_event(vad.VADEventType.START_OF_SPEECH), time.time())
+            recognition._speaking = True
+            assert session.user_state == "speaking"
+
+            recognition._clear_user_turn()
+            assert session.user_state == "speaking"
+
+            activity.on_end_of_speech(_vad_event(vad.VADEventType.END_OF_SPEECH))
+            assert session.user_state == "listening"
+        finally:
+            await _close_test_session(session)
+
+    async def test_stt_reconnect_notifies_stream_reset(self) -> None:
+        # a reconnect after an APIError drops the in-flight utterance's
+        # END_OF_SPEECH with the old stream; the pipeline must notify its
+        # owner so the open segment can be closed
+        calls: list[int] = []
+
+        def _failing_node(audio_ch: object, settings: object) -> AsyncIterable[object]:
+            async def _gen() -> AsyncIterator[object]:
+                raise APIError("connection dropped")
+                yield  # unreachable; makes this an async generator
+
+            return _gen()
+
+        pipeline = _STTPipeline(_failing_node, on_stream_reset=lambda: calls.append(1))  # type: ignore[arg-type]
+        try:
+            for _ in range(100):
+                if calls:
+                    break
+                await asyncio.sleep(0.01)
+            assert calls, "on_stream_reset was not invoked on reconnect"
+        finally:
+            await pipeline.aclose()