diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index a77a0f837..48102d0b2 100644 --- a/src/pipecat/services/tts_service.py +++ b/src/pipecat/services/tts_service.py @@ -883,6 +883,16 @@ class TTSService(AIService): self._turn_context_id = None self._word_last_pts = 0 self._create_audio_context_task() + # When pause_frame_processing=True, the process task may be blocked at + # __process_event.wait() because pause_processing_frames() was called + # after LLMFullResponseEndFrame and an UninterruptibleFrame was dequeued + # before the interrupt arrived. _start_interruption() in the base class + # handles the common case (non-uninterruptible frames) by cancelling and + # recreating the process task. But when _start_interruption() detects an + # UninterruptibleFrame it only resets the queue, leaving the process task + # blocked. BotStoppedSpeakingFrame never arrives (no audio played), so we + # must resume here to prevent a permanent deadlock. + await self._maybe_resume_frame_processing() async def _maybe_pause_frame_processing(self): if self._processing_text and self._pause_frame_processing: diff --git a/tests/test_tts_frame_ordering.py b/tests/test_tts_frame_ordering.py index 70d2a0323..9fb3a895c 100644 --- a/tests/test_tts_frame_ordering.py +++ b/tests/test_tts_frame_ordering.py @@ -20,6 +20,8 @@ repeated for each TTSSpeakFrame, with no cross-group contamination. Also covers LLM response flow with push_text_frames=True (non-word-timestamp TTS): verifies TTSTextFrame ordering relative to LLMFullResponseEndFrame. + +Also covers the interruption-during-pause deadlock scenario (see test_no_deadlock_on_interrupt_*). """ import asyncio @@ -31,8 +33,10 @@ import pytest from pipecat.frames.frames import ( AggregatedTextFrame, + ControlFrame, DataFrame, Frame, + InterruptionFrame, LLMAssistantPushAggregationFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, @@ -42,9 +46,10 @@ from pipecat.frames.frames import ( TTSStartedFrame, TTSStoppedFrame, TTSTextFrame, + UninterruptibleFrame, ) from pipecat.services.tts_service import TTSService -from pipecat.tests.utils import run_test +from pipecat.tests.utils import SleepFrame, run_test # --------------------------------------------------------------------------- # Test-only frame @@ -61,6 +66,18 @@ class FooFrame(DataFrame): label: str = "" +@dataclass +class UninterruptibleMarkerFrame(ControlFrame, UninterruptibleFrame): + """Test-only uninterruptible marker frame used to trigger the deadlock code path. + + When this is in the process queue with __should_block_frames=True, and an + InterruptionFrame arrives, _start_interruption() takes the non-cancel path + (because of the UninterruptibleFrame) leaving __should_block_frames=True. + """ + + label: str = "" + + # --------------------------------------------------------------------------- # Mock TTS services # --------------------------------------------------------------------------- @@ -209,6 +226,34 @@ class MockWebSocketPauseTTSService(TTSService): yield +class MockWebSocketPauseTTSServiceNoAudio(TTSService): + """Simulates a WebSocket TTS service with pause but no audio delivery. + + Used to test the interruption-during-pause deadlock. Audio is never + delivered within the test window, so BotStoppedSpeakingFrame is never + sent by the transport, and on_audio_context_completed is never called. + Without the fix, an interruption arriving while the process task is + blocked behind an UninterruptibleFrame causes a permanent deadlock. + """ + + def __init__(self, **kwargs): + super().__init__( + push_start_frame=True, + push_text_frames=False, + pause_frame_processing=True, + sample_rate=_SAMPLE_RATE, + **kwargs, + ) + + def can_generate_metrics(self) -> bool: + return False + + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: + # Intentionally never deliver audio — simulates extreme TTS latency. + if False: + yield + + class _MockWordTimestampHttpTTSService(TTSService): """HTTP-style TTS: yields audio synchronously, calls add_word_timestamps first. @@ -708,5 +753,104 @@ async def test_push_aggregation_no_pts_without_word_timestamps(): assert push_frames[0].pts is None or push_frames[0].pts == 0 +@pytest.mark.asyncio +async def test_no_deadlock_on_interrupt_before_audio_simple(): + """Interrupting before any TTS audio arrives must not deadlock. + + This simpler scenario (no UninterruptibleFrame in the queue at interrupt + time) is handled by _start_interruption() in the base class: it cancels and + recreates the process task, resetting __should_block_frames to False. + + Timeline: + 1. LLM response → _processing_text=True. + 2. LLMFullResponseEndFrame → pause_processing_frames() called. + 3. (No audio from TTS yet; BotStoppedSpeakingFrame never sent.) + 4. InterruptionFrame → _start_interruption() cancels + recreates process + task → __should_block_frames=False. + 5. FooFrame must arrive downstream within the timeout. + """ + tts = MockWebSocketPauseTTSServiceNoAudio() + + frames_to_send = [ + LLMFullResponseStartFrame(), + TextFrame(text="Hello."), + LLMFullResponseEndFrame(), + SleepFrame(sleep=0.1), # window: after pause set, before audio + InterruptionFrame(), + SleepFrame(sleep=0.1), + FooFrame(label="after_interrupt"), + ] + + frames_received = await asyncio.wait_for( + run_test(tts, frames_to_send=frames_to_send), + timeout=3.0, + ) + + down = frames_received[0] + foo_frames = [f for f in down if isinstance(f, FooFrame)] + assert any(f.label == "after_interrupt" for f in foo_frames), ( + "FooFrame after interruption was not received — possible deadlock" + ) + + +@pytest.mark.asyncio +async def test_no_deadlock_on_interrupt_before_audio_with_uninterruptible(): + """Interrupting during pause with an UninterruptibleFrame queued must not deadlock. + + This is the harder scenario that requires the fix in _handle_interruption(). + + Without the fix the pipeline deadlocks permanently: + - pause_processing_frames() blocks __should_block_frames=True. + - The process task dequeues UninterruptibleMarkerFrame and blocks at + __process_event.wait(). + - InterruptionFrame arrives → _start_interruption() sees the + UninterruptibleFrame and takes the reset-queue-only path, leaving + __should_block_frames=True and the process task blocked. + - BotStoppedSpeakingFrame is never sent (no audio played). + - resume_processing_frames() is never called → deadlock. + + With the fix (_maybe_resume_frame_processing() inside _handle_interruption()), + the event is set and the process task unblocks, allowing subsequent frames + (FooFrame, EndFrame) to be processed. + + Timeline: + 1. LLM response → _processing_text=True. + 2. LLMFullResponseEndFrame → pause_processing_frames(). + 3. UninterruptibleMarkerFrame enters process queue. + 4. Process task picks up UninterruptibleMarkerFrame, blocks at wait(). + 5. InterruptionFrame → _start_interruption() keeps the task running + (uninterruptible) but WITHOUT the fix __should_block_frames stays True. + 6. With the fix: _handle_interruption() calls _maybe_resume_frame_processing() + → __process_event.set() → process task unblocked. + 7. FooFrame arrives downstream. + """ + tts = MockWebSocketPauseTTSServiceNoAudio() + + frames_to_send = [ + LLMFullResponseStartFrame(), + TextFrame(text="Hello."), + LLMFullResponseEndFrame(), + # Queue right after: process task will pick this up after setting pause, + # then block at __process_event.wait() because __should_block_frames=True. + UninterruptibleMarkerFrame(label="uninterruptible"), + SleepFrame(sleep=0.1), # let process task dequeue and block on the frame + InterruptionFrame(), + SleepFrame(sleep=0.1), # let interruption handling complete + FooFrame(label="after_interrupt"), + ] + + frames_received = await asyncio.wait_for( + run_test(tts, frames_to_send=frames_to_send), + timeout=3.0, + ) + + down = frames_received[0] + foo_frames = [f for f in down if isinstance(f, FooFrame)] + assert any(f.label == "after_interrupt" for f in foo_frames), ( + "FooFrame after interruption was not received — pipeline deadlocked " + "(missing _maybe_resume_frame_processing() in _handle_interruption)" + ) + + if __name__ == "__main__": unittest.main()