From 73a56f5d81b84bbc5442c60cffe6ffb500d4ff80 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 12 Mar 2026 00:15:01 -0700 Subject: [PATCH] Fix ParallelPipeline flush ordering and buffered frame handling Flush buffered frames before pushing the synchronization frame so downstream processors see the buffered frames first. Switch to a while-loop with pop(0) so frames added to the buffer during flush are also drained. --- src/pipecat/pipeline/parallel_pipeline.py | 27 ++++++++++++++++++----- 1 file changed, 22 insertions(+), 5 deletions(-) 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)