Merge pull request #3454 from pipecat-ai/mb/external-turn-strategies-timeout

fix to make on_user_turn_stop_timeout work with ExternalUserTurnStrat…

changelog/3454.fixed.md

@@ -0,0 +1 @@
+- Fixed an issue where `on_user_turn_stop_timeout` could fire while a user is talking when using `ExternalUserTurnStrategies`.

src/pipecat/turns/user_turn_controller.py

@@ -12,6 +12,8 @@ from typing import Optional, Type
 from pipecat.frames.frames import (
     Frame,
     TranscriptionFrame,
+    UserStartedSpeakingFrame,
+    UserStoppedSpeakingFrame,
     VADUserStartedSpeakingFrame,
     VADUserStoppedSpeakingFrame,
 )
@@ -80,7 +82,7 @@ class UserTurnController(BaseObject):
 
         self._task_manager: Optional[BaseTaskManager] = None
 
-        self._vad_user_speaking = False
+        self._user_speaking = False
 
         self._user_turn = False
         self._user_turn_stop_timeout_event = asyncio.Event()
@@ -146,7 +148,11 @@ class UserTurnController(BaseObject):
             frame: The frame to be processed.
 
         """
-        if isinstance(frame, VADUserStartedSpeakingFrame):
+        if isinstance(frame, UserStartedSpeakingFrame):
+            await self._handle_user_started_speaking(frame)
+        elif isinstance(frame, UserStoppedSpeakingFrame):
+            await self._handle_user_stopped_speaking(frame)
+        elif isinstance(frame, VADUserStartedSpeakingFrame):
             await self._handle_vad_user_started_speaking(frame)
         elif isinstance(frame, VADUserStoppedSpeakingFrame):
             await self._handle_vad_user_stopped_speaking(frame)
@@ -179,14 +185,26 @@ class UserTurnController(BaseObject):
         for s in self._user_turn_strategies.stop or []:
             await s.cleanup()
 
+    async def _handle_user_started_speaking(self, frame: UserStartedSpeakingFrame):
+        self._user_speaking = True
+
+        # The user started talking, let's reset the user turn timeout.
+        self._user_turn_stop_timeout_event.set()
+
+    async def _handle_user_stopped_speaking(self, frame: UserStoppedSpeakingFrame):
+        self._user_speaking = False
+
+        # The user stopped talking, let's reset the user turn timeout.
+        self._user_turn_stop_timeout_event.set()
+
     async def _handle_vad_user_started_speaking(self, frame: VADUserStartedSpeakingFrame):
-        self._vad_user_speaking = True
+        self._user_speaking = True
 
         # The user started talking, let's reset the user turn timeout.
         self._user_turn_stop_timeout_event.set()
 
     async def _handle_vad_user_stopped_speaking(self, frame: VADUserStoppedSpeakingFrame):
-        self._vad_user_speaking = False
+        self._user_speaking = False
 
         # The user stopped talking, let's reset the user turn timeout.
         self._user_turn_stop_timeout_event.set()
@@ -260,7 +278,7 @@ class UserTurnController(BaseObject):
                 )
                 self._user_turn_stop_timeout_event.clear()
             except asyncio.TimeoutError:
-                if self._user_turn and not self._vad_user_speaking:
+                if self._user_turn and not self._user_speaking:
                     await self._call_event_handler("on_user_turn_stop_timeout")
                     await self._trigger_user_turn_stop(
                         None, UserTurnStoppedParams(enable_user_speaking_frames=True)

tests/test_user_turn_controller.py

@@ -9,11 +9,13 @@ import unittest
 
 from pipecat.frames.frames import (
     TranscriptionFrame,
+    UserStartedSpeakingFrame,
+    UserStoppedSpeakingFrame,
     VADUserStartedSpeakingFrame,
     VADUserStoppedSpeakingFrame,
 )
 from pipecat.turns.user_turn_controller import UserTurnController
-from pipecat.turns.user_turn_strategies import UserTurnStrategies
+from pipecat.turns.user_turn_strategies import ExternalUserTurnStrategies, UserTurnStrategies
 from pipecat.utils.asyncio.task_manager import TaskManager, TaskManagerParams
 
 USER_TURN_STOP_TIMEOUT = 0.2
@@ -96,3 +98,53 @@ class TestUserTurnController(unittest.IsolatedAsyncioTestCase):
         self.assertTrue(should_start)
         self.assertTrue(should_stop)
         self.assertTrue(timeout)
+
+    async def test_external_user_turn_strategies_no_timeout_while_speaking(self):
+        """Test that timeout does not trigger when user is still speaking with external strategies."""
+        controller = UserTurnController(
+            user_turn_strategies=ExternalUserTurnStrategies(),
+            user_turn_stop_timeout=USER_TURN_STOP_TIMEOUT,
+        )
+
+        await controller.setup(self.task_manager)
+
+        should_start = None
+        should_stop = None
+        timeout = None
+
+        @controller.event_handler("on_user_turn_started")
+        async def on_user_turn_started(controller, strategy, params):
+            nonlocal should_start
+            should_start = True
+
+        @controller.event_handler("on_user_turn_stopped")
+        async def on_user_turn_stopped(controller, strategy, params):
+            nonlocal should_stop
+            should_stop = True
+
+        @controller.event_handler("on_user_turn_stop_timeout")
+        async def on_user_turn_stop_timeout(controller):
+            nonlocal timeout
+            timeout = True
+
+        # Simulate external service (like Deepgram Flux) broadcasting UserStartedSpeakingFrame
+        await controller.process_frame(UserStartedSpeakingFrame())
+        self.assertTrue(should_start)
+        self.assertFalse(should_stop)
+        self.assertFalse(timeout)
+
+        # User is still speaking, timeout should not trigger
+        await asyncio.sleep(USER_TURN_STOP_TIMEOUT + 0.1)
+        self.assertTrue(should_start)
+        self.assertFalse(should_stop)
+        self.assertFalse(timeout)
+
+        # Now external service broadcasts UserStoppedSpeakingFrame
+        await controller.process_frame(UserStoppedSpeakingFrame())
+
+        # But no transcription, so timeout should trigger
+        await asyncio.sleep(USER_TURN_STOP_TIMEOUT + 0.1)
+
+        self.assertTrue(should_start)
+        self.assertTrue(should_stop)
+        self.assertTrue(timeout)