diff --git a/changelog/4007.fixed.2.md b/changelog/4007.fixed.2.md new file mode 100644 index 000000000..0c50b83e9 --- /dev/null +++ b/changelog/4007.fixed.2.md @@ -0,0 +1 @@ +- Fixed `TTSService` potentially canceling in-flight audio during shutdown. The stop sequence now waits for all queued audio contexts to finish processing before canceling the stop frame task. diff --git a/changelog/4007.fixed.md b/changelog/4007.fixed.md new file mode 100644 index 000000000..8a90aea43 --- /dev/null +++ b/changelog/4007.fixed.md @@ -0,0 +1 @@ +- Fixed `ParallelPipeline` dropping or misordering frames during lifecycle synchronization. Buffered frames are now flushed in the correct order relative to synchronization frames (`StartFrame` goes first, `EndFrame`/`CancelFrame` go after), and frames added to the buffer during flush are also drained. diff --git a/src/pipecat/pipeline/parallel_pipeline.py b/src/pipecat/pipeline/parallel_pipeline.py index 88ea04638..1e2e03a8f 100644 --- a/src/pipecat/pipeline/parallel_pipeline.py +++ b/src/pipecat/pipeline/parallel_pipeline.py @@ -143,6 +143,19 @@ class ParallelPipeline(BasePipeline): await super().process_frame(frame, direction) # Parallel pipeline synchronized frames. + # + # - StartFrame: If a fast branch completes first, processors in + # other branches that haven't received StartFrame yet could + # receive other frames before it, causing errors. + # + # - EndFrame: If EndFrame escapes from a fast branch, downstream + # processors (e.g. output transport) begin shutting down while + # other branches still have frames to flush, causing lost output. + # + # - CancelFrame: PipelineTask waits for CancelFrame to reach the + # pipeline sink. If it escapes from a fast branch while slower + # branches are still running, the task considers cancellation + # complete prematurely. if isinstance(frame, (StartFrame, EndFrame, CancelFrame)): self._frame_counter[frame.id] = len(self._pipelines) self._synchronizing = True @@ -179,8 +192,13 @@ class ParallelPipeline(BasePipeline): # Only push the frame when all pipelines have processed it. if frame_counter == 0: self._synchronizing = False - await self._parallel_push_frame(frame, direction) - await self._flush_buffered_frames() + # StartFrame should always go before any other frame. + if isinstance(frame, StartFrame): + await self._parallel_push_frame(frame, direction) + await self._flush_buffered_frames() + else: + await self._flush_buffered_frames() + await self._parallel_push_frame(frame, direction) await self.resume_processing_system_frames() await self.resume_processing_frames() else: @@ -188,7 +206,6 @@ class ParallelPipeline(BasePipeline): async def _flush_buffered_frames(self): """Flush frames that were buffered during lifecycle frame synchronization.""" - frames = self._buffered_frames - self._buffered_frames = [] - for frame, direction in frames: + while len(self._buffered_frames) > 0: + frame, direction = self._buffered_frames.pop(0) await self.push_frame(frame, direction) diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index a79156018..44fb98826 100644 --- a/src/pipecat/services/tts_service.py +++ b/src/pipecat/services/tts_service.py @@ -536,15 +536,15 @@ class TTSService(AIService): frame: The end frame. """ await super().stop(frame) - if self._stop_frame_task: - await self.cancel_task(self._stop_frame_task) - self._stop_frame_task = None if self._audio_context_task: # Indicate no more audio contexts are available; this will end the # task cleanly after all contexts have been processed. await self._contexts_queue.put(None) await self._audio_context_task self._audio_context_task = None + if self._stop_frame_task: + await self.cancel_task(self._stop_frame_task) + self._stop_frame_task = None async def cancel(self, frame: CancelFrame): """Cancel the TTS service.