From 804e3ea9ecc9d8b94222cd4427256e501c94656f Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Mon, 13 Apr 2026 18:11:32 -0400 Subject: [PATCH] Trigger turn stop immediately when transcript arrives after p99 timeout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When the STT p99 timeout fires without a transcript, the turn stop strategy previously did nothing — falling through to the 5-second user_turn_stop_timeout. Now, a _timeout_expired flag tracks when the timeout has elapsed so that a late transcript triggers the turn stop immediately instead of waiting for the fallback. --- .../speech_timeout_user_turn_stop_strategy.py | 9 +++++ .../turn_analyzer_user_turn_stop_strategy.py | 11 ++++++ tests/test_user_turn_stop_strategy.py | 35 +++++++++++++++++++ 3 files changed, 55 insertions(+) diff --git a/src/pipecat/turns/user_stop/speech_timeout_user_turn_stop_strategy.py b/src/pipecat/turns/user_stop/speech_timeout_user_turn_stop_strategy.py index db6ef3c00..0a8feeb2e 100644 --- a/src/pipecat/turns/user_stop/speech_timeout_user_turn_stop_strategy.py +++ b/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): diff --git a/src/pipecat/turns/user_stop/turn_analyzer_user_turn_stop_strategy.py b/src/pipecat/turns/user_stop/turn_analyzer_user_turn_stop_strategy.py index 2931965aa..adc987583 100644 --- a/src/pipecat/turns/user_stop/turn_analyzer_user_turn_stop_strategy.py +++ b/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): diff --git a/tests/test_user_turn_stop_strategy.py b/tests/test_user_turn_stop_strategy.py index 96876fec5..ecaf5317f 100644 --- a/tests/test_user_turn_stop_strategy.py +++ b/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.