fix(voice): drop the recognition turn a false interruption resumed over

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

@@ -4531,12 +4531,25 @@ def _on_false_interruption() -> None:
                 and audio_output.can_pause
                 and not self._paused_speech.handle.done()
             ):
+                if (recognition := self._audio_recognition) is not None:
+                    # the interruption was false, so the turn that raised it is abandoned: let it
+                    # go before the agent's next speech interval opens, or the next real utterance
+                    # inherits its speech-start anchor. How much can go depends on whether the
+                    # turn was decided, which only the decision itself knows: this can be reached
+                    # with no decision ever made (turn_detection="stt" starts no bounce on VAD
+                    # END_OF_SPEECH), and then a slow stt final for real speech may still be on
+                    # its way.
+                    if recognition._user_turn_dropped:
+                        recognition._clear_user_turn()
+                    else:
+                        recognition._release_user_turn_anchors()
+
                 self._session._update_agent_state(
                     self._paused_speech.agent_state,
                     otel_context=self._paused_speech.handle._agent_turn_context,
                 )
-                if self._audio_recognition and self._paused_speech.agent_state == "speaking":
-                    self._audio_recognition._on_start_of_agent_speech(started_at=time.time())
+                if recognition is not None and self._paused_speech.agent_state == "speaking":
+                    recognition._on_start_of_agent_speech(started_at=time.time())
                 if self.interruption_enabled:
                     self._disable_vad_interruption_soon()
                 audio_output.resume()

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

@@ -283,6 +283,9 @@ def __init__(
         # used for manual commit_user_turn
         self._final_transcript_received = asyncio.Event()
         self._final_transcript_confidence: list[float] = []
+        # set by the end-of-turn decision; read by the false-interruption resume, which has to
+        # know whether this turn was decided and dropped or is simply still open
+        self._user_turn_dropped = False
         self._audio_transcript = ""
         self._audio_interim_transcript = ""
         # used for STTs that support preflight mode, so it could start preemptive generation earlier
@@ -977,6 +980,34 @@ def _detach_turn_detector(self) -> _StreamingTurnDetectorStream | None:
         self._turn_detector_prediction_fut = None
         return stream
 
+    def _open_user_turn(self, speech_start_time: float) -> None:
+        """Anchor a new logical user turn, superseding the previous turn's verdict.
+
+        Every path that takes a fresh ``_speech_start_time`` goes through here.
+        ``_user_turn_dropped`` describes the turn that just ended, and reading it against
+        this one is how a real transcript ends up discarded.
+        """
+        self._speech_start_time = speech_start_time
+        self._user_turn_dropped = False
+
+    def _release_user_turn_anchors(self) -> None:
+        """Release an open turn's anchors without discarding what could still arrive.
+
+        For a turn abandoned before any end-of-turn decision ran: its speech-start anchor
+        must not be inherited by the next utterance, but an stt final for this speech may
+        still be in flight. Unlike `_clear_user_turn`, the transcript, the accumulated
+        confidence, the turn detector's buffer and the stt pipeline are left alone, so a
+        late final can still commit the turn.
+        """
+        self._speech_start_time = None
+        self._vad_speech_started = False
+        self._last_emitted_prediction = None
+        self._user_turn_dropped = False
+        self._turn_tracker = _UserTurnTracker()
+
+        # end any in-progress user_turn span so the next speech starts a fresh one
+        self._end_user_turn_span()
+
     def _clear_user_turn(self) -> None:
         self._audio_transcript = ""
         self._audio_interim_transcript = ""
@@ -987,6 +1018,7 @@ def _clear_user_turn(self) -> None:
         self._last_speaking_time = None
         self._vad_speech_started = False
         self._user_turn_committed = False
+        self._user_turn_dropped = False
         self._last_emitted_prediction = None
         if self._turn_detector_stream is not None:
             self._turn_detector_stream.flush(reason="clear_user_turn")
@@ -1340,11 +1372,12 @@ def _process_stt_event(self, ev: stt.SpeechEvent) -> None:
         elif ev.type == stt.SpeechEventType.START_OF_SPEECH and self._turn_detection_mode == "stt":
             # If the plugin provided a server onset timestamp, use it;
             # otherwise fall back to message arrival time.
-            if self._speech_start_time is None:
-                self._speech_start_time = ev.speech_start_time or time.time()
+            if (speech_start_time := self._speech_start_time) is None:
+                speech_start_time = ev.speech_start_time or time.time()
+                self._open_user_turn(speech_start_time)
 
-            with tracer.use_span(self._ensure_user_turn_span(start_time=self._speech_start_time)):
-                self._hooks.on_start_of_speech(None, speech_start_time=self._speech_start_time)
+            with tracer.use_span(self._ensure_user_turn_span(start_time=speech_start_time)):
+                self._hooks.on_start_of_speech(None, speech_start_time=speech_start_time)
 
             self._speaking = True
             self._last_speaking_time = stt_last_speaking_time
@@ -1358,7 +1391,7 @@ async def _on_vad_event(self, ev: vad.VADEvent) -> None:
             speech_start_time = time.time() - ev.speech_duration - ev.inference_duration
             self._active_vad_speech_started_at = speech_start_time
             if not self._vad_speech_started:
-                self._speech_start_time = speech_start_time
+                self._open_user_turn(speech_start_time)
                 self._vad_speech_started = True
 
             self._cancel_transcription_timeout()
@@ -1387,7 +1420,7 @@ async def _on_vad_event(self, ev: vad.VADEvent) -> None:
                 self._last_speaking_time = time.time()
 
                 if self._speech_start_time is None:
-                    self._speech_start_time = time.time() - ev.raw_accumulated_speech
+                    self._open_user_turn(time.time() - ev.raw_accumulated_speech)
                 if self._speaking and self._turn_detector_prediction_fut is not None:
                     if self._turn_detector_stream is not None:
                         self._turn_detector_stream.cancel_inference()
@@ -1751,6 +1784,9 @@ async def _bounce_eou_task(
                     self._turn_detector_prediction_fut = None
                     self._turn_detector_flushed = True
 
+            # a dropped turn keeps accumulating, so nothing else records the verdict
+            self._user_turn_dropped = not committed
+
             # reset turn-scoped barge-in state once per logical turn (commit or drop)
             self._turn_backchannel_over_agent = False
             self._overlap_in_current_turn = False

tests/test_false_interruption_resume.py

@@ -14,7 +14,7 @@
 
 import pytest
 
-from livekit.agents import Agent, AgentSession, TurnHandlingOptions
+from livekit.agents import Agent, AgentSession, TurnHandlingOptions, stt, vad
 from livekit.agents.inference import OverlappingSpeechEvent
 from livekit.agents.voice.agent_activity import AgentActivity, _PausedSpeechInfo
 from livekit.agents.voice.audio_recognition import (
@@ -46,6 +46,7 @@ def _recognition(hooks: AgentActivity, last_speaking_time: float) -> AudioRecogn
     ar._session = MagicMock()
     ar._hooks = hooks
     ar._stt = None  # realtime model, no STT
+    ar._stt_aligned_transcript = False
     ar._audio_transcript = ""
     ar._turn_detection_mode = None
     ar._turn_detector = MagicMock(spec=_StreamingTurnDetector)
@@ -102,6 +103,7 @@ def _recognition(hooks: AgentActivity, last_speaking_time: float) -> AudioRecogn
     ar._vad_speech_started = False
     ar._end_of_turn_task = None
     ar._user_turn_committed = False
+    ar._user_turn_dropped = False
     ar._vad = None
     ar._last_language = None
     ar._last_emitted_prediction = None
@@ -385,3 +387,159 @@ async def test_resume_is_immediate_when_no_turn_decision_is_open(
 
     assert [name for name, _ in events] == ["resume"]
     assert events[0][1] - t0 == pytest.approx(FALSE_INTERRUPTION_TIMEOUT, abs=0.1)
+
+
+async def test_resume_drops_the_turn_it_resumed_over(monkeypatch: pytest.MonkeyPatch) -> None:
+    # a confirmed false interruption abandons the recognition turn that caused it; leaving that
+    # turn open lets the next real utterance inherit its anchor, because _on_vad_event only
+    # takes a new _speech_start_time while _vad_speech_started is still False
+    monkeypatch.setenv("LIVEKIT_API_KEY", "k")
+    monkeypatch.setenv("LIVEKIT_API_SECRET", "s")
+
+    session = _session()
+    activity, _ = _paused_activity(session)
+
+    stale_start = time.time() - 12.0
+
+    activity.on_end_of_speech(None)
+    recognition = _recognition(activity, last_speaking_time=time.time() - VAD_MIN_SILENCE)
+    activity._audio_recognition = recognition
+    # the VAD turn behind the interruption: opened, never transcribed
+    recognition._speech_start_time = stale_start
+    recognition._vad_speech_started = True
+    # a confirmed backchannel is what makes the turn drop rather than commit
+    recognition._turn_backchannel_over_agent = True
+    recognition._run_eou_detection(MagicMock(), trigger="vad")
+
+    await asyncio.sleep(MAX_DELAY + 0.3)
+    assert activity._paused_speech is None  # the speech resumed
+
+    # the later, real utterance
+    onset = time.time()
+    await recognition._on_vad_event(
+        vad.VADEvent(
+            type=vad.VADEventType.START_OF_SPEECH,
+            samples_index=0,
+            timestamp=onset,
+            speech_duration=0.1,
+            silence_duration=0.0,
+        )
+    )
+    await session.aclose()
+
+    assert recognition._speech_start_time == pytest.approx(onset - 0.1, abs=0.1)
+
+
+async def test_resume_without_a_turn_decision_keeps_a_late_transcript_alive(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # turn_detection="stt" starts no eou bounce on VAD END_OF_SPEECH, so the resume timer fires
+    # with no decision open. The speech may have been real with a slow stt final still on its
+    # way: letting the turn go must release its anchor without taking the pipeline — and the
+    # transcript it can still commit — with it
+    monkeypatch.setenv("LIVEKIT_API_KEY", "k")
+    monkeypatch.setenv("LIVEKIT_API_SECRET", "s")
+
+    session = _session()
+    activity, _ = _paused_activity(session)
+
+    recognition = _recognition(activity, last_speaking_time=time.time() - VAD_MIN_SILENCE)
+    activity._audio_recognition = recognition
+    recognition._speech_start_time = time.time() - 12.0
+    recognition._vad_speech_started = True
+    recognition._audio_transcript = "what the caller actually said"
+    pipeline = MagicMock()
+    pipeline.aclose = AsyncMock()
+    recognition._stt_pipeline = pipeline
+
+    # no _run_eou_detection: on an stt pipeline the bounce waits for the stt final
+    activity.on_end_of_speech(None)
+    assert recognition._end_of_turn_task is None
+
+    await asyncio.sleep(FALSE_INTERRUPTION_TIMEOUT + 0.2)
+    await session.aclose()
+
+    assert activity._paused_speech is None  # the speech resumed
+    assert recognition._speech_start_time is None  # the anchor is released
+    # everything a late final needs to commit the turn survived
+    assert recognition._audio_transcript == "what the caller actually said"
+    assert recognition._stt_pipeline is pipeline
+    pipeline.aclose.assert_not_called()
+
+
+async def test_a_backchannel_dropped_before_the_timeout_does_not_leak_forward(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # the shipped false_interruption_timeout is 2s and a confirmed backchannel drops in a few
+    # hundred ms, so the decision normally lands well before the resume timer. The verdict has
+    # to survive that gap: a dropped turn keeps its transcript, which would otherwise be
+    # prepended to the next real utterance (_audio_transcript accumulates)
+    monkeypatch.setenv("LIVEKIT_API_KEY", "k")
+    monkeypatch.setenv("LIVEKIT_API_SECRET", "s")
+
+    session = _session()
+    activity, handle = _paused_activity(session)
+    # a resume timer that outlives the turn decision, unlike the module default
+    late_timeout = MAX_DELAY + 0.3
+    activity._paused_speech = _PausedSpeechInfo(
+        handle=handle, agent_state="speaking", timeout=late_timeout
+    )
+
+    activity.on_end_of_speech(None)
+    recognition = _recognition(activity, last_speaking_time=time.time() - VAD_MIN_SILENCE)
+    activity._audio_recognition = recognition
+    recognition._speech_start_time = time.time() - 12.0
+    recognition._vad_speech_started = True
+    recognition._audio_transcript = "mm-hmm"
+    recognition._turn_backchannel_over_agent = True
+    recognition._run_eou_detection(MagicMock(), trigger="vad")
+
+    # the turn is dropped here, while the resume timer is still running
+    await asyncio.sleep(MAX_DELAY + 0.1)
+    assert recognition._user_turn_dropped is True
+
+    await asyncio.sleep(late_timeout - MAX_DELAY + 0.2)
+    await session.aclose()
+
+    assert activity._paused_speech is None  # the speech resumed
+    assert recognition._speech_start_time is None
+    assert recognition._audio_transcript == ""  # nothing left to prepend
+
+
+async def test_an_stt_anchored_turn_supersedes_the_previous_verdict(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # the resume timer is also armed from the stt hooks, so a session can anchor its next turn
+    # through _on_stt_event rather than a VAD start. The dropped verdict belongs to the turn
+    # that ended: read against the new one it would erase a real transcript
+    monkeypatch.setenv("LIVEKIT_API_KEY", "k")
+    monkeypatch.setenv("LIVEKIT_API_SECRET", "s")
+
+    session = _session()
+    activity, _ = _paused_activity(session)
+
+    recognition = _recognition(activity, last_speaking_time=time.time() - VAD_MIN_SILENCE)
+    activity._audio_recognition = recognition
+    recognition._turn_detection_mode = "stt"
+    # the previous turn was decided and dropped
+    recognition._user_turn_dropped = True
+
+    # the next turn takes its anchor from the stt stream, never from a VAD start
+    onset = time.time()
+    await recognition._on_stt_event(
+        stt.SpeechEvent(
+            type=stt.SpeechEventType.START_OF_SPEECH,
+            alternatives=[stt.SpeechData(language="en", text="", start_time=onset)],
+        )
+    )
+    assert recognition._speech_start_time == pytest.approx(onset, abs=1.0)
+    assert recognition._user_turn_dropped is False
+
+    # a slow final for this new turn is still on its way when the resume fires
+    recognition._audio_transcript = "what the caller actually said"
+    activity.on_end_of_speech(None)
+    await asyncio.sleep(FALSE_INTERRUPTION_TIMEOUT + 0.2)
+    await session.aclose()
+
+    assert activity._paused_speech is None
+    assert recognition._audio_transcript == "what the caller actually said"