fix(voice): defer false-interruption resume while a transcript is pending

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

@@ -98,6 +98,11 @@
     _StreamingTurnDetectorStream,
 )
 
+# false-interruption resume: re-check cadence and cap while a speech window's
+# transcript is still in flight (see _start_false_interruption_timer)
+_PENDING_TRANSCRIPT_RECHECK = 0.1
+_PENDING_TRANSCRIPT_MAX_DEFERRAL = 2.0
+
 if TYPE_CHECKING:
     from ..llm import mcp
     from .agent_session import AgentSession, ExpressiveOptions
@@ -4287,6 +4292,10 @@ def _cancel_false_interruption_timer(self) -> None:
     def _start_false_interruption_timer(self, timeout: float) -> None:
         self._cancel_false_interruption_timer()
 
+        # set when resume is first deferred on a pending transcript; a dead STT
+        # stream with interims never finalizing degrades to plain timeout behavior
+        transcript_wait_deadline: float | None = None
+
         def _on_false_interruption() -> None:
             if self._paused_speech is None or (
                 self._current_speech and self._current_speech is not self._paused_speech.handle
@@ -4341,6 +4350,7 @@ def _on_turn_settled(settled: asyncio.Task[None]) -> None:
             _on_false_interruption()
 
         def _on_timeout() -> None:
+            nonlocal transcript_wait_deadline
             self._false_interruption_timer = None
 
             # an open turn decision owns the paused speech: it either commits and interrupts it
@@ -4353,6 +4363,22 @@ def _on_timeout() -> None:
                 eot_task.add_done_callback(_on_turn_settled)
                 return
 
+            # a speech window closed but its transcript is still in flight: the
+            # coming final will start (or refresh) the turn decision that owns
+            # this pause — re-check instead of resuming stale audio into the gap
+            if self._audio_recognition and getattr(
+                self._audio_recognition, "_audio_interim_transcript", ""
+            ):
+                if transcript_wait_deadline is None:
+                    transcript_wait_deadline = time.monotonic() + _PENDING_TRANSCRIPT_MAX_DEFERRAL
+                if time.monotonic() < transcript_wait_deadline:
+                    self._false_interruption_timer = self._session._loop.call_later(
+                        _PENDING_TRANSCRIPT_RECHECK, _on_timeout
+                    )
+                    return
+                # the transcript never materialized (e.g. the STT stream died):
+                # fall through and let the timeout rule, as before this guard
+
             _on_false_interruption()
 
         self._false_interruption_timer = self._session._loop.call_later(timeout, _on_timeout)

tests/test_stale_resume_stt_windows.py

@@ -0,0 +1,148 @@
+"""A pause backed by live speech evidence must not resume on a bare END_OF_SPEECH.
+
+With STT-based turn detection (Cartesia Ink-2 / Deepgram Flux), START_OF_SPEECH is
+speech-selective and a turn's FINAL_TRANSCRIPT can lag its speech windows: a caller
+telling a long story produces SOS/EOS pairs (breaths) with word-bearing interims,
+and the final that commits the turn arrives seconds later.
+
+Today a bare EOS arms the false-interruption resume timer with the pause's own
+timeout — which is 0 for a pre-playout pause taken by ``on_start_of_speech`` —
+and ``_on_timeout`` only defers on an *open* ``_end_of_turn_task``. Between a
+speech window's EOS and its (still pending) final there is no open decision, so
+the stale pause resumes straight into the caller's breath, emits a word or two,
+and is then killed by the commit. Observed in production twice in one call:
+held replies leaking "I'm" … "I" into each pause of a caller's narration.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import time
+from unittest.mock import MagicMock
+
+import pytest
+
+from livekit.agents import Agent
+from livekit.agents.voice.agent_activity import AgentActivity, _PausedSpeechInfo
+
+from .fake_io import FakeAudioOutput
+from .test_false_interruption_resume import _session
+
+pytestmark = pytest.mark.unit
+
+FALSE_INTERRUPTION_TIMEOUT = 0.3
+
+
+def _activity_with_pending_reply(session, *, interim_words: str) -> tuple[AgentActivity, MagicMock]:
+    """An activity whose reply is current but unplayed, with the caller's words
+    (interims) already in flight — the state right before a pre-playout pause."""
+    activity = AgentActivity(Agent(instructions="test"), session)
+    activity._scheduling_paused = False
+    session.output.audio = FakeAudioOutput(can_pause=True)
+
+    handle = MagicMock()
+    handle.done.return_value = False
+    handle.interrupted = False
+    handle.allow_interruptions = True
+    handle._agent_turn_context = None
+    activity._current_speech = handle
+
+    recognition = MagicMock()
+    recognition._end_of_turn_task = None  # the final hasn't arrived: no bounce yet
+    recognition._audio_interim_transcript = interim_words
+    activity._audio_recognition = recognition
+    return activity, handle
+
+
+async def test_first_breath_must_not_resume_a_preplayout_pause(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """SOS pauses the pending reply (timeout=0); the caller's first breath (bare
+    EOS, final still pending) must NOT resume it — the speech evidence (SOS +
+    word-bearing interims) says a turn decision is coming."""
+    monkeypatch.setenv("LIVEKIT_API_KEY", "k")
+    monkeypatch.setenv("LIVEKIT_API_SECRET", "s")
+
+    session = _session()
+    activity, handle = _activity_with_pending_reply(session, interim_words="thirteen one oh eight")
+
+    events: list[str] = []
+    session.on("agent_false_interruption", lambda _: events.append("resume"))
+
+    # the real pre-playout pause: caller starts speaking while the reply is unplayed
+    activity.on_start_of_speech(None, speech_start_time=time.time())
+    assert activity._paused_speech is not None
+    assert activity._paused_speech.timeout == 0
+
+    # mid-utterance breath: ink-2 closes the speech window; the final that will
+    # commit this turn is still in flight
+    activity.on_end_of_speech(None)
+
+    await asyncio.sleep(0.25)
+    await session.aclose()
+
+    # the pause must survive the breath: it is owned by the upcoming turn
+    # decision (which will commit and interrupt it), not by a 0-second timer
+    assert events == []
+    assert activity._paused_speech is not None
+
+
+async def test_speech_window_gap_must_not_resume_a_held_reply(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Same shape with a mid-playout pause (configured timeout): the gap between
+    two speech windows of ONE utterance must not resume the held reply. Distinct
+    from test_resume_is_immediate_when_no_turn_decision_is_open: there the EOS is
+    a lone noise glitch; here a real speech window (SOS, words) preceded it and
+    its final is still pending."""
+    monkeypatch.setenv("LIVEKIT_API_KEY", "k")
+    monkeypatch.setenv("LIVEKIT_API_SECRET", "s")
+
+    session = _session()
+    activity, handle = _activity_with_pending_reply(
+        session, interim_words="and this is like the sixth time"
+    )
+    activity._paused_speech = _PausedSpeechInfo(
+        handle=handle, agent_state="speaking", timeout=FALSE_INTERRUPTION_TIMEOUT
+    )
+
+    events: list[str] = []
+    session.on("agent_false_interruption", lambda _: events.append("resume"))
+
+    activity.on_start_of_speech(None, speech_start_time=time.time())  # window 1 (keeps the pause)
+    activity.on_end_of_speech(None)  # breath between windows; final still pending
+
+    await asyncio.sleep(FALSE_INTERRUPTION_TIMEOUT + 0.15)
+    await session.aclose()
+
+    assert events == []
+    assert activity._paused_speech is not None
+
+
+async def test_pending_transcript_deferral_is_bounded(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """If the pending transcript never materializes (e.g. the STT stream died),
+    the deferral gives up after ``_PENDING_TRANSCRIPT_MAX_DEFERRAL`` and the
+    plain timeout behavior rules — the pause must not be held forever."""
+    monkeypatch.setenv("LIVEKIT_API_KEY", "k")
+    monkeypatch.setenv("LIVEKIT_API_SECRET", "s")
+    from livekit.agents.voice import agent_activity as aa_mod
+
+    monkeypatch.setattr(aa_mod, "_PENDING_TRANSCRIPT_MAX_DEFERRAL", 0.4)
+
+    session = _session()
+    activity, handle = _activity_with_pending_reply(session, interim_words="thirteen one")
+
+    events: list[tuple[str, float]] = []
+    session.on("agent_false_interruption", lambda _: events.append(("resume", time.time())))
+
+    t0 = time.time()
+    activity.on_start_of_speech(None, speech_start_time=t0)
+    activity.on_end_of_speech(None)  # interims stay pending forever: no final ever arrives
+
+    await asyncio.sleep(0.8)
+    await session.aclose()
+
+    assert [name for name, _ in events] == ["resume"]
+    assert events[0][1] - t0 == pytest.approx(0.4, abs=0.2)