From f3993f177506cd952930e077ce5eb5a1b725bb54 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 14 Jan 2026 20:01:50 -0500 Subject: [PATCH] fix to make on_user_turn_stop_timeout work with ExternalUserTurnStrategies --- changelog/3454.fixed.md | 1 + src/pipecat/turns/user_turn_controller.py | 28 +++++++++--- tests/test_user_turn_controller.py | 54 ++++++++++++++++++++++- 3 files changed, 77 insertions(+), 6 deletions(-) create mode 100644 changelog/3454.fixed.md diff --git a/changelog/3454.fixed.md b/changelog/3454.fixed.md new file mode 100644 index 000000000..0269370d7 --- /dev/null +++ b/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`. diff --git a/src/pipecat/turns/user_turn_controller.py b/src/pipecat/turns/user_turn_controller.py index a9b267ae0..56b6ad711 100644 --- a/src/pipecat/turns/user_turn_controller.py +++ b/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) diff --git a/tests/test_user_turn_controller.py b/tests/test_user_turn_controller.py index 6762bd219..67c1a7860 100644 --- a/tests/test_user_turn_controller.py +++ b/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)