Accept inbound RTVI speaking signals as turn frames

changelog/5715.added.md

@@ -0,0 +1 @@
+- RTVI speaking messages (`user-started-speaking` / `user-stopped-speaking` and `vad-user-*`) are now accepted client-to-server, so a cooperating client can drive turn boundaries without a server-side VAD analyzer.

src/pipecat/processors/frameworks/rtvi/models.py

@@ -538,14 +538,25 @@ class UserLLMTextMessage(BaseModel):
 
 
 class UserStartedSpeakingMessage(BaseModel):
-    """Message indicating user has started speaking."""
+    """Message indicating the user has started speaking.
+
+    Bidirectional. Server-to-client: the bot's turn strategy finalized a
+    start. Client-to-server: a cooperating client that ran VAD at the
+    microphone is proposing a turn start
+    (:class:`~pipecat.frames.frames.ProposedUserStartedSpeakingFrame`).
+    """
 
     label: MessageLiteral = MESSAGE_LABEL
     type: Literal["user-started-speaking"] = "user-started-speaking"
 
 
 class UserStoppedSpeakingMessage(BaseModel):
-    """Message indicating user has stopped speaking."""
+    """Message indicating the user has stopped speaking.
+
+    Bidirectional. Server-to-client: the bot's turn strategy finalized a
+    stop. Client-to-server: a cooperating client is proposing a turn stop
+    (:class:`~pipecat.frames.frames.ProposedUserStoppedSpeakingFrame`).
+    """
 
     label: MessageLiteral = MESSAGE_LABEL
     type: Literal["user-stopped-speaking"] = "user-stopped-speaking"
@@ -554,8 +565,10 @@ class UserStoppedSpeakingMessage(BaseModel):
 class VADUserStartedSpeakingMessage(BaseModel):
     """Message indicating VAD detected the user started speaking.
 
-    Raw VAD signal, emitted independently of turn finalization (unlike
+    Bidirectional raw VAD signal, independent of turn finalization (unlike
     ``user-started-speaking``, which a turn strategy may gate or defer).
+    Client-to-server, this becomes a
+    :class:`~pipecat.frames.frames.VADUserStartedSpeakingFrame`.
     """
 
     label: MessageLiteral = MESSAGE_LABEL
@@ -565,8 +578,10 @@ class VADUserStartedSpeakingMessage(BaseModel):
 class VADUserStoppedSpeakingMessage(BaseModel):
     """Message indicating VAD detected the user stopped speaking.
 
-    Raw VAD signal, emitted independently of turn finalization (unlike
+    Bidirectional raw VAD signal, independent of turn finalization (unlike
     ``user-stopped-speaking``, which a turn strategy may gate or defer).
+    Client-to-server, this becomes a
+    :class:`~pipecat.frames.frames.VADUserStoppedSpeakingFrame`.
     """
 
     label: MessageLiteral = MESSAGE_LABEL

src/pipecat/processors/frameworks/rtvi/processor.py

@@ -30,8 +30,12 @@
     LLMConfigureOutputFrame,
     LLMMessagesAppendFrame,
     OutputTransportMessageUrgentFrame,
+    ProposedUserStartedSpeakingFrame,
+    ProposedUserStoppedSpeakingFrame,
     StartFrame,
     SystemFrame,
+    VADUserStartedSpeakingFrame,
+    VADUserStoppedSpeakingFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.processors.frameworks.rtvi.frames import (
@@ -369,6 +373,13 @@ async def _handle_message(self, message: RTVI.Message):
                 case "dtmf":
                     data = RTVI.DTMFInputData.model_validate(message.data)
                     await self._handle_dtmf(data)
+                case (
+                    "user-started-speaking"
+                    | "user-stopped-speaking"
+                    | "vad-user-started-speaking"
+                    | "vad-user-stopped-speaking"
+                ):
+                    await self._handle_inbound_speaking(message.type)
 
                 case _:
                     await self._send_error_response(message.id, f"Unsupported type {message.type}")
@@ -453,6 +464,27 @@ async def _handle_audio_buffer(self, data):
             # Handle missing keys, decoding errors, and invalid types
             logger.error(f"Error processing audio buffer: {e}")
 
+    async def _handle_inbound_speaking(self, message_type: str):
+        """Turn a client-originated speaking RTVI message into pipeline frames.
+
+        These types were previously server-to-client notifications only. A
+        client that runs VAD at the microphone can now send them inbound so a
+        bot with no local Silero analyzer can still take turns.
+
+        ``user-started-speaking`` / ``user-stopped-speaking`` become proposals
+        so :class:`~pipecat.turns.user_start.ExternalUserTurnStartStrategy`
+        owns the interruption. ``vad-user-*`` become VAD frames so the default
+        VAD start strategy can drive turns without a server-side analyzer.
+        """
+        if message_type == "user-started-speaking":
+            await self.push_frame(ProposedUserStartedSpeakingFrame())
+        elif message_type == "user-stopped-speaking":
+            await self.push_frame(ProposedUserStoppedSpeakingFrame())
+        elif message_type == "vad-user-started-speaking":
+            await self.push_frame(VADUserStartedSpeakingFrame())
+        elif message_type == "vad-user-stopped-speaking":
+            await self.push_frame(VADUserStoppedSpeakingFrame())
+
     async def _handle_dtmf(self, data: RTVI.DTMFInputData):
         """Handle DTMF keypresses from the client.
 

tests/test_rtvi_processor.py

@@ -16,6 +16,10 @@
     InputAudioRawFrame,
     InputDTMFFrame,
     InputTransportStartAudioStreamingFrame,
+    ProposedUserStartedSpeakingFrame,
+    ProposedUserStoppedSpeakingFrame,
+    VADUserStartedSpeakingFrame,
+    VADUserStoppedSpeakingFrame,
 )
 from pipecat.processors.frameworks.rtvi.processor import RTVIProcessor
 
@@ -226,5 +230,52 @@ def test_dtmf_input_data_rejects_legacy_button_field(self):
             RTVI.DTMFInputData.model_validate({"button": "1"})
 
 
+class TestRTVIInboundSpeaking(unittest.IsolatedAsyncioTestCase):
+    async def asyncTearDown(self):
+        if hasattr(self, "processor"):
+            await self.processor.cleanup()
+
+    async def _pushed_from(self, message_type: str):
+        self.processor = RTVIProcessor()
+        self.processor.push_frame = AsyncMock()
+        await self.processor._handle_inbound_speaking(message_type)
+        return [c.args[0] for c in self.processor.push_frame.call_args_list]
+
+    async def test_user_started_speaking_is_a_turn_proposal(self):
+        # A proposal, not UserStartedSpeakingFrame: the aggregator's external
+        # start strategy owns the interruption instead of adopting a
+        # already-announced turn (which would skip barge-in).
+        pushed = await self._pushed_from("user-started-speaking")
+        self.assertEqual(len(pushed), 1)
+        self.assertIsInstance(pushed[0], ProposedUserStartedSpeakingFrame)
+
+    async def test_user_stopped_speaking_is_a_turn_proposal(self):
+        pushed = await self._pushed_from("user-stopped-speaking")
+        self.assertEqual(len(pushed), 1)
+        self.assertIsInstance(pushed[0], ProposedUserStoppedSpeakingFrame)
+
+    async def test_vad_user_started_speaking_is_a_vad_frame(self):
+        pushed = await self._pushed_from("vad-user-started-speaking")
+        self.assertEqual(len(pushed), 1)
+        self.assertIsInstance(pushed[0], VADUserStartedSpeakingFrame)
+
+    async def test_vad_user_stopped_speaking_is_a_vad_frame(self):
+        pushed = await self._pushed_from("vad-user-stopped-speaking")
+        self.assertEqual(len(pushed), 1)
+        self.assertIsInstance(pushed[0], VADUserStoppedSpeakingFrame)
+
+    async def test_handle_message_routes_inbound_speaking_types(self):
+        self.processor = RTVIProcessor()
+        self.processor.push_frame = AsyncMock()
+        self.processor._send_error_response = AsyncMock()
+        await self.processor._handle_message(
+            RTVI.Message(label="rtvi-ai", type="user-started-speaking", id="1")
+        )
+        pushed = [c.args[0] for c in self.processor.push_frame.call_args_list]
+        self.assertEqual(len(pushed), 1)
+        self.assertIsInstance(pushed[0], ProposedUserStartedSpeakingFrame)
+        self.processor._send_error_response.assert_not_called()
+
+
 if __name__ == "__main__":
     unittest.main()