fix(voice): don't start reply playout while the user is speaking

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

@@ -2856,12 +2856,16 @@ def _on_first_frame(fut: asyncio.Future[float] | asyncio.Future[None]) -> None:
                     text_source = timed_texts
 
                 forward_audio_task, audio_out = perform_audio_forwarding(
-                    audio_output=audio_output, tts_output=tts_gen_data.audio_ch
+                    audio_output=audio_output,
+                    tts_output=tts_gen_data.audio_ch,
+                    hold_playout=lambda: self._hold_playout_if_user_speaking(speech_handle),
                 )
             else:
                 # use the provided audio
                 forward_audio_task, audio_out = perform_audio_forwarding(
-                    audio_output=audio_output, tts_output=audio
+                    audio_output=audio_output,
+                    tts_output=audio,
+                    hold_playout=lambda: self._hold_playout_if_user_speaking(speech_handle),
                 )
 
             audio_out.first_frame_fut.add_done_callback(_on_first_frame)
@@ -3369,6 +3373,7 @@ async def _next_segment() -> _SpeechSegment | None:
                 audio_source=audio_source,
                 text_source=text_source,
                 on_first_frame=_on_first_frame,
+                hold_playout=lambda: self._hold_playout_if_user_speaking(speech_handle),
             )
             segment_outputs.append(out)
             if speech_handle.interrupted:
@@ -4278,6 +4283,43 @@ def _pause_enabled(self) -> bool:
             and self._session.output.audio.can_pause
         )
 
+    def _hold_playout_if_user_speaking(self, speech_handle: SpeechHandle) -> bool:
+        """Level-triggered pre-playout hold, checked when audio forwarding starts.
+
+        The ``on_start_of_speech`` hold is edge-triggered: it can only pause a
+        speech that already exists when the user's onset arrives, and even a
+        hold it did place was released by ``_audio_forwarding_task``'s
+        unconditional ``resume()`` once the reply's TTS started streaming. A
+        user who starts speaking inside the reply's generation window (after
+        the turn commit, before the first TTS frame) therefore got the reply
+        launched into their speech. Re-checking the user state at the moment
+        audio forwarding starts closes that window.
+
+        Returns True when playout must stay held. Release is the existing
+        machinery, both directions: the user's end of speech arms the false
+        interruption timer (timeout 0 → immediate resume, unless an
+        interruption updated it), and a committed user turn interrupts the
+        paused speech so the superseding reply is generated instead.
+        """
+        if (
+            self._session.agent_state != "speaking"
+            and self._session.user_state == "speaking"
+            and self._pause_enabled()
+            and not speech_handle.interrupted
+            and speech_handle.allow_interruptions
+        ):
+            assert (audio_output := self._session.output.audio) is not None
+
+            # don't downgrade a timeout already recorded for this handle
+            # (_interrupt_by_audio_activity may have upgraded it to
+            # false_interruption_timeout); only place a fresh hold
+            if self._paused_speech is None or self._paused_speech.handle is not speech_handle:
+                self._update_paused_speech(speech_handle, timeout=0)
+            audio_output.pause()
+            return True
+
+        return False
+
     def _cancel_false_interruption_timer(self) -> None:
         if self._false_interruption_timer is not None:
             self._false_interruption_timer.cancel()

livekit-agents/livekit/agents/voice/generation.py

@@ -495,14 +495,17 @@ def perform_audio_forwarding(
     *,
     audio_output: io.AudioOutput,
     tts_output: AsyncIterable[rtc.AudioFrame],
+    hold_playout: Callable[[], bool] | None = None,
 ) -> tuple[asyncio.Task[None], _AudioOutput]:
     out = _AudioOutput(audio=[], first_frame_fut=asyncio.Future())
     # out.first_frame_fut should be cancelled in the caller after the playout is finished or interrupted
     audio_output.on("playback_started", out._resolve_first_frame_fut)
     out.first_frame_fut.add_done_callback(
         lambda _: audio_output.off("playback_started", out._resolve_first_frame_fut)
     )
-    task = asyncio.create_task(_audio_forwarding_task(audio_output, tts_output, out))
+    task = asyncio.create_task(
+        _audio_forwarding_task(audio_output, tts_output, out, hold_playout=hold_playout)
+    )
     return task, out
 
 
@@ -511,12 +514,18 @@ async def _audio_forwarding_task(
     audio_output: io.AudioOutput,
     tts_output: AsyncIterable[rtc.AudioFrame],
     out: _AudioOutput,
+    hold_playout: Callable[[], bool] | None = None,
 ) -> None:
     resampler: rtc.AudioResampler | None = None
 
     cancelled = False
     try:
-        audio_output.resume()
+        # this resume() clears a pause left over from an earlier speech; when the
+        # caller reports the output must stay held right now (e.g. the user is
+        # speaking at this very moment), skip it — the pause/false-interruption
+        # machinery owns the release.
+        if hold_playout is None or not hold_playout():
+            audio_output.resume()
 
         async for frame in tts_output:
             out.audio.append(frame)
@@ -588,6 +597,7 @@ async def forward_generation(
     audio_source: AsyncIterable[rtc.AudioFrame] | None,
     text_source: AsyncIterable[str] | None,
     on_first_frame: Callable[[asyncio.Future[Any], _AudioOutput | None], None],
+    hold_playout: Callable[[], bool] | None = None,
 ) -> _ForwardOutput:
     """Forward one segment's audio/text to the outputs, then wait for its playout.
 
@@ -602,7 +612,7 @@ async def forward_generation(
         audio_out: _AudioOutput | None = None
         if audio_output is not None and audio_source is not None:
             forward_audio_task, audio_out = perform_audio_forwarding(
-                audio_output=audio_output, tts_output=audio_source
+                audio_output=audio_output, tts_output=audio_source, hold_playout=hold_playout
             )
             forward_tasks.append(forward_audio_task)
             audio_out.first_frame_fut.add_done_callback(lambda fut: on_first_frame(fut, audio_out))

tests/test_playout_launch_hold.py

@@ -0,0 +1,156 @@
+"""A reply must not start playout into the user's speech.
+
+The pre-playout hold (``on_start_of_speech``) is edge-triggered, and
+``_audio_forwarding_task`` used to ``resume()`` the output unconditionally the
+moment the reply's TTS started streaming — so a user who resumed speaking
+inside the reply's generation window (after the turn commit, before the first
+TTS frame) had the reply launched into their speech, and any hold already
+placed was released mid-utterance.
+
+These tests cover the two halves of the fix:
+- ``_hold_playout_if_user_speaking`` — the level check + pause bookkeeping;
+- ``_audio_forwarding_task`` honoring ``hold_playout`` instead of blindly
+  resuming.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import MagicMock
+
+import pytest
+
+from livekit.agents.voice.agent_activity import AgentActivity
+from livekit.agents.voice.generation import _audio_forwarding_task, _AudioOutput
+
+pytestmark = pytest.mark.unit
+
+
+class _RecordingOutput:
+    """Minimal AudioOutput stand-in for driving ``_audio_forwarding_task``."""
+
+    def __init__(self) -> None:
+        self.paused = 0
+        self.resumed = 0
+        self.flushed = 0
+        self.sample_rate = None
+        self.can_pause = True
+
+    def pause(self) -> None:
+        self.paused += 1
+
+    def resume(self) -> None:
+        self.resumed += 1
+
+    def flush(self) -> None:
+        self.flushed += 1
+
+    def clear_buffer(self) -> None:
+        pass
+
+
+def _make_activity(
+    *,
+    agent_state: str = "thinking",
+    user_state: str = "speaking",
+    resume_false_interruption: bool = True,
+) -> tuple[AgentActivity, _RecordingOutput]:
+    activity: AgentActivity = AgentActivity.__new__(AgentActivity)
+    output = _RecordingOutput()
+    session = MagicMock()
+    session.agent_state = agent_state
+    session.user_state = user_state
+    session.output.audio = output
+    session.options.interruption = {
+        "resume_false_interruption": resume_false_interruption,
+        "false_interruption_timeout": 2.0,
+    }
+    activity._session = session
+    activity._paused_speech = None
+    return activity, output
+
+
+def _speech(*, interrupted: bool = False, allow_interruptions: bool = True) -> Any:
+    return MagicMock(interrupted=interrupted, allow_interruptions=allow_interruptions)
+
+
+def test_holds_when_user_speaking_at_playout_start() -> None:
+    activity, output = _make_activity()
+    speech = _speech()
+
+    assert activity._hold_playout_if_user_speaking(speech)
+    assert output.paused == 1
+    assert activity._paused_speech is not None
+    assert activity._paused_speech.handle is speech
+    assert activity._paused_speech.timeout == 0
+
+
+def test_hold_preserves_upgraded_timeout_for_same_handle() -> None:
+    # _interrupt_by_audio_activity may have raised this handle's timeout to
+    # false_interruption_timeout; re-holding at playout start must not downgrade
+    # it back to 0, or the resume fires the instant the user pauses
+    from livekit.agents.voice.agent_activity import _PausedSpeechInfo
+
+    activity, output = _make_activity()
+    speech = _speech()
+    activity._paused_speech = _PausedSpeechInfo(handle=speech, agent_state="thinking", timeout=2.0)
+
+    assert activity._hold_playout_if_user_speaking(speech)
+    assert output.paused == 1
+    assert activity._paused_speech.timeout == 2.0
+    assert activity._paused_speech.handle is speech
+
+
+def test_no_hold_when_user_silent() -> None:
+    activity, output = _make_activity(user_state="listening")
+    assert not activity._hold_playout_if_user_speaking(_speech())
+    assert output.paused == 0
+    assert activity._paused_speech is None
+
+
+def test_no_hold_when_agent_already_speaking() -> None:
+    # overlap with audible agent speech belongs to the interruption paths
+    activity, output = _make_activity(agent_state="speaking")
+    assert not activity._hold_playout_if_user_speaking(_speech())
+    assert output.paused == 0
+
+
+def test_no_hold_when_pause_disabled() -> None:
+    activity, output = _make_activity(resume_false_interruption=False)
+    assert not activity._hold_playout_if_user_speaking(_speech())
+    assert output.paused == 0
+
+
+def test_no_hold_for_uninterruptible_speech() -> None:
+    activity, output = _make_activity()
+    assert not activity._hold_playout_if_user_speaking(_speech(allow_interruptions=False))
+    assert not activity._hold_playout_if_user_speaking(_speech(interrupted=True))
+    assert output.paused == 0
+
+
+async def _no_frames() -> Any:
+    if False:  # pragma: no cover - async generator
+        yield
+
+
+@pytest.mark.asyncio
+async def test_forwarding_skips_resume_while_held() -> None:
+    output = _RecordingOutput()
+    out = _AudioOutput(audio=[], first_frame_fut=MagicMock())
+
+    await _audio_forwarding_task(output, _no_frames(), out, hold_playout=lambda: True)
+
+    assert output.resumed == 0
+    assert output.flushed == 1
+
+
+@pytest.mark.asyncio
+async def test_forwarding_resumes_when_not_held() -> None:
+    for hold_playout in (None, lambda: False):
+        output = _RecordingOutput()
+        out = _AudioOutput(audio=[], first_frame_fut=MagicMock())
+
+        await _audio_forwarding_task(output, _no_frames(), out, hold_playout=hold_playout)
+
+        assert output.resumed == 1
+        assert output.flushed == 1