wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fixing TTSService deadlock.

Browse Source
This commit is contained in:
filipi87
2026-05-06 13:32:26 -03:00
parent 95db08646c
commit a23baf9de6
2 changed files with 155 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/pipecat/services/tts_service.py
View File

@@ -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:

146
tests/test_tts_frame_ordering.py
View File

@@ -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()