feat(voice): expose SpeechHandle.hold_interruptions context manager

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

@@ -2057,11 +2057,11 @@ def _on_input_speech_started(self, _: llm.InputSpeechStartedEvent) -> None:
         try:
             self.interrupt()  # input_speech_started is also interrupting on the serverside realtime session  # noqa: E501
         except RuntimeError:
-            # only out of sync when the server cancelled its own response, with client-side turn
-            # taking an uninterruptible speech is expected
+            # held / allow_interruptions=False speech is expected under server turn detection;
+            # force=True still cuts through. Avoid exception-level noise on that path.
             if self._rt_turn_detection_enabled:
-                logger.exception(
-                    "RealtimeAPI input_speech_started, but current speech is not interruptable, this should never happen!"  # noqa: E501
+                logger.debug(
+                    "RealtimeAPI input_speech_started while current speech is not interruptable"
                 )
 
     def _on_input_speech_stopped(self, ev: llm.InputSpeechStoppedEvent) -> None:

livekit-agents/livekit/agents/voice/speech_handle.py

@@ -121,6 +121,10 @@ def allow_interruptions(self, value: bool) -> None:
         interruption requests until re-enabled. If the handle is already
         interrupted, clearing interruptions is not allowed.
 
+        While one or more ``hold_interruptions()`` holders are active, assignments
+        update the value restored after the final release; the effective
+        interruption state stays False until then.
+
         Args:
             value (bool): True to allow interruptions, False to disallow.
 
@@ -132,6 +136,11 @@ def allow_interruptions(self, value: bool) -> None:
                 "Cannot set allow_interruptions to False, the SpeechHandle is already interrupted"
             )
 
+        if self._interruption_holds > 0:
+            # Keep the hold effective; remember what to restore on final release.
+            self._interruption_holds_restore = value
+            return
+
         self._allow_interruptions = value
 
     def _hold_interruptions(self) -> None:
@@ -157,6 +166,25 @@ def _release_interruptions(self) -> None:
             with contextlib.suppress(RuntimeError):
                 self.allow_interruptions = self._interruption_holds_restore
 
+    @contextlib.contextmanager
+    def hold_interruptions(self) -> Generator[SpeechHandle, None, None]:
+        """Temporarily disallow interruptions on this speech (counted, restoring).
+
+        Nested or overlapping holders compose: the first holder remembers the previous
+        ``allow_interruptions`` value and the last release restores it. Assignments to
+        ``allow_interruptions`` during a hold update that restored value; the effective
+        state stays False until the final release.
+        ``interrupt(force=True)`` still cuts through a hold.
+
+        Yields:
+            SpeechHandle: this handle.
+        """
+        self._hold_interruptions()
+        try:
+            yield self
+        finally:
+            self._release_interruptions()
+
     @property
     def chat_items(self) -> list[llm.ChatItem]:
         return self._chat_items

tests/test_speech_handle_hold_interruptions.py

@@ -0,0 +1,100 @@
+"""Public SpeechHandle.hold_interruptions() context manager (issue #7191)."""
+
+from __future__ import annotations
+
+import pytest
+
+from livekit.agents.voice.speech_handle import SpeechHandle
+
+pytestmark = pytest.mark.unit
+
+
+def test_hold_interruptions_disallows_then_restores() -> None:
+    handle = SpeechHandle.create(allow_interruptions=True)
+
+    with handle.hold_interruptions() as held:
+        assert held is handle
+        assert handle.allow_interruptions is False
+        with pytest.raises(RuntimeError, match="does not allow interruptions"):
+            handle.interrupt()
+
+    assert handle.allow_interruptions is True
+    handle.interrupt()
+    assert handle.interrupted
+
+
+def test_hold_interruptions_nested_restores_once() -> None:
+    handle = SpeechHandle.create(allow_interruptions=True)
+
+    with handle.hold_interruptions():
+        assert handle.allow_interruptions is False
+        with handle.hold_interruptions():
+            assert handle.allow_interruptions is False
+        # inner release must not restore while outer still holds
+        assert handle.allow_interruptions is False
+
+    assert handle.allow_interruptions is True
+
+
+def test_hold_interruptions_preserves_prior_false() -> None:
+    handle = SpeechHandle.create(allow_interruptions=False)
+
+    with handle.hold_interruptions():
+        assert handle.allow_interruptions is False
+
+    assert handle.allow_interruptions is False
+
+
+def test_force_interrupt_bypasses_hold() -> None:
+    handle = SpeechHandle.create(allow_interruptions=True)
+
+    with handle.hold_interruptions():
+        handle.interrupt(force=True)
+        assert handle.interrupted
+
+
+def test_allow_interruptions_true_during_hold_restores_after_release() -> None:
+    handle = SpeechHandle.create(allow_interruptions=False)
+
+    with handle.hold_interruptions():
+        assert handle.allow_interruptions is False
+        handle.allow_interruptions = True
+        # assignment is deferred; hold stays effective
+        assert handle.allow_interruptions is False
+        with pytest.raises(RuntimeError, match="does not allow interruptions"):
+            handle.interrupt()
+
+    assert handle.allow_interruptions is True
+    handle.interrupt()
+    assert handle.interrupted
+
+
+def test_allow_interruptions_false_during_nested_holds_restores_after_final() -> None:
+    handle = SpeechHandle.create(allow_interruptions=True)
+
+    with handle.hold_interruptions():
+        assert handle.allow_interruptions is False
+        handle.allow_interruptions = True
+        assert handle.allow_interruptions is False
+        with handle.hold_interruptions():
+            handle.allow_interruptions = False
+            assert handle.allow_interruptions is False
+        # outer still holds; still not interruptible
+        assert handle.allow_interruptions is False
+        with pytest.raises(RuntimeError, match="does not allow interruptions"):
+            handle.interrupt()
+
+    assert handle.allow_interruptions is False
+    with pytest.raises(RuntimeError, match="does not allow interruptions"):
+        handle.interrupt()
+
+
+def test_force_interrupt_still_works_after_deferred_true_assignment() -> None:
+    handle = SpeechHandle.create(allow_interruptions=True)
+
+    with handle.hold_interruptions():
+        handle.allow_interruptions = True
+        assert handle.allow_interruptions is False
+        handle.interrupt(force=True)
+        assert handle.interrupted
+