fix: release stale transcript gates and playout pauses

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

@@ -4449,6 +4449,11 @@ async def _wait_for_auto_tool_reply() -> None:
                 "thinking" if self._background_speeches else "listening"
             )
 
+    def _disallow_interruptions(self, speech_handle: SpeechHandle) -> None:
+        speech_handle.allow_interruptions = False
+        if self._paused_speech is not None and self._paused_speech.handle is speech_handle:
+            self._reconcile_playout_pause(speech_handle)
+
     def _update_paused_speech(self, speech_handle: SpeechHandle, timeout: float) -> None:
         """Record that ``speech_handle`` is paused.
 

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

@@ -505,36 +505,32 @@ def _on_end_of_agent_speech(self, *, ended_at: float) -> None:
         This can occur while the generation remains active, such as when playout is paused.
         """
         self._cancel_backchannel_boundary()
+        agent_was_speaking = self._agent_speaking
 
-        if self._agent_speaking:
+        if agent_was_speaking:
             self._endpointing.on_end_of_agent_speech(ended_at=ended_at)
+        # Replayed STT events must observe the post-playout state.
+        self._agent_speaking = False
+
         if not self._adaptive_interruption_active:
             self._flush_held_transcripts()
             self._overlap_open = False
-            self._agent_speaking = False
             self._agent_speech_started_at = None
             return
 
-        if self._agent_speaking:
+        if agent_was_speaking:
             # close any unresolved overlap before resetting the detector
             self._on_end_of_overlap_speech(ended_at=ended_at, agent_ended=True)
 
         self._interruption_ch.send_nowait(_AgentSpeechEndedSentinel())  # type: ignore[union-attr]
 
-        if self._agent_speaking and self._transcript_gate_active:
-            logger.trace(
-                "flushing held transcripts",
-                extra={
-                    "vad_speech_started_at": self._active_vad_speech_started_at,
-                },
-            )
-            self._flush_held_transcripts(
-                resolved_at=ended_at,
-                vad_speech_started_at=self._active_vad_speech_started_at,
-            )
-
         self._overlap_open = False
-        self._agent_speaking = False
+
+        self._flush_held_transcripts(
+            resolved_at=ended_at,
+            vad_speech_started_at=self._active_vad_speech_started_at,
+        )
+
         self._agent_speech_started_at = None
 
     def _on_start_of_speech(
@@ -698,6 +694,18 @@ def _flush_held_transcripts(
         vad_speech_started_at: float | None = None,
     ) -> None:
         """Stop holding transcripts and emit the retained events in provider order."""
+        gate_was_active = self._transcript_gate_active
+        if gate_was_active or self._transcript_buffer:
+            logger.trace(
+                "flushing held transcripts",
+                extra={
+                    "event_count": len(self._transcript_buffer),
+                    "gate_was_active": gate_was_active,
+                    "resolved_at": resolved_at,
+                    "vad_speech_started_at": vad_speech_started_at,
+                },
+            )
+
         self._transcript_gate_active = False
         if resolved_at is not None:
             self._trim_held_transcripts(

livekit-agents/livekit/agents/voice/events.py

@@ -50,7 +50,9 @@ def __init__(
         session: AgentSession[Userdata_T],
         speech_handle: SpeechHandle,
         function_call: FunctionCall,
+        activity: AgentActivity | None = None,
     ) -> None:
+        self._activity = activity
         self._session = session
         self._speech_handle = speech_handle
         self._function_call = function_call
@@ -88,13 +90,15 @@ def userdata(self) -> Userdata_T:
     def disallow_interruptions(self) -> None:
         """Disable interruptions for this FunctionCall.
 
-        Delegates to the SpeechHandle.allow_interruptions setter,
-        which will raise a RuntimeError if the handle is already interrupted.
+        Also releases an active false-interruption pause.
 
         Raises:
             RuntimeError: If the SpeechHandle is already interrupted.
         """
-        self.speech_handle.allow_interruptions = False
+        if self._activity is not None:
+            self._activity._disallow_interruptions(self.speech_handle)
+        else:
+            self.speech_handle.allow_interruptions = False
 
     async def wait_for_playout(self) -> None:
         """Waits for the speech playout corresponding to this function call step.

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

@@ -875,7 +875,10 @@ def _tool_completed(out: ToolExecutionOutput) -> None:
                 mocked = mock is not None
 
                 run_ctx = RunContext(
-                    session=session, speech_handle=speech_handle, function_call=fnc_call
+                    activity=activity,
+                    session=session,
+                    speech_handle=speech_handle,
+                    function_call=fnc_call,
                 )
 
                 logger.debug(

tests/test_disallow_interruptions_pause.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from livekit.agents import Agent, RunContext, function_tool
+from livekit.agents.llm import FunctionToolCall
+from livekit.agents.voice.io import PlaybackFinishedEvent
+
+from .fake_session import FakeActions, create_session, run_session
+
+pytestmark = [pytest.mark.unit, pytest.mark.virtual_time, pytest.mark.no_concurrent]
+
+TRANSFER_MESSAGE = "I'll connect you now."
+
+
+class TransferAgent(Agent):
+    def __init__(self) -> None:
+        super().__init__(instructions="Transfer callers when requested.")
+        self.transfer_started = asyncio.Event()
+        self.transfer_completed = asyncio.Event()
+
+    @function_tool
+    async def transfer_to_live_agent(self, context: RunContext) -> None:
+        """Transfer the caller to a live agent."""
+        self.transfer_started.set()
+        context.disallow_interruptions()
+        context.session.input.set_audio_enabled(False)
+        await context.session.say(TRANSFER_MESSAGE, allow_interruptions=False)
+        self.transfer_completed.set()
+
+
+async def test_uninterruptible_tool_speech_plays_after_paused_parent() -> None:
+    actions = FakeActions()
+    actions.add_user_speech(0.5, 2.5, "Connect me to an agent.")
+    actions.add_llm(
+        content="Happy to connect you.",
+        tool_calls=[
+            FunctionToolCall(
+                name="transfer_to_live_agent",
+                arguments="{}",
+                call_id="transfer-1",
+            )
+        ],
+        ttft=0.1,
+        duration=1.0,
+    )
+    # VAD pauses the parent after its first frame is forwarded but before the tool starts.
+    actions.add_tts(2.0, ttfb=0.1, duration=0.2)
+    actions.add_user_speech(3.1, 5.0, "")
+    actions.add_tts(1.0, input=TRANSFER_MESSAGE)
+
+    session = create_session(actions, can_pause_audio=True)
+    agent = TransferAgent()
+    playback_finished_events: list[PlaybackFinishedEvent] = []
+    session.output.audio.on("playback_finished", playback_finished_events.append)
+
+    run_task = asyncio.create_task(run_session(session, agent))
+    try:
+        await asyncio.wait_for(agent.transfer_started.wait(), timeout=5.0)
+        await asyncio.wait_for(agent.transfer_completed.wait(), timeout=5.0)
+    finally:
+        if not run_task.done():
+            await session.aclose()
+        await run_task
+
+    assert [event.playback_position for event in playback_finished_events] == pytest.approx(
+        [2.0, 1.0], abs=0.02
+    )

tests/test_realtime_adaptive_interruption.py

@@ -565,6 +565,42 @@ def test_end_of_speech_passes_interruption_verdict_to_endpointing(
     )
 
 
+def test_agent_speech_end_keeps_gate_closed_during_transcript_replay() -> None:
+    ar, _ = _recognition_with_interruption_ch()
+    ar._on_start_of_agent_speech(started_at=9.0)
+    ar._on_start_of_speech(started_at=9.5)
+    ar._active_vad_speech_started_at = 9.5
+    ar._transcript_buffer.append(MagicMock(created_at=9.5, speech_end_time=None))
+    ar._process_stt_event = MagicMock(  # type: ignore[method-assign]
+        side_effect=lambda _: ar._on_start_of_speech(started_at=9.5)
+    )
+
+    ar._on_end_of_agent_speech(ended_at=10.0)
+
+    assert not ar._agent_speaking
+    assert not ar._transcript_gate_active
+    assert not ar._transcript_buffer
+
+
+@pytest.mark.parametrize("gate_active", [False, True])
+def test_agent_speech_end_releases_stale_transcripts_when_already_not_speaking(
+    gate_active: bool,
+) -> None:
+    ar, _ = _recognition_with_interruption_ch()
+    ar._agent_speech_started_at = 9.0
+    ar._active_vad_speech_started_at = 9.5
+    ar._transcript_gate_active = gate_active
+    held_event = MagicMock(created_at=9.5, speech_end_time=None)
+    ar._transcript_buffer.append(held_event)
+    ar._process_stt_event = MagicMock()  # type: ignore[method-assign]
+
+    ar._on_end_of_agent_speech(ended_at=10.0)
+
+    ar._process_stt_event.assert_called_once_with(held_event)  # type: ignore[attr-defined]
+    assert not ar._transcript_gate_active
+    assert not ar._transcript_buffer
+
+
 async def test_agent_speech_end_closes_overlap_before_reset() -> None:
     ar, ch = _recognition_with_interruption_ch()
     ar._on_start_of_agent_speech(started_at=time.time())