Surface server-side interruption from Nova Sonic and Ultravox

BaseOutputTransport only clears buffered audio mid-playback on
InterruptionFrame. Realtime services stream audio downstream as fast as
they produce it, and playback necessarily trails the buffer — so when the
user interrupts, the bot keeps talking past the interruption unless the
service surfaces the interruption to the pipeline.

Two realtime services were missing this signal:

  - AWS Nova Sonic acknowledged the INTERRUPTED stop reason internally
    (closing its own response state) but never broadcast InterruptionFrame.
  - Ultravox's playback_clear_buffer message — the server's explicit
    "drop buffered output audio" signal for interruptions — was not
    handled at all.

In both cases the latent bug was masked by enabling local VAD on the
user aggregator, which produced UserStartedSpeakingFrame and triggered
the aggregator-side interruption path. The realtime context aggregator
work makes local VAD optional, so the underlying gap needs fixing first.

Wire broadcast_interruption() into both services on the server-side
interruption signal, firing before the response-end signal so the
assistant aggregator marks the message interrupted=True before
LLMFullResponseEndFrame closes the turn.

changelog/+nova-sonic-server-interruption.fixed.md

@@ -0,0 +1 @@
+- Fixed AWS Nova Sonic not surfacing server-side interruption. When the user interrupted the bot mid-response, the `INTERRUPTED` stop reason was acknowledged internally but no `InterruptionFrame` was emitted, so `BaseOutputTransport` kept draining its audio buffer and the bot kept talking past the interruption. Nova Sonic now broadcasts `InterruptionFrame` on both `INTERRUPTED` paths (text-stage and audio-stage). This was previously masked by enabling local VAD on the user aggregator, which generated `UserStartedSpeakingFrame` and triggered the aggregator-side interruption path; the fix makes the behavior correct without local VAD as a workaround.

changelog/+ultravox-server-interruption.fixed.md

@@ -0,0 +1 @@
+- Fixed Ultravox Realtime not surfacing server-side interruption. The server sends a `playback_clear_buffer` message when the user interrupts the bot mid-speech, instructing clients to drop buffered output audio; this was previously unhandled, so `BaseOutputTransport` kept playing the buffered audio and the bot kept talking past the interruption. Ultravox now broadcasts `InterruptionFrame` on `playback_clear_buffer`. This was previously masked by enabling local VAD on the user aggregator, which generated `UserStartedSpeakingFrame` and triggered the aggregator-side interruption path; the fix makes the behavior correct without local VAD as a workaround.

src/pipecat/services/aws/nova_sonic/llm.py

@@ -1434,7 +1434,14 @@ class AWSNovaSonicLLMService(LLMService[AWSNovaSonicLLMAdapter]):
                 else:
                     if self._assistant_is_responding:
                         # TEXT INTERRUPTED before audio started means no AUDIO
-                        # contentEnd will arrive — end the response here.
+                        # contentEnd will arrive — end the response here. Emit
+                        # InterruptionFrame upstream so the assistant aggregator
+                        # marks the message interrupted=True, and downstream so
+                        # BaseOutputTransport can clear any audio it had already
+                        # buffered. Must fire before _report_assistant_response_ended
+                        # so the aggregator handles InterruptionFrame before
+                        # LLMFullResponseEndFrame closes the turn.
+                        await self.broadcast_interruption()
                         self._assistant_is_responding = False
                         await self._report_assistant_response_ended()
                     # Session continuation: TEXT INTERRUPTED is a completion
@@ -1447,6 +1454,18 @@ class AWSNovaSonicLLMService(LLMService[AWSNovaSonicLLMAdapter]):
                 if stop_reason in ("END_TURN", "INTERRUPTED"):
                     # END_TURN: normal completion. INTERRUPTED: user interrupted
                     # mid-audio. Both mean no more audio for this turn.
+                    if stop_reason == "INTERRUPTED":
+                        # Emit InterruptionFrame upstream so the assistant
+                        # aggregator marks the message interrupted=True, and
+                        # downstream so BaseOutputTransport clears the audio
+                        # buffer (without this the bot keeps talking past the
+                        # interruption while the buffer drains, since Nova
+                        # Sonic doesn't surface server-side interruption any
+                        # other way). Must fire before
+                        # _report_assistant_response_ended so the aggregator
+                        # handles InterruptionFrame before LLMFullResponseEndFrame
+                        # closes the turn.
+                        await self.broadcast_interruption()
                     self._assistant_is_responding = False
                     await self._report_assistant_response_ended()
         elif content.role == Role.USER:

src/pipecat/services/ultravox/llm.py

@@ -604,6 +604,18 @@ class UltravoxRealtimeLLMService(LLMService):
                     case "state":
                         if self._bot_responding and data.get("state") != "speaking":
                             await self._handle_response_end()
+                    case "playback_clear_buffer":
+                        # Server signals that the user interrupted the bot
+                        # mid-speech and any buffered output audio should be
+                        # dropped. Broadcast InterruptionFrame so the assistant
+                        # aggregator records the message interrupted=True
+                        # (upstream) and BaseOutputTransport clears its audio
+                        # buffer (downstream). The subsequent "state" message
+                        # transitioning off "speaking" is what closes the
+                        # response via _handle_response_end; firing the
+                        # interruption first ensures the aggregator handles
+                        # InterruptionFrame before LLMFullResponseEndFrame.
+                        await self.broadcast_interruption()
                     case "client_tool_invocation":
                         await self._handle_tool_invocation(
                             data.get("toolName"), data.get("invocationId"), data.get("parameters")