Merge pull request #4283 from pipecat-ai/mb/user-stop-transcript-improvements

changelog/4283.fixed.md

@@ -0,0 +1 @@
+- Fixed delayed turn completion when STT transcripts arrive after the p99 timeout. Previously, a late transcript (beyond the p99 window) would fall through to the 5-second `user_turn_stop_timeout` fallback. Now the turn stop triggers immediately when the late transcript arrives.

src/pipecat/turns/user_stop/speech_timeout_user_turn_stop_strategy.py

@@ -61,6 +61,7 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy):
         self._transcript_finalized = False
         self._vad_stopped_time: Optional[float] = None
         self._timeout_task: Optional[asyncio.Task] = None
+        self._timeout_expired: bool = False
 
     async def reset(self):
         """Reset the strategy to its initial state."""
@@ -69,6 +70,7 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy):
         self._vad_user_speaking = False
         self._transcript_finalized = False
         self._vad_stopped_time = None
+        self._timeout_expired = False
         if self._timeout_task:
             await self.task_manager.cancel_task(self._timeout_task)
             self._timeout_task = None
@@ -117,6 +119,7 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy):
         self._vad_user_speaking = True
         self._transcript_finalized = False
         self._vad_stopped_time = None
+        self._timeout_expired = False
         # Cancel any pending timeout
         if self._timeout_task:
             await self.task_manager.cancel_task(self._timeout_task)
@@ -163,6 +166,11 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy):
             self._transcript_finalized = True
             # For finalized transcripts, check if we can trigger early
             await self._maybe_trigger_user_turn_stopped()
+        elif self._timeout_expired:
+            # The p99 timeout already elapsed without a transcript. Now that
+            # we have one, trigger the turn stop immediately.
+            await self.trigger_user_turn_stopped()
+            return
 
         # Fallback: handle transcripts when no VAD stop was received.
         # This handles edge cases where transcripts arrive without VAD firing.
@@ -207,6 +215,7 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy):
         finally:
             self._timeout_task = None
 
+        self._timeout_expired = True
         await self._maybe_trigger_user_turn_stopped()
 
     async def _maybe_trigger_user_turn_stopped(self):

src/pipecat/turns/user_stop/turn_analyzer_user_turn_stop_strategy.py

@@ -65,6 +65,7 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy):
         self._vad_stopped_time: Optional[float] = None  # Track when VAD stopped was received
         self._transcript_finalized = False
         self._timeout_task: Optional[asyncio.Task] = None
+        self._timeout_expired: bool = False
 
     async def reset(self):
         """Reset the strategy to its initial state."""
@@ -74,6 +75,7 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy):
         self._vad_user_speaking = False
         self._vad_stopped_time = None
         self._transcript_finalized = False
+        self._timeout_expired = False
         if self._timeout_task:
             await self.task_manager.cancel_task(self._timeout_task)
             self._timeout_task = None
@@ -149,6 +151,7 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy):
         self._vad_user_speaking = True
         self._vad_stopped_time = None
         self._transcript_finalized = False
+        self._timeout_expired = False
         # Cancel any pending timeout
         if self._timeout_task:
             await self.task_manager.cancel_task(self._timeout_task)
@@ -204,6 +207,13 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy):
             self._transcript_finalized = True
             # For finalized transcripts, trigger immediately if turn is complete
             await self._maybe_trigger_user_turn_stopped()
+        elif self._timeout_expired and self._turn_complete:
+            # The p99 timeout already elapsed without a transcript. Now that
+            # we have one, trigger the turn stop immediately. This handles the
+            # case where the transcript is slower to arrive than the p99 timeout,
+            # trigger the user turn to stop immediately.
+            await self.trigger_user_turn_stopped()
+            return
 
         # Fallback: handle transcripts when no VAD stop was received.
         # This handles edge cases where transcripts arrive without VAD firing.
@@ -240,6 +250,7 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy):
         finally:
             self._timeout_task = None
 
+        self._timeout_expired = True
         await self._maybe_trigger_user_turn_stopped()
 
     async def _maybe_trigger_user_turn_stopped(self):

tests/test_user_turn_stop_strategy.py

@@ -494,6 +494,41 @@ class TestSpeechTimeoutUserTurnStopStrategy(unittest.IsolatedAsyncioTestCase):
         # Finalized transcript received after timeout, triggers immediately
         self.assertTrue(should_start)
 
+    async def test_sie_delay_t(self):
+        """Non-finalized transcript arriving after timeout triggers immediately."""
+        strategy = await self._create_strategy()
+
+        should_start = None
+
+        @strategy.event_handler("on_user_turn_stopped")
+        async def on_user_turn_stopped(strategy, params):
+            nonlocal should_start
+            should_start = True
+
+        # S
+        await strategy.process_frame(VADUserStartedSpeakingFrame())
+        self.assertIsNone(should_start)
+
+        # I
+        await strategy.process_frame(
+            InterimTranscriptionFrame(text="Hello!", user_id="cat", timestamp="")
+        )
+
+        # E
+        await strategy.process_frame(VADUserStoppedSpeakingFrame())
+        self.assertIsNone(should_start)
+
+        # Delay - timeout expires but no transcript yet
+        await asyncio.sleep(AGGREGATION_TIMEOUT + 0.1)
+        # Still no trigger because no finalized transcript received
+        self.assertIsNone(should_start)
+
+        # T (non-finalized) - triggers immediately since timeout already elapsed
+        await strategy.process_frame(TranscriptionFrame(text="Hello!", user_id="cat", timestamp=""))
+
+        # Non-finalized transcript received after timeout, triggers immediately
+        self.assertTrue(should_start)
+
     async def test_reset_clears_stale_text_no_premature_stop(self):
         """Test that reset() clears stale text and cancels timeout, preventing premature stop.