Merge pull request #4007 from pipecat-ai/aleix/fix-parallel-pipeline-flush-and-tts-stop-order

Fix ParallelPipeline flush ordering and TTS stop sequence

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.

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.

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)
+                # StartFrame should always go before any other frame.
-                await self._flush_buffered_frames()
+                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
+        while len(self._buffered_frames) > 0:
-        self._buffered_frames = []
+            frame, direction = self._buffered_frames.pop(0)
-        for frame, direction in frames:
             await self.push_frame(frame, direction)

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.