feat(voice): expose a counted SpeechHandle interruption hold

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

@@ -2057,12 +2057,23 @@ 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
-            if self._rt_turn_detection_enabled:
+            # with client-side turn taking an uninterruptible speech is expected, and so is a
+            # speech someone deliberately held (`SpeechHandle.hold_interruptions`) -- that is
+            # what a hold is for, and `interrupt(force=True)` still cuts through it.
+            # Anything else is the desync this log was added for: the server cancelled its own
+            # response while this speech still believed it could not be interrupted.
+            speech = self._current_speech
+            if self._rt_turn_detection_enabled and not (
+                speech is not None and speech.interruptions_held
+            ):
                 logger.exception(
                     "RealtimeAPI input_speech_started, but current speech is not interruptable, this should never happen!"  # noqa: E501
                 )
+            else:
+                logger.debug(
+                    "RealtimeAPI input_speech_started while the current speech is held",
+                    extra={"speech_id": speech.id if speech is not None else None},
+                )
 
     def _on_input_speech_stopped(self, ev: llm.InputSpeechStoppedEvent) -> None:
         if self.vad is None or self.using_default_vad:

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

@@ -36,7 +36,6 @@ def __init__(
         self._id = speech_id
         self._allow_interruptions = allow_interruptions
         self._interruption_holds = 0
-        self._interruption_holds_restore = allow_interruptions
         self._input_details = input_details
 
         self._interrupt_fut = asyncio.Future[None]()
@@ -111,7 +110,17 @@ def interrupted(self) -> bool:
 
     @property
     def allow_interruptions(self) -> bool:
-        return self._allow_interruptions
+        """Whether this speech may be interrupted right now.
+
+        The **effective** state: the value last assigned, unless one or more
+        `hold_interruptions()` holders are active, in which case it is False for as long
+        as they are.
+
+        A hold and an assignment are kept separately, so neither can lose the other: an
+        assignment made during a hold cannot defeat the hold, and the last release cannot
+        discard an assignment made while it was held.
+        """
+        return self._allow_interruptions and self._interruption_holds == 0
 
     @allow_interruptions.setter
     def allow_interruptions(self, value: bool) -> None:
@@ -121,6 +130,9 @@ def allow_interruptions(self, value: bool) -> None:
         interruption requests until re-enabled. If the handle is already
         interrupted, clearing interruptions is not allowed.
 
+        Assigning while a `hold_interruptions()` holder is active records the value
+        without lifting the hold; it takes effect when the last holder releases.
+
         Args:
             value (bool): True to allow interruptions, False to disallow.
 
@@ -134,28 +146,90 @@ def allow_interruptions(self, value: bool) -> None:
 
         self._allow_interruptions = value
 
+    @property
+    def interruptions_held(self) -> bool:
+        """Whether a `hold_interruptions()` holder is keeping this speech uninterruptible.
+
+        Narrower than `not allow_interruptions`, which is also True for a speech simply
+        created or configured uninterruptible.
+        """
+        return self._interruption_holds > 0
+
     def _hold_interruptions(self) -> None:
         """Disallow interruptions until every hold taken here is released.
 
         Counted rather than set, because the holders of one speech are not serialised
         against each other: the inline tasks awaited from a turn's parallel tool calls run
         one at a time, and a hold released between them would let the user turns of one
-        task's sub-conversation interrupt the speech the rest are still anchored to. An
-        interrupted handle can no longer disallow interruptions, so those tasks could then
-        never run. The first holder owns the value the last one restores.
+        task's sub-conversation interrupt the speech the rest are still anchored to.
+
+        The count lives beside the assigned `allow_interruptions` value rather than
+        overwriting it. Overwriting it meant an assignment during a hold silently defeated
+        the hold -- `allow_interruptions = True` under a held speech made it interruptible
+        again -- and meant the last release overwrote whatever had been assigned since.
         """
-        if self._interruption_holds == 0:
-            self._interruption_holds_restore = self._allow_interruptions
-            self.allow_interruptions = False
+        if self.interrupted:
+            raise RuntimeError("Cannot hold interruptions, the SpeechHandle is already interrupted")
 
         self._interruption_holds += 1
 
     def _release_interruptions(self) -> None:
-        self._interruption_holds -= 1
-        if self._interruption_holds == 0:
-            # a forced interrupt lands regardless of the hold, and leaves nothing to restore
-            with contextlib.suppress(RuntimeError):
-                self.allow_interruptions = self._interruption_holds_restore
+        # Clamped rather than asserted: this runs from a `finally`, where raising would
+        # replace whatever exception is already unwinding.
+        if self._interruption_holds > 0:
+            self._interruption_holds -= 1
+
+    @contextlib.contextmanager
+    def hold_interruptions(self) -> Generator[SpeechHandle, None, None]:
+        """Keep this speech uninterruptible for the duration of the block.
+
+        Counted, and independent of `allow_interruptions`: overlapping holders compose,
+        and the speech becomes interruptible again only once the last one releases.
+
+        Written for wording that has to arrive whole -- a consent question, a regulatory
+        disclosure, a handover announcement -- on a session whose realtime model does
+        server-side turn detection, where `allow_interruptions=False` is dropped with a
+        warning per `say()` and refused outright per session.
+
+        `interrupt(force=True)` still lands, so a hold survives ordinary barge-in without
+        making the speech impossible to stop.
+
+        **What a hold does not do**, in the order it bites:
+
+        1. *It protects LiveKit-side playout, not a provider's own generation.* `say()`
+           speaks through the TTS plugin when there is one, and that audio is LiveKit's
+           to protect. With no TTS and a realtime model whose
+           `capabilities.supports_say` is True, the provider generates the audio and
+           cancels it itself when its own turn detection fires -- a hold cannot reach
+           that, so a held speech on that path can still end mid-sentence. Attach a TTS
+           model for wording that has to arrive whole.
+        2. *The caller is not heard while it plays.* Their audio is replaced with silence
+           on the paths feeding the STT and the realtime model, unless the session opts
+           out with
+           `turn_handling=TurnHandlingOptions(interruption={"discard_audio_if_uninterruptible": False})`.
+           Barge-in is not merely ignored by default; it is discarded.
+        3. *A user turn completing during a hold generates no reply for that turn.* A hold
+           is for wording, not for a question whose answer is expected mid-sentence.
+
+        Example:
+            ```python
+            handle = session.say("Before we go on, do I have your consent?")
+            with handle.hold_interruptions():
+                await handle.wait_for_playout()
+            ```
+
+        Yields:
+            SpeechHandle: this handle, so the block may be written as
+                `with handle.hold_interruptions() as speech:`.
+
+        Raises:
+            RuntimeError: If the speech has already been interrupted.
+        """
+        self._hold_interruptions()
+        try:
+            yield self
+        finally:
+            self._release_interruptions()
 
     @property
     def chat_items(self) -> list[llm.ChatItem]:
@@ -196,7 +270,7 @@ def interrupt(self, *, force: bool = False) -> SpeechHandle:
             # already cancelled or finished: nothing to interrupt, and protection is moot
             return self
 
-        if not force and not self._allow_interruptions:
+        if not force and not self.allow_interruptions:
             raise RuntimeError("This generation handle does not allow interruptions")
 
         self._cancel()

tests/test_realtime_input_speech_interrupt.py

@@ -80,3 +80,42 @@ def test_input_speech_started_interrupts_interruptible_speech() -> None:
 
     assert handle.interrupted is True
     activity._rt_session.interrupt.assert_called_once()
+
+
+def test_input_speech_started_keeps_a_held_speech(caplog: pytest.LogCaptureFixture) -> None:
+    # the point of a hold: server-side turn detection stays on, the model still reports user
+    # speech, and the speech someone deliberately held keeps playing through it. No
+    # response.cancel is sent, because AgentActivity.interrupt() raises before reaching it.
+    activity = _activity(server_turn_detection=True)
+    handle = SpeechHandle.create(allow_interruptions=True)
+    activity._current_speech = handle
+    activity._rt_session = MagicMock()
+
+    with handle.hold_interruptions():
+        with caplog.at_level(logging.DEBUG, logger="livekit.agents"):
+            activity._on_input_speech_started(llm.InputSpeechStartedEvent())
+
+        assert handle.interrupted is False
+        activity._rt_session.interrupt.assert_not_called()
+
+    assert not [record for record in caplog.records if record.levelno >= logging.ERROR]
+    assert any("held" in record.message for record in caplog.records)
+    handle._mark_done()
+
+
+def test_input_speech_started_still_reports_an_unheld_uninterruptible_speech(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    # the loud log is narrowed, not removed. A speech that refuses interruption *without*
+    # anyone holding it is the desync the log was added for -- the server cancelled its own
+    # response while this speech still believed it could not be interrupted.
+    activity = _activity(server_turn_detection=True)
+
+    with caplog.at_level(logging.ERROR, logger="livekit.agents"):
+        handle = _speech_started(activity, allow_interruptions=False)
+
+    assert handle.interrupted is False
+    assert [record for record in caplog.records if record.levelno >= logging.ERROR], (
+        "an uninterruptible speech nobody held is still reported loudly"
+    )
+    handle._mark_done()

tests/test_speech_handle_hold_interruptions.py

@@ -0,0 +1,152 @@
+"""``SpeechHandle.hold_interruptions()``: the counted, public interruption hold.
+
+The count is kept beside the assigned ``allow_interruptions`` value rather than
+overwriting it. Overwriting it is what the private hold used to do, and it lost
+information in both directions: an assignment during a hold defeated the hold, and
+the last release discarded whatever had been assigned since. Both are covered below.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from livekit.agents.voice.speech_handle import SpeechHandle
+
+pytestmark = pytest.mark.unit
+
+
+def _handle(*, allow_interruptions: bool = True) -> SpeechHandle:
+    return SpeechHandle.create(allow_interruptions=allow_interruptions)
+
+
+async def test_a_hold_makes_the_speech_uninterruptible_and_restores_it() -> None:
+    handle = _handle()
+
+    with handle.hold_interruptions() as held:
+        assert held is handle
+        assert handle.allow_interruptions is False
+        assert handle.interruptions_held is True
+        with pytest.raises(RuntimeError, match="does not allow interruptions"):
+            handle.interrupt()
+
+    assert handle.allow_interruptions is True
+    assert handle.interruptions_held is False
+    handle.interrupt()
+    assert handle.interrupted
+
+
+async def test_overlapping_holds_release_only_on_the_last_one() -> None:
+    handle = _handle()
+
+    with handle.hold_interruptions():
+        with handle.hold_interruptions():
+            assert handle.allow_interruptions is False
+        assert handle.allow_interruptions is False, "the outer holder still holds"
+
+    assert handle.allow_interruptions is True
+    handle._mark_done()
+
+
+async def test_a_hold_does_not_make_an_uninterruptible_speech_interruptible() -> None:
+    """The release restores nothing -- it only stops holding."""
+    handle = _handle(allow_interruptions=False)
+
+    with handle.hold_interruptions():
+        assert handle.allow_interruptions is False
+
+    assert handle.allow_interruptions is False
+    handle._mark_done()
+
+
+async def test_an_assignment_during_a_hold_cannot_defeat_the_hold() -> None:
+    """The regression. The hold used to overwrite ``allow_interruptions``, so assigning
+    it back to True during a hold made the speech interruptible again -- the hold was
+    silently gone while its holder still believed it was protected."""
+    handle = _handle()
+
+    with handle.hold_interruptions():
+        handle.allow_interruptions = True
+
+        assert handle.allow_interruptions is False, "still held"
+        with pytest.raises(RuntimeError, match="does not allow interruptions"):
+            handle.interrupt()
+
+    handle._mark_done()
+
+
+async def test_an_assignment_during_a_hold_survives_the_release() -> None:
+    """And the other direction: the release must not discard it."""
+    handle = _handle()
+
+    with handle.hold_interruptions():
+        handle.allow_interruptions = False
+
+    assert handle.allow_interruptions is False, "the assignment outlives the hold"
+    handle._mark_done()
+
+
+async def test_a_forced_interrupt_lands_through_a_hold() -> None:
+    """A hold survives barge-in; it does not make a speech impossible to stop."""
+    handle = _handle()
+
+    with handle.hold_interruptions():
+        handle.interrupt(force=True)
+        assert handle.interrupted
+
+
+async def test_a_release_after_a_forced_interrupt_does_not_raise() -> None:
+    handle = _handle()
+
+    with handle.hold_interruptions():
+        handle.interrupt(force=True)
+
+    assert handle.interrupted
+    assert handle.interruptions_held is False
+
+
+async def test_an_already_interrupted_speech_refuses_to_be_held() -> None:
+    """Holding a speech that has already been cut off would report protection that
+    cannot exist, and leave the holder waiting on a speech that is not playing."""
+    handle = _handle()
+    handle.interrupt()
+
+    with pytest.raises(RuntimeError, match="already interrupted"):
+        with handle.hold_interruptions():
+            pass
+
+    assert handle.interruptions_held is False, "the failed hold left the count balanced"
+    handle._mark_done()
+
+
+async def test_a_hold_released_by_an_exception_still_releases() -> None:
+    handle = _handle()
+
+    with pytest.raises(ValueError, match="boom"):
+        with handle.hold_interruptions():
+            raise ValueError("boom")
+
+    assert handle.interruptions_held is False
+    assert handle.allow_interruptions is True
+    handle._mark_done()
+
+
+async def test_repeated_hold_cycles_do_not_accumulate() -> None:
+    handle = _handle()
+
+    for _ in range(3):
+        with handle.hold_interruptions():
+            assert handle.allow_interruptions is False
+        assert handle.allow_interruptions is True
+
+    assert handle.interruptions_held is False
+    handle._mark_done()
+
+
+async def test_interruptions_held_is_narrower_than_allow_interruptions() -> None:
+    """A speech configured uninterruptible is not a *held* speech, and the realtime
+    barge-in path tells them apart to decide whether a refusal is expected."""
+    handle = _handle(allow_interruptions=False)
+
+    assert handle.allow_interruptions is False
+    assert handle.interruptions_held is False
+    handle._mark_done()