fix(voice): discard the uncommitted recognition turn on false-interruption resume

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

@@ -4528,6 +4528,28 @@ def _on_false_interruption() -> None:
                 self._paused_speech = None
                 return
 
+            # the interruption was judged false, so the uncommitted recognition
+            # turn it opened is dead. Discard it before agent speech resumes:
+            # the open user_turn span and its _user_turn_start anchor otherwise
+            # survive, and the next real utterance reuses them
+            # (_ensure_user_turn_span returns a recording span as-is), committing
+            # with a started_speaking_at that predates the resumed agent speech.
+            # Skip while the user is speaking right now: those anchors belong
+            # to a live utterance, not to the discarded turn. A live VAD segment
+            # counts as speaking even when _speaking is already False, because a
+            # premature STT END_OF_SPEECH clears _speaking while VAD is still
+            # mid-segment. That same EOS flushes the VAD without an
+            # END_OF_SPEECH, so a flush-abandoned segment (_vad_speech_flushed,
+            # no fresh SOS since) is dead, not live: without that distinction a
+            # correct STT ending would suppress the cleanup forever.
+            # reset_stt=False keeps the live STT stream so audio still being
+            # decoded can deliver a late final and interrupt the resumed speech.
+            if (recognition := self._audio_recognition) is not None and not (
+                recognition._speaking
+                or (recognition._vad_speech_started and not recognition._vad_speech_flushed)
+            ):
+                recognition._clear_user_turn(reset_stt=False)
+
             resumed = False
             if (
                 self._session.options.interruption["resume_false_interruption"]

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

@@ -335,6 +335,10 @@ def __init__(
         self.__stt_context: BaseModel | None = None
 
         self._vad_speech_started: bool = False
+        # a premature STT END_OF_SPEECH flushes the VAD stream mid-segment; the flush
+        # emits no END_OF_SPEECH, so _vad_speech_started alone cannot say whether the
+        # segment is still live. Set at the flush, cleared by the next VAD SOS.
+        self._vad_speech_flushed: bool = False
 
         self._transcription_timeout_handle: asyncio.TimerHandle | None = None
         self._turn_speech_duration: float = 0.0
@@ -977,7 +981,7 @@ def _detach_turn_detector(self) -> _StreamingTurnDetectorStream | None:
         self._turn_detector_prediction_fut = None
         return stream
 
-    def _clear_user_turn(self) -> None:
+    def _clear_user_turn(self, *, reset_stt: bool = True) -> None:
         self._audio_transcript = ""
         self._audio_interim_transcript = ""
         self._audio_preflight_transcript = ""
@@ -986,6 +990,7 @@ def _clear_user_turn(self) -> None:
         self._speech_start_time = None
         self._last_speaking_time = None
         self._vad_speech_started = False
+        self._vad_speech_flushed = False
         self._user_turn_committed = False
         self._last_emitted_prediction = None
         if self._turn_detector_stream is not None:
@@ -1000,6 +1005,13 @@ def _clear_user_turn(self) -> None:
         self._stt_request_ids = []
         self._reset_transcription_timeout()
 
+        if not reset_stt:
+            # keep the live STT stream: the false-interruption resume discards
+            # the turn bookkeeping, but audio the provider is still decoding
+            # must be able to deliver a late final (a real barge-in that VAD
+            # dropped still interrupts the resumed speech)
+            return
+
         # reset stt to clear the buffer from previous user turn
         stt = self._stt
         self._update_stt(None)
@@ -1308,6 +1320,12 @@ def _process_stt_event(self, ev: stt.SpeechEvent) -> None:
                     else:
                         self._update_vad(self._vad)
 
+                    # the flush ends the segment without an END_OF_SPEECH, so
+                    # _vad_speech_started stays True with nothing left to clear
+                    # it; mark the segment as abandoned until a fresh SOS proves
+                    # the user is actually still speaking
+                    self._vad_speech_flushed = True
+
                     logger.warning(
                         "stt end of speech received while vad is still in a speech segment, "
                         "flushing vad",
@@ -1357,6 +1375,8 @@ async def _on_vad_event(self, ev: vad.VADEvent) -> None:
         if ev.type == vad.VADEventType.START_OF_SPEECH:
             speech_start_time = time.time() - ev.speech_duration - ev.inference_duration
             self._active_vad_speech_started_at = speech_start_time
+            # a fresh SOS supersedes any flush-abandoned segment: the user is speaking
+            self._vad_speech_flushed = False
             if not self._vad_speech_started:
                 self._speech_start_time = speech_start_time
                 self._vad_speech_started = True

tests/test_false_interruption_resume.py

@@ -100,6 +100,7 @@ def _recognition(hooks: AgentActivity, last_speaking_time: float) -> AudioRecogn
     ar._last_final_transcript_time = None
     ar._speech_start_time = None
     ar._vad_speech_started = False
+    ar._vad_speech_flushed = False
     ar._end_of_turn_task = None
     ar._user_turn_committed = False
     ar._vad = None
@@ -385,3 +386,140 @@ 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_discards_the_uncommitted_recognition_turn(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # 7063: the false turn's open user_turn span and _user_turn_start anchor must
+    # not survive the resume. _ensure_user_turn_span returns a recording span
+    # as-is, so the next real utterance would inherit the abandoned turn's start
+    # and commit with a started_speaking_at seconds before it was spoken.
+    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())
+    activity._audio_recognition = recognition
+
+    # the false turn left an open span and a stale start anchor behind
+    stale_start = time.time() - 12.0
+    recognition._user_turn_start = stale_start
+    stale_span = MagicMock()
+    stale_span.is_recording.return_value = True
+    recognition._user_turn_span = stale_span
+    stt_pipeline = MagicMock()
+    stt_pipeline.aclose = AsyncMock()
+    recognition._stt_pipeline = stt_pipeline
+
+    order: list[str] = []
+    original_clear = recognition._clear_user_turn
+
+    def _record_clear(**kwargs: bool) -> None:
+        order.append("clear")
+        original_clear(**kwargs)
+
+    recognition._clear_user_turn = _record_clear  # type: ignore[method-assign]
+    audio_output = session.output.audio
+    assert audio_output is not None
+    original_resume = audio_output.resume
+
+    def _record_resume() -> None:
+        order.append("resume")
+        original_resume()
+
+    monkeypatch.setattr(audio_output, "resume", _record_resume)
+
+    events: list[bool] = []
+    session.on("agent_false_interruption", lambda ev: events.append(ev.resumed))
+
+    activity._start_false_interruption_timer(0.01)
+    await asyncio.sleep(0.1)
+
+    assert events == [True]
+    # the dead turn is discarded before agent audio resumes
+    assert order == ["clear", "resume"]
+    # the span is closed and the anchor dropped, so the next utterance starts fresh
+    stale_span.end.assert_called_once()
+    assert recognition._user_turn_start is None
+    # the live STT stream is kept: a late final for a real barge-in must still arrive
+    assert recognition._stt_pipeline is stt_pipeline
+    stt_pipeline.aclose.assert_not_called()
+
+    recognition._stt_pipeline = None  # teardown owns the real pipeline lifecycle
+    await session.aclose()
+
+
+async def test_resume_keeps_the_anchors_of_live_user_speech(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # if the user is speaking when the timer fires (e.g. VAD activity that the
+    # min_duration gate kept from resetting the timer), the recognition anchors
+    # belong to a live utterance and must survive the resume
+    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())
+    recognition._speaking = True
+    activity._audio_recognition = recognition
+    recognition._clear_user_turn = MagicMock()  # type: ignore[method-assign]
+
+    activity._start_false_interruption_timer(0.01)
+    await asyncio.sleep(0.1)
+    await session.aclose()
+
+    recognition._clear_user_turn.assert_not_called()
+
+
+async def test_resume_keeps_the_anchors_of_a_live_vad_segment(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # stt turn detection: a premature STT END_OF_SPEECH sets _speaking False while
+    # VAD is still mid-segment (the flushed VAD corrects it with a new SOS). The
+    # discard must treat that live segment as speech, or it wipes the transcript
+    # and timing of an utterance that is still going.
+    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())
+    recognition._speaking = False
+    recognition._vad_speech_started = True
+    activity._audio_recognition = recognition
+    recognition._clear_user_turn = MagicMock()  # type: ignore[method-assign]
+
+    activity._start_false_interruption_timer(0.01)
+    await asyncio.sleep(0.1)
+    await session.aclose()
+
+    recognition._clear_user_turn.assert_not_called()
+
+
+async def test_resume_discards_a_flush_abandoned_vad_segment(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # the premature-STT-EOS flush emits no VAD END_OF_SPEECH, so
+    # _vad_speech_started stays True with nothing left to clear it. When STT was
+    # right and no speech follows, that stuck flag must not suppress the
+    # discard, or the stale-anchor bug returns on this path.
+    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())
+    recognition._speaking = False
+    recognition._vad_speech_started = True
+    recognition._vad_speech_flushed = True  # STT END_OF_SPEECH flushed the segment
+    activity._audio_recognition = recognition
+    recognition._clear_user_turn = MagicMock()  # type: ignore[method-assign]
+
+    activity._start_false_interruption_timer(0.01)
+    await asyncio.sleep(0.1)
+    await session.aclose()
+
+    recognition._clear_user_turn.assert_called_once_with(reset_stt=False)