fix(voice): enforce interruption min_duration on STT transcript triggers (#3515)

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

@@ -2003,14 +2003,20 @@ def _on_generation_created(self, ev: llm.GenerationCreatedEvent) -> None:
         self._schedule_speech(handle, SpeechHandle.SPEECH_PRIORITY_NORMAL)
 
     def _interrupt_by_audio_activity(
-        self, *, ignore_user_transcript_until: float | None = None
+        self,
+        *,
+        ignore_user_transcript_until: float | None = None,
+        enforce_min_duration: bool = True,
     ) -> None:
         """
         Interrupt the current speech or generation, and optionally ignore the user transcript until the given timestamp.
 
         Args:
             ignore_user_transcript_until: The timestamp until which the user transcript should be ignored.
                 If None, the user transcript will be ignored until the current time.
+            enforce_min_duration: Whether to enforce ``interruption.min_duration`` against the
+                tracked user speech duration. Set to False by callers that already made an
+                explicit interruption decision (e.g. the interruption detection model).
         """
         if not self._interruption_by_audio_activity_enabled:
             return
@@ -2035,6 +2041,20 @@ def _interrupt_by_audio_activity(
             if len(split_words(text, split_character=True)) < interruption_options["min_words"]:
                 return
 
+        # enforce min_duration on every audio-activity trigger, not only the VAD path;
+        # STT interim/final transcripts would otherwise interrupt regardless of how
+        # short the user speech was (https://github.com/livekit/agents/issues/3515).
+        # an unknown duration (None) is allowed through to keep the VAD-failure
+        # failsafe in on_final_transcript working.
+        if (
+            enforce_min_duration
+            and interruption_options["min_duration"] > 0
+            and self._audio_recognition is not None
+            and (speech_duration := self._audio_recognition.current_speech_duration) is not None
+            and speech_duration < interruption_options["min_duration"]
+        ):
+            return
+
         if self._rt_session is not None:
             self._rt_session.start_user_activity()
 
@@ -2183,10 +2203,13 @@ def on_backchannel_confirmed(self) -> None:
             self._rt_session.clear_audio()
 
     def on_interruption(self, ev: inference.OverlappingSpeechEvent) -> None:
-        # restore interruption by audio activity and then immediately interrupt
+        # restore interruption by audio activity and then immediately interrupt.
+        # the interruption detection model already decided this is a real interruption,
+        # so min_duration is not re-checked here.
         self._restore_interruption_by_audio_activity()
         self._interrupt_by_audio_activity(
-            ignore_user_transcript_until=ev.overlap_started_at or ev.detected_at
+            ignore_user_transcript_until=ev.overlap_started_at or ev.detected_at,
+            enforce_min_duration=False,
         )
         # flush held transcripts again if possible
         if self._audio_recognition:

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

@@ -273,6 +273,10 @@ def __init__(
         self._last_final_transcript_time: float | None = None
         self._last_speaking_time: float | None = None
         self._speech_start_time: float | None = None
+        # VAD-measured voiced duration for the current/most recent segment; preferred
+        # by ``current_speech_duration`` so STT-triggered min_duration matches the
+        # VAD path (wall-clock includes trailing silence / STT latency).
+        self._vad_speech_duration: float | None = None
 
         # used for manual commit_user_turn
         self._final_transcript_received = asyncio.Event()
@@ -623,6 +627,30 @@ def _on_end_of_overlap_speech(
             _OverlapSpeechEndedSentinel(ended_at=ended_at or time.time(), agent_ended=agent_ended)
         )
 
+    @property
+    def current_speech_duration(self) -> float | None:
+        """Voiced duration (s) of the current or most recent user speech segment.
+
+        Prefers the VAD-measured ``speech_duration`` (same metric
+        ``on_vad_inference_done`` uses for ``interruption.min_duration``). Falls
+        back to wall-clock elapsed when VAD has not reported a duration for this
+        segment (e.g. STT-only turn detection). Returns None when no speech start
+        has been tracked.
+        """
+        if self._vad_speech_duration is not None:
+            return self._vad_speech_duration
+
+        if self._speech_start_time is None:
+            return None
+
+        if self._speaking:
+            return max(time.time() - self._speech_start_time, 0.0)
+
+        if self._last_speaking_time is None or self._last_speaking_time < self._speech_start_time:
+            return None
+
+        return self._last_speaking_time - self._speech_start_time
+
     @property
     def _speaking(self) -> bool:
         return not self._user_silence_ev.is_set()
@@ -1035,6 +1063,7 @@ def _clear_user_turn(self) -> None:
         self._last_final_transcript_time = None
         self._speech_start_time = None
         self._last_speaking_time = None
+        self._vad_speech_duration = None
         self._vad_speech_started = False
         self._user_turn_committed = False
         self._last_emitted_prediction = None
@@ -1396,6 +1425,7 @@ async def _on_vad_event(self, ev: vad.VADEvent) -> None:
                 self._speech_start_time = speech_start_time
                 self._vad_speech_started = True
 
+            self._vad_speech_duration = ev.speech_duration
             self._cancel_transcription_timeout()
 
             with trace.use_span(self._ensure_user_turn_span(start_time=speech_start_time)):
@@ -1415,6 +1445,17 @@ async def _on_vad_event(self, ev: vad.VADEvent) -> None:
                 self._session.amd._on_user_speech_started()
 
         elif ev.type == vad.VADEventType.INFERENCE_DONE:
+            # Store duration before the hook: on_vad_inference_done →
+            # _interrupt_by_audio_activity reads current_speech_duration, which
+            # must match ev.speech_duration on the same frame or barge-in is
+            # delayed by one VAD window (~32ms).
+            # Only positive VAD durations. Silero zeros pub_speech_duration after
+            # EOS; writing that 0 makes a late STT final look "too short".
+            # `_speaking` is not a substitute: turn_detection="stt" sets it from
+            # STT START_OF_SPEECH while Silero still reports 0, which would
+            # store 0.0 instead of None and disable the VAD-miss failsafe.
+            if ev.speech_duration > 0.0:
+                self._vad_speech_duration = ev.speech_duration
             self._hooks.on_vad_inference_done(ev)
 
             # for metrics, get the "earliest" signal of speech as possible
@@ -1444,6 +1485,9 @@ async def _on_vad_event(self, ev: vad.VADEvent) -> None:
             self._speaking = False
             speech_end_time = time.time() - ev.silence_duration - ev.inference_duration
             self._last_speaking_time = speech_end_time
+            # keep the final voiced duration so a late STT final still sees the
+            # same metric the VAD path used for min_duration
+            self._vad_speech_duration = ev.speech_duration
 
             # A committed turn clears _vad_speech_started before its late VAD EOS arrives.
             if self._stt_pipeline is not None and vad_speech_started:
@@ -1771,6 +1815,10 @@ async def _bounce_eou_task(
                     self._speech_start_time = None
                     self._vad_speech_started = False
                     self._last_speaking_time = None
+                    # drop the prior segment's voiced duration so a late STT
+                    # failsafe (no new VAD events) sees unknown duration, not
+                    # a stale min_duration gate from the previous turn
+                    self._vad_speech_duration = None
 
                 if self._turn_detector_stream is not None:
                     self._turn_detector_stream.flush(reason="turn committed")

tests/fake_session.py

@@ -141,6 +141,7 @@ def add_user_speech(
         *,
         stt_delay: float = 0.2,
         final: bool = True,
+        interim_interval: float | None = None,
     ) -> None:
         self._items.append(
             FakeUserSpeech(
@@ -149,6 +150,7 @@ def add_user_speech(
                 transcript=transcript,
                 stt_delay=stt_delay,
                 final=final,
+                interim_interval=interim_interval,
             )
         )
 

tests/fake_stt.py

@@ -33,12 +33,17 @@ class FakeUserSpeech(BaseModel):
     transcript: str  # empty string fires VAD SOS/EOS only — no STT events
     stt_delay: float
     final: bool = True
+    interim_interval: float | None = None
+    """If set, stream growing interim transcripts every ``interim_interval`` seconds
+    while the user is speaking (like providers that emit continuous partial results)."""
 
     def speed_up(self, factor: float) -> FakeUserSpeech:
         obj = copy.deepcopy(self)
         obj.start_time /= factor
         obj.end_time /= factor
         obj.stt_delay /= factor
+        if obj.interim_interval is not None:
+            obj.interim_interval /= factor
         return obj
 
 
@@ -222,6 +227,20 @@ def curr_time() -> float:
                 if curr_time() < final_transcript_time:
                     await asyncio.sleep(final_transcript_time - curr_time())
                 continue
+
+            if fake_speech.interim_interval is not None:
+                # stream growing interim transcripts while the user is speaking
+                words = fake_speech.transcript.split()
+                num_sent = 0
+                next_interim_time = fake_speech.start_time + fake_speech.interim_interval
+                while next_interim_time < fake_speech.end_time:
+                    if curr_time() < next_interim_time:
+                        await asyncio.sleep(next_interim_time - curr_time())
+                    num_sent += 1
+                    prefix = " ".join(words[: min(num_sent, len(words))])
+                    self.send_fake_transcript(prefix, is_final=False)
+                    next_interim_time += fake_speech.interim_interval
+
             interim_transcript_time = fake_speech.end_time + fake_speech.stt_delay * 0.5
             if curr_time() < interim_transcript_time:
                 await asyncio.sleep(interim_transcript_time - curr_time())

tests/test_agent_session.py

@@ -497,6 +497,51 @@ async def test_interruption_options() -> None:
     check_timestamp(playback_finished_events[0].playback_position, 5.0, speed_factor=speed)
 
 
+async def test_min_interruption_duration_applies_to_stt_transcripts() -> None:
+    """Regression test for https://github.com/livekit/agents/issues/3515.
+
+    STTs that stream continuous interim results (e.g. Amazon Transcribe) used to
+    trigger an interruption on the first non-empty interim transcript, bypassing
+    ``interruption.min_duration`` entirely (it was only enforced on the VAD path).
+    The interruption must not fire before the user has spoken for min_duration.
+    """
+    speed = 1
+    actions = FakeActions()
+    actions.add_user_speech(0.5, 2.5, "Tell me a story.")
+    actions.add_llm("Here is a long story for you ... the end.")
+    actions.add_tts(10.0)  # playout starts at 3.5s
+    # user speaks for 4s while interim transcripts stream every 0.3s;
+    # the first interim lands at ~5.3s, min_duration is reached at 7.0s
+    actions.add_user_speech(
+        5.0, 9.0, "please stop talking right now", stt_delay=0.2, interim_interval=0.3
+    )
+
+    session = create_session(
+        actions,
+        speed_factor=speed,
+        turn_handling={"interruption": {"min_duration": 2.0}},
+    )
+    agent_state_events: list[AgentStateChangedEvent] = []
+    playback_finished_events: list[PlaybackFinishedEvent] = []
+    session.on("agent_state_changed", agent_state_events.append)
+    session.output.audio.on("playback_finished", playback_finished_events.append)
+
+    t_origin = await asyncio.wait_for(run_session(session, MyAgent()), timeout=SESSION_TIMEOUT)
+
+    assert len(playback_finished_events) == 1
+    assert playback_finished_events[0].interrupted is True
+    # interrupted at 7.0s (5.0 + min_duration), i.e. 3.5s into the playout —
+    # not at ~5.3s when the first interim transcript arrived
+    check_timestamp(playback_finished_events[0].playback_position, 3.5, speed_factor=speed)
+
+    interrupted_at = next(
+        ev.created_at
+        for ev in agent_state_events
+        if ev.new_state == "listening" and ev.old_state == "speaking"
+    )
+    check_timestamp(interrupted_at - t_origin, 7.0, speed_factor=speed)
+
+
 async def test_interruption_by_text_input() -> None:
     speed = 1
     actions = FakeActions()
@@ -1535,6 +1580,9 @@ async def test_vad_fallback_uses_next_vad_inference_event(
     )
 
     audio_recognition = MagicMock()
+    # unknown speech duration: the min_duration gate lets it through (the VAD
+    # event below carries its own speech_duration check)
+    audio_recognition.current_speech_duration = None
     current_speech = MagicMock()
     current_speech.interrupted = False
     current_speech.allow_interruptions = True