fix(voice): stop the interrupt walk at a queued speech that disallows interruptions

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

@@ -1509,9 +1509,16 @@ def _interrupt_background_speeches(self, force: bool = False) -> list[SpeechHand
     def interrupt(self, *, force: bool = False) -> asyncio.Future[None]:
         """Interrupt the current speech generation and any queued speeches.
 
+        A queued speech that disallows interruptions keeps playing, along with the ones
+        behind it, unless ``force`` is set.
+
         Returns:
             An asyncio.Future that completes when the interruption is fully processed
             and chat context has been updated
+
+        Raises:
+            RuntimeError: If the speech currently playing disallows interruptions and
+                ``force`` is False.
         """
         self._cancel_preemptive_generation()
 
@@ -1523,13 +1530,25 @@ def interrupt(self, *, force: bool = False) -> asyncio.Future[None]:
             self._current_speech.interrupt(force=force)
             interrupted_speeches.append(self._current_speech)
 
-        for _, _, speech in self._speech_q:
-            speech.interrupt(force=force)
-            interrupted_speeches.append(speech)
-
         if self._rt_session is not None:
             self._rt_session.interrupt()
 
+        # _speech_q is a heap, so its list order is not the order it pops in
+        for _, _, speech in sorted(self._speech_q, key=lambda item: (item[0], item[1])):
+            try:
+                speech.interrupt(force=force)
+            except RuntimeError:
+                # the speeches behind this one are going to play, so stopping
+                # here keeps the conversation contiguous
+                logger.warning(
+                    "a queued speech does not allow interruptions and will play after the "
+                    "interruption, use interrupt(force=True) to interrupt it as well",
+                    extra={"speech_id": speech.id},
+                )
+                break
+
+            interrupted_speeches.append(speech)
+
         if not interrupted_speeches:
             future.set_result(None)
         else:

livekit-agents/livekit/agents/voice/agent_session.py

@@ -1368,9 +1368,16 @@ def generate_reply(
     def interrupt(self, *, force: bool = False) -> asyncio.Future[None]:
         """Interrupt the current speech generation.
 
+        A queued speech created with ``allow_interruptions=False`` keeps playing,
+        along with the ones behind it, unless ``force`` is set.
+
         Returns:
             An asyncio.Future that completes when the interruption is fully processed
             and chat context has been updated.
+
+        Raises:
+            RuntimeError: If the session isn't running, or if the speech currently
+                playing disallows interruptions and ``force`` is False.
         """
         if self._activity is None:
             raise RuntimeError("AgentSession isn't running")

tests/test_interrupt_protected_speech.py

@@ -0,0 +1,149 @@
+"""``AgentActivity.interrupt()`` and queued speeches that disallow interruptions.
+
+``SpeechHandle.interrupt()`` raises for such a handle, and the queue loop used
+to let that raise escape: the remaining queued speeches were left playing and
+the returned future never resolved. Interrupting *past* the protected speech
+would be wrong too — the ones behind it still play, so skipping one in the
+middle would leave a gap in the conversation.
+"""
+
+import heapq
+import logging
+import time
+from unittest.mock import Mock
+
+import pytest
+
+from livekit.agents.voice.agent_activity import AgentActivity
+from livekit.agents.voice.speech_handle import SpeechHandle
+
+from .fake_session import FakeActions, create_session
+from .test_agent_session import MyAgent, _close_test_session
+
+pytestmark = pytest.mark.unit
+
+
+def _make_activity() -> AgentActivity:
+    return AgentActivity(MyAgent(), create_session(FakeActions()))
+
+
+def _enqueue(activity: AgentActivity, speech: SpeechHandle, *, priority: int = 0) -> None:
+    """Queue a speech the way _schedule_speech does (a heap, not a list)."""
+    heapq.heappush(activity._speech_q, (-priority, time.perf_counter_ns(), speech))
+
+
+class TestInterruptQueuedSpeeches:
+    async def test_queue_stops_at_the_protected_speech(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        activity = _make_activity()
+        activity._rt_session = Mock()
+        current = SpeechHandle.create(allow_interruptions=True)
+        first = SpeechHandle.create(allow_interruptions=True)
+        protected = SpeechHandle.create(allow_interruptions=False)
+        behind = SpeechHandle.create(allow_interruptions=True)
+        activity._current_speech = current
+        for speech in (first, protected, behind):
+            _enqueue(activity, speech)
+
+        try:
+            with caplog.at_level(logging.WARNING, logger="livekit.agents"):
+                activity.interrupt()  # must not raise
+
+            assert current.interrupted
+            assert first.interrupted
+            assert not protected.interrupted
+            # no hole: what plays after the protected speech is untouched
+            assert not behind.interrupted
+            # the rest of the sequence still ran
+            activity._rt_session.interrupt.assert_called_once()
+            assert any("force=True" in record.message for record in caplog.records)
+        finally:
+            await _close_test_session(activity._session)
+
+    async def test_a_protected_head_shields_the_whole_queue(self) -> None:
+        # [protected, interruptible, protected]: interrupting only the middle
+        # one would play the first and third with a gap between them
+        activity = _make_activity()
+        activity._rt_session = Mock()
+        head = SpeechHandle.create(allow_interruptions=False)
+        middle = SpeechHandle.create(allow_interruptions=True)
+        tail = SpeechHandle.create(allow_interruptions=False)
+        for speech in (head, middle, tail):
+            _enqueue(activity, speech)
+
+        try:
+            activity.interrupt()
+
+            assert not head.interrupted
+            assert not middle.interrupted
+            assert not tail.interrupted
+        finally:
+            await _close_test_session(activity._session)
+
+    async def test_the_queue_is_walked_in_playout_order_not_heap_order(self) -> None:
+        # a higher-priority speech is queued last but plays first; the heap's
+        # list order does not reflect that, the walk must
+        activity = _make_activity()
+        activity._rt_session = Mock()
+        low = SpeechHandle.create(allow_interruptions=True)
+        urgent_protected = SpeechHandle.create(allow_interruptions=False)
+        _enqueue(activity, low, priority=SpeechHandle.SPEECH_PRIORITY_NORMAL)
+        _enqueue(activity, urgent_protected, priority=SpeechHandle.SPEECH_PRIORITY_HIGH)
+
+        try:
+            activity.interrupt()
+
+            # the protected speech plays first, so the walk stops immediately
+            assert not urgent_protected.interrupted
+            assert not low.interrupted
+        finally:
+            await _close_test_session(activity._session)
+
+    async def test_a_protected_playing_speech_still_raises(self) -> None:
+        # unchanged behaviour: SpeechHandle.interrupt() is explicit about it
+        activity = _make_activity()
+        activity._current_speech = SpeechHandle.create(allow_interruptions=False)
+
+        try:
+            with pytest.raises(RuntimeError):
+                activity.interrupt()
+        finally:
+            await _close_test_session(activity._session)
+
+    async def test_force_interrupts_the_whole_chain(self) -> None:
+        activity = _make_activity()
+        activity._rt_session = Mock()
+        current = SpeechHandle.create(allow_interruptions=False)
+        queued = SpeechHandle.create(allow_interruptions=False)
+        behind = SpeechHandle.create(allow_interruptions=True)
+        activity._current_speech = current
+        for speech in (queued, behind):
+            _enqueue(activity, speech)
+
+        try:
+            activity.interrupt(force=True)
+
+            assert current.interrupted
+            assert queued.interrupted
+            assert behind.interrupted
+            activity._rt_session.interrupt.assert_called_once()
+        finally:
+            await _close_test_session(activity._session)
+
+    async def test_interruptible_chain_is_unaffected(self) -> None:
+        activity = _make_activity()
+        activity._rt_session = Mock()
+        current = SpeechHandle.create(allow_interruptions=True)
+        queued = SpeechHandle.create(allow_interruptions=True)
+        activity._current_speech = current
+        _enqueue(activity, queued)
+
+        try:
+            activity.interrupt()
+
+            assert current.interrupted
+            assert queued.interrupted
+            activity._rt_session.interrupt.assert_called_once()
+        finally:
+            await _close_test_session(activity._session)