From 0be40846834d8937f04712ca7a2348d800fdb90d Mon Sep 17 00:00:00 2001
From: Paul Kompfner <paul@daily.co>
Date: Fri, 13 Mar 2026 12:29:32 -0400
Subject: [PATCH 01/12] Fix bug resulting in `SyncParallelPipeline` breaking
 the Whisker debugger

---
 src/pipecat/pipeline/sync_parallel_pipeline.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/pipecat/pipeline/sync_parallel_pipeline.py b/src/pipecat/pipeline/sync_parallel_pipeline.py
index cb3f1bbe0..48d66d3b8 100644
--- a/src/pipecat/pipeline/sync_parallel_pipeline.py
+++ b/src/pipecat/pipeline/sync_parallel_pipeline.py
@@ -184,7 +184,7 @@ class SyncParallelPipeline(BasePipeline):
         Returns:
             The list of entry processors.
         """
-        return self._sources
+        return [s["processor"] for s in self._sources]
 
     def processors_with_metrics(self) -> List[FrameProcessor]:
         """Collect processors that can generate metrics from all parallel pipelines.

From 463db59bb5b1cda95cc697b0b2aabd2d70240571 Mon Sep 17 00:00:00 2001
From: Paul Kompfner <paul@daily.co>
Date: Fri, 13 Mar 2026 13:40:26 -0400
Subject: [PATCH 02/12] Minor comment typo fix

---
 src/pipecat/pipeline/sync_parallel_pipeline.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/pipecat/pipeline/sync_parallel_pipeline.py b/src/pipecat/pipeline/sync_parallel_pipeline.py
index 48d66d3b8..3597b0e9e 100644
--- a/src/pipecat/pipeline/sync_parallel_pipeline.py
+++ b/src/pipecat/pipeline/sync_parallel_pipeline.py
@@ -225,7 +225,7 @@ class SyncParallelPipeline(BasePipeline):
         # this element won't work. Since, we know it should be synchronous we
         # push a SyncFrame. Since frames are ordered we know this frame will be
         # pushed after the synchronous processor has pushed its data allowing us
-        # to synchrnonize all the internal pipelines by waiting for the
+        # to synchronize all the internal pipelines by waiting for the
         # SyncFrame in all of them.
         async def wait_for_sync(
             obj, main_queue: asyncio.Queue, frame: Frame, direction: FrameDirection

From 1ede8460a2d775ea9580f53af0f7a772b39c7df6 Mon Sep 17 00:00:00 2001
From: Paul Kompfner <paul@daily.co>
Date: Fri, 13 Mar 2026 14:17:35 -0400
Subject: [PATCH 03/12] Fix SyncParallelPipeline race condition with concurrent
 SystemFrame processing
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The FrameProcessor two-queue architecture processes SystemFrames and
non-SystemFrames on separate concurrent async tasks. Both paths called
SyncParallelPipeline.process_frame(), which used the same per-pipeline
sink queues. A SystemFrame's wait_for_sync could steal frames from a
concurrent non-SystemFrame's wait_for_sync, corrupting synchronization
and stalling the pipeline.

This was triggered by the auto-embedded RTVI processor (added in
v0.0.101) which floods OutputTransportMessageUrgentFrame SystemFrames
through the pipeline during LLM responses.

Fix: SystemFrames (except EndFrame) now take a fast path — passed
through internal pipelines and pushed downstream directly without
touching the sink queues or drain logic. EndFrame retains the full
drain behavior as a lifecycle frame.
---
 .../pipeline/sync_parallel_pipeline.py        | 20 ++++++++++++++++---
 1 file changed, 17 insertions(+), 3 deletions(-)

diff --git a/src/pipecat/pipeline/sync_parallel_pipeline.py b/src/pipecat/pipeline/sync_parallel_pipeline.py
index 3597b0e9e..00af35bbf 100644
--- a/src/pipecat/pipeline/sync_parallel_pipeline.py
+++ b/src/pipecat/pipeline/sync_parallel_pipeline.py
@@ -221,6 +221,20 @@ class SyncParallelPipeline(BasePipeline):
         """
         await super().process_frame(frame, direction)
 
+        # SystemFrames (but not EndFrame) are simply passed through all
+        # internal pipelines without draining queued output. This avoids
+        # the race condition where a SystemFrame's wait_for_sync steals
+        # frames from a concurrent non-SystemFrame's wait_for_sync.
+        if isinstance(frame, SystemFrame) and not isinstance(frame, EndFrame):
+            if direction == FrameDirection.UPSTREAM:
+                for s in self._sinks:
+                    await s["processor"].process_frame(frame, direction)
+            elif direction == FrameDirection.DOWNSTREAM:
+                for s in self._sources:
+                    await s["processor"].process_frame(frame, direction)
+            await self.push_frame(frame, direction)
+            return
+
         # The last processor of each pipeline needs to be synchronous otherwise
         # this element won't work. Since, we know it should be synchronous we
         # push a SyncFrame. Since frames are ordered we know this frame will be
@@ -235,12 +249,12 @@ class SyncParallelPipeline(BasePipeline):
 
             await processor.process_frame(frame, direction)
 
-            if isinstance(frame, (SystemFrame, EndFrame)):
+            if isinstance(frame, EndFrame):
                 new_frame = await queue.get()
-                if isinstance(new_frame, (SystemFrame, EndFrame)):
+                if isinstance(new_frame, EndFrame):
                     await main_queue.put(new_frame)
                 else:
-                    while not isinstance(new_frame, (SystemFrame, EndFrame)):
+                    while not isinstance(new_frame, EndFrame):
                         await main_queue.put(new_frame)
                         queue.task_done()
                         new_frame = await queue.get()

From 0f1ff16af12e26e9cf05e4c6b2add398ff6f96b5 Mon Sep 17 00:00:00 2001
From: Paul Kompfner <paul@daily.co>
Date: Fri, 13 Mar 2026 15:15:45 -0400
Subject: [PATCH 04/12] Add sync_with_audio support for OutputImageRawFrame

Add a `sync_with_audio` field to `OutputImageRawFrame` that routes image
frames through the audio queue in the output transport, ensuring images
are only displayed after all preceding audio has been sent. This enables
proper audio/image synchronization in pipelines like the calendar month
narration example.

Update the 05-sync-speech-and-image example to use an `ImageAudioSync`
processor that sets this flag on image frames.
---
 .../foundational/05-sync-speech-and-image.py   | 18 +++++++++++++++++-
 src/pipecat/frames/frames.py                   |  8 ++++++++
 src/pipecat/transports/base_output.py          |  6 +++++-
 3 files changed, 30 insertions(+), 2 deletions(-)

diff --git a/examples/foundational/05-sync-speech-and-image.py b/examples/foundational/05-sync-speech-and-image.py
index 1e4cf34a4..35f8f411c 100644
--- a/examples/foundational/05-sync-speech-and-image.py
+++ b/examples/foundational/05-sync-speech-and-image.py
@@ -16,6 +16,7 @@ from pipecat.frames.frames import (
     Frame,
     LLMContextFrame,
     LLMFullResponseStartFrame,
+    OutputImageRawFrame,
     TextFrame,
 )
 from pipecat.pipeline.pipeline import Pipeline
@@ -44,6 +45,18 @@ class MonthFrame(DataFrame):
         return f"{self.name}(month: {self.month})"
 
 
+class ImageAudioSync(FrameProcessor):
+    """Marks output image frames to be synchronized with audio playback."""
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, OutputImageRawFrame):
+            frame.sync_with_audio = True
+
+        await self.push_frame(frame, direction)
+
+
 class MonthPrepender(FrameProcessor):
     def __init__(self):
         super().__init__()
@@ -129,7 +142,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
                 sentence_aggregator,  # Aggregates LLM output into full sentences
                 SyncParallelPipeline(  # Run pipelines in parallel aggregating the result
                     [month_prepender, tts],  # Create "Month: sentence" and output audio
-                    [imagegen],  # Generate image
+                    [
+                        imagegen,  # Generate image
+                        ImageAudioSync(),  # Mark image as needing sync output w/audio
+                    ],
                 ),
                 transport.output(),  # Transport output
             ]
diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py
index fbc01c488..7107cfd97 100644
--- a/src/pipecat/frames/frames.py
+++ b/src/pipecat/frames/frames.py
@@ -274,8 +274,16 @@ class OutputImageRawFrame(DataFrame, ImageRawFrame):
     An image that will be shown by the transport. If the transport supports
     multiple video destinations (e.g. multiple video tracks) the destination
     name can be specified in transport_destination.
+
+    Parameters:
+        sync_with_audio: If True, the image is queued with audio frames so
+            it is only displayed after all preceding audio has been sent.
+            Defaults to False (image is displayed immediately when the output
+            transport receives it).
     """
 
+    sync_with_audio: bool = field(default=False, init=False)
+
     def __str__(self):
         pts = format_pts(self.pts)
         return f"{self.name}(pts: {pts}, destination: {self.transport_destination}, size: {self.size}, format: {self.format})"
diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py
index e14ae3828..01af97be8 100644
--- a/src/pipecat/transports/base_output.py
+++ b/src/pipecat/transports/base_output.py
@@ -569,7 +569,11 @@ class BaseOutputTransport(FrameProcessor):
             if not self._params.video_out_enabled:
                 return
 
-            if self._params.video_out_is_live and isinstance(frame, OutputImageRawFrame):
+            if isinstance(frame, OutputImageRawFrame) and frame.sync_with_audio:
+                # Route through the audio queue so the image is only
+                # displayed after all preceding audio has been sent.
+                await self._audio_queue.put(frame)
+            elif self._params.video_out_is_live and isinstance(frame, OutputImageRawFrame):
                 await self._video_queue.put(frame)
             elif isinstance(frame, OutputImageRawFrame):
                 await self._set_video_image(frame)

From c3d6e965d80f10f491d0cf62bec7a57663c79fd5 Mon Sep 17 00:00:00 2001
From: Paul Kompfner <paul@daily.co>
Date: Fri, 13 Mar 2026 18:43:54 -0400
Subject: [PATCH 05/12] Use TextAggregationMode.TOKEN in the
 05-sync-speech-and-image example since the SentenceAggregator already
 provides complete sentences.

---
 examples/foundational/05-sync-speech-and-image.py | 5 +++++
 src/pipecat/processors/frame_processor.py         | 4 ++--
 2 files changed, 7 insertions(+), 2 deletions(-)

diff --git a/examples/foundational/05-sync-speech-and-image.py b/examples/foundational/05-sync-speech-and-image.py
index 35f8f411c..b896261ae 100644
--- a/examples/foundational/05-sync-speech-and-image.py
+++ b/examples/foundational/05-sync-speech-and-image.py
@@ -31,6 +31,7 @@ from pipecat.runner.utils import create_transport
 from pipecat.services.cartesia.tts import CartesiaHttpTTSService
 from pipecat.services.fal.image import FalImageGenService
 from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.services.tts_service import TextAggregationMode
 from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.daily.transport import DailyParams
 
@@ -114,6 +115,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
             settings=CartesiaHttpTTSService.Settings(
                 voice="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
             ),
+            # No need to aggregate by sentences (the default), as we already know we're getting full sentences
+            # (Otherwise the service will unnecessarily wait for follow-up input to confirm the sentence is complete,
+            #  which, sadly, actually breaks the synchronization mechanism)
+            text_aggregation_mode=TextAggregationMode.TOKEN,
         )
 
         imagegen = FalImageGenService(
diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py
index f3d9fbdea..a74ee4f81 100644
--- a/src/pipecat/processors/frame_processor.py
+++ b/src/pipecat/processors/frame_processor.py
@@ -633,7 +633,7 @@ class FrameProcessor(BaseObject):
 
     async def pause_processing_frames(self):
         """Pause processing of queued frames."""
-        logger.trace(f"{self}: pausing frame processing")
+        logger.debug(f"{self}: pausing frame processing")
         self.__should_block_frames = True
         if self.__process_event:
             self.__process_event.clear()
@@ -647,7 +647,7 @@ class FrameProcessor(BaseObject):
 
     async def resume_processing_frames(self):
         """Resume processing of queued frames."""
-        logger.trace(f"{self}: resuming frame processing")
+        logger.debug(f"{self}: resuming frame processing")
         if self.__process_event:
             self.__process_event.set()
 

From ba779f920fcd2f7a3f8f48218299911ed782785d Mon Sep 17 00:00:00 2001
From: Paul Kompfner <paul@daily.co>
Date: Fri, 13 Mar 2026 21:04:16 -0400
Subject: [PATCH 06/12] Revert a couple of logs that were changed from `trace`
 to `debug` just for debugging

---
 src/pipecat/processors/frame_processor.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py
index a74ee4f81..f3d9fbdea 100644
--- a/src/pipecat/processors/frame_processor.py
+++ b/src/pipecat/processors/frame_processor.py
@@ -633,7 +633,7 @@ class FrameProcessor(BaseObject):
 
     async def pause_processing_frames(self):
         """Pause processing of queued frames."""
-        logger.debug(f"{self}: pausing frame processing")
+        logger.trace(f"{self}: pausing frame processing")
         self.__should_block_frames = True
         if self.__process_event:
             self.__process_event.clear()
@@ -647,7 +647,7 @@ class FrameProcessor(BaseObject):
 
     async def resume_processing_frames(self):
         """Resume processing of queued frames."""
-        logger.debug(f"{self}: resuming frame processing")
+        logger.trace(f"{self}: resuming frame processing")
         if self.__process_event:
             self.__process_event.set()
 

From 5e7639812aee582ab9f1435cfdfa5b25ba318706 Mon Sep 17 00:00:00 2001
From: Paul Kompfner <paul@daily.co>
Date: Fri, 13 Mar 2026 22:09:44 -0400
Subject: [PATCH 07/12] Add ImageBeforeAudioReorderer to sync-speech-and-image
 example

Add a processor after SyncParallelPipeline that ensures each image frame
precedes its corresponding TTS audio frames. SyncParallelPipeline batches
them together but doesn't guarantee branch ordering. The reorderer detects
when TTS frames arrive before their image (via context_id tracking) and
holds them until the image arrives.

Also rename ImageAudioSync to MarkImageForPlaybackSync for clarity.
---
 .../foundational/05-sync-speech-and-image.py  | 65 ++++++++++++++++++-
 1 file changed, 63 insertions(+), 2 deletions(-)

diff --git a/examples/foundational/05-sync-speech-and-image.py b/examples/foundational/05-sync-speech-and-image.py
index b896261ae..646235e7c 100644
--- a/examples/foundational/05-sync-speech-and-image.py
+++ b/examples/foundational/05-sync-speech-and-image.py
@@ -12,12 +12,17 @@ from dotenv import load_dotenv
 from loguru import logger
 
 from pipecat.frames.frames import (
+    AggregatedTextFrame,
     DataFrame,
     Frame,
     LLMContextFrame,
     LLMFullResponseStartFrame,
     OutputImageRawFrame,
     TextFrame,
+    TTSAudioRawFrame,
+    TTSStartedFrame,
+    TTSStoppedFrame,
+    TTSTextFrame,
 )
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
@@ -46,7 +51,7 @@ class MonthFrame(DataFrame):
         return f"{self.name}(month: {self.month})"
 
 
-class ImageAudioSync(FrameProcessor):
+class MarkImageForPlaybackSync(FrameProcessor):
     """Marks output image frames to be synchronized with audio playback."""
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
@@ -58,6 +63,61 @@ class ImageAudioSync(FrameProcessor):
         await self.push_frame(frame, direction)
 
 
+class ImageBeforeAudioReorderer(FrameProcessor):
+    """Ensures each image frame precedes its corresponding TTS audio frames.
+
+    SyncParallelPipeline guarantees that each image is in the same synchronized
+    batch as its audio, but doesn't guarantee which branch's output comes first.
+    This processor detects when TTS frames arrive before their image and holds
+    them until the image arrives.
+
+    All frames pass through immediately unless we detect an ordering problem:
+    TTS frames arrived without a preceding image for the current batch (identified
+    by context_id). In that case, the TTS frames are held until the next image
+    frame, which is pushed first.
+    """
+
+    def __init__(self):
+        super().__init__()
+        self._held_tts_frames = []
+        self._seen_image = False
+        self._current_context_id = None
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, OutputImageRawFrame):
+            self._seen_image = True
+            if self._held_tts_frames:
+                # Image arrived after TTS frames — push image first, then release held frames.
+                logger.debug("ImageBeforeAudioReorderer: reordered — moved image before audio")
+                await self.push_frame(frame, direction)
+                for f in self._held_tts_frames:
+                    await self.push_frame(f, direction)
+                self._held_tts_frames = []
+            else:
+                logger.debug(
+                    "ImageBeforeAudioReorderer: no reorder needed — image was already first"
+                )
+                await self.push_frame(frame, direction)
+        elif isinstance(
+            frame,
+            (AggregatedTextFrame, TTSStartedFrame, TTSAudioRawFrame, TTSStoppedFrame, TTSTextFrame),
+        ):
+            # A new context_id means a new batch — reset image tracking.
+            context_id = frame.context_id
+            if context_id and context_id != self._current_context_id:
+                self._current_context_id = context_id
+                self._seen_image = False
+
+            if self._seen_image:
+                await self.push_frame(frame, direction)
+            else:
+                self._held_tts_frames.append(frame)
+        else:
+            await self.push_frame(frame, direction)
+
+
 class MonthPrepender(FrameProcessor):
     def __init__(self):
         super().__init__()
@@ -149,9 +209,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
                     [month_prepender, tts],  # Create "Month: sentence" and output audio
                     [
                         imagegen,  # Generate image
-                        ImageAudioSync(),  # Mark image as needing sync output w/audio
+                        MarkImageForPlaybackSync(),  # Mark image as needing sync w/audio during playback
                     ],
                 ),
+                ImageBeforeAudioReorderer(),  # Ensure each image precedes its audio (important for playback)
                 transport.output(),  # Transport output
             ]
         )

From 61ff53f2b968c8acc435c8492e2027fa8035ab0e Mon Sep 17 00:00:00 2001
From: Paul Kompfner <paul@daily.co>
Date: Fri, 13 Mar 2026 22:15:32 -0400
Subject: [PATCH 08/12] Add changelog entries for PR #4029

---
 changelog/4029.added.md   | 1 +
 changelog/4029.fixed.2.md | 1 +
 changelog/4029.fixed.md   | 1 +
 3 files changed, 3 insertions(+)
 create mode 100644 changelog/4029.added.md
 create mode 100644 changelog/4029.fixed.2.md
 create mode 100644 changelog/4029.fixed.md

diff --git a/changelog/4029.added.md b/changelog/4029.added.md
new file mode 100644
index 000000000..ba3714483
--- /dev/null
+++ b/changelog/4029.added.md
@@ -0,0 +1 @@
+- Added `sync_with_audio` field to `OutputImageRawFrame`. When set to `True`, the output transport queues image frames with audio so they are displayed only after all preceding audio has been sent, enabling synchronized audio/image playback.
diff --git a/changelog/4029.fixed.2.md b/changelog/4029.fixed.2.md
new file mode 100644
index 000000000..faf54592e
--- /dev/null
+++ b/changelog/4029.fixed.2.md
@@ -0,0 +1 @@
+- Fixed TTS frame ordering when an audio context is active. Pass-through frames are now routed through the audio context queue so they stay properly ordered with TTS audio frames. Previously, a frame that preceded some incoming text to a TTS service could "jump the queue", getting ahead of the outgoing audio.
diff --git a/changelog/4029.fixed.md b/changelog/4029.fixed.md
new file mode 100644
index 000000000..57930a997
--- /dev/null
+++ b/changelog/4029.fixed.md
@@ -0,0 +1 @@
+- Fixed `SyncParallelPipeline` race condition where concurrent SystemFrame processing (e.g. from RTVI) could corrupt sink queues and cause deadlocks. SystemFrames now take a fast path that passes them through without draining queued output.

From 26fc238eb7e558ee414e3faf00bb0b3456b3d546 Mon Sep 17 00:00:00 2001
From: Paul Kompfner <paul@daily.co>
Date: Fri, 13 Mar 2026 22:20:53 -0400
Subject: [PATCH 09/12] Add changelog entry for Whisker debugger fix

---
 changelog/4029.fixed.3.md | 1 +
 1 file changed, 1 insertion(+)
 create mode 100644 changelog/4029.fixed.3.md

diff --git a/changelog/4029.fixed.3.md b/changelog/4029.fixed.3.md
new file mode 100644
index 000000000..3c812d590
--- /dev/null
+++ b/changelog/4029.fixed.3.md
@@ -0,0 +1 @@
+- Fixed `SyncParallelPipeline` breaking the Whisker debugger.

From d702ebd6a29519af2ea50f5a6b8d208901342139 Mon Sep 17 00:00:00 2001
From: Paul Kompfner <paul@daily.co>
Date: Mon, 16 Mar 2026 09:59:13 -0400
Subject: [PATCH 10/12] Add frame_order parameter to SyncParallelPipeline
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Adds a FrameOrder enum with ARRIVAL (default, existing behavior) and
PIPELINE (pushes frames in pipeline definition order). This lets callers
guarantee output ordering between parallel pipelines — e.g. ensuring
image frames precede audio frames — without needing a separate reordering
processor downstream.

Updates the 05-sync-speech-and-image example to use FrameOrder.PIPELINE,
removing the ImageBeforeAudioReorderer class entirely.
---
 changelog/4029.added.2.md                     |   1 +
 .../foundational/05-sync-speech-and-image.py  |  73 ++---------
 .../pipeline/sync_parallel_pipeline.py        | 111 +++++++++++++----
 tests/test_sync_parallel_pipeline.py          | 117 ++++++++++++++++++
 4 files changed, 214 insertions(+), 88 deletions(-)
 create mode 100644 changelog/4029.added.2.md
 create mode 100644 tests/test_sync_parallel_pipeline.py

diff --git a/changelog/4029.added.2.md b/changelog/4029.added.2.md
new file mode 100644
index 000000000..1ae691442
--- /dev/null
+++ b/changelog/4029.added.2.md
@@ -0,0 +1 @@
+- Added `frame_order` parameter to `SyncParallelPipeline`. Set `frame_order=FrameOrder.PIPELINE` to push synchronized output frames in pipeline definition order (all frames from the first pipeline, then the second, etc.) instead of the default arrival order.
diff --git a/examples/foundational/05-sync-speech-and-image.py b/examples/foundational/05-sync-speech-and-image.py
index 646235e7c..f0e2ff9c7 100644
--- a/examples/foundational/05-sync-speech-and-image.py
+++ b/examples/foundational/05-sync-speech-and-image.py
@@ -12,21 +12,16 @@ from dotenv import load_dotenv
 from loguru import logger
 
 from pipecat.frames.frames import (
-    AggregatedTextFrame,
     DataFrame,
     Frame,
     LLMContextFrame,
     LLMFullResponseStartFrame,
     OutputImageRawFrame,
     TextFrame,
-    TTSAudioRawFrame,
-    TTSStartedFrame,
-    TTSStoppedFrame,
-    TTSTextFrame,
 )
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.sync_parallel_pipeline import SyncParallelPipeline
+from pipecat.pipeline.sync_parallel_pipeline import FrameOrder, SyncParallelPipeline
 from pipecat.pipeline.task import PipelineTask
 from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.sentence import SentenceAggregator
@@ -63,61 +58,6 @@ class MarkImageForPlaybackSync(FrameProcessor):
         await self.push_frame(frame, direction)
 
 
-class ImageBeforeAudioReorderer(FrameProcessor):
-    """Ensures each image frame precedes its corresponding TTS audio frames.
-
-    SyncParallelPipeline guarantees that each image is in the same synchronized
-    batch as its audio, but doesn't guarantee which branch's output comes first.
-    This processor detects when TTS frames arrive before their image and holds
-    them until the image arrives.
-
-    All frames pass through immediately unless we detect an ordering problem:
-    TTS frames arrived without a preceding image for the current batch (identified
-    by context_id). In that case, the TTS frames are held until the next image
-    frame, which is pushed first.
-    """
-
-    def __init__(self):
-        super().__init__()
-        self._held_tts_frames = []
-        self._seen_image = False
-        self._current_context_id = None
-
-    async def process_frame(self, frame: Frame, direction: FrameDirection):
-        await super().process_frame(frame, direction)
-
-        if isinstance(frame, OutputImageRawFrame):
-            self._seen_image = True
-            if self._held_tts_frames:
-                # Image arrived after TTS frames — push image first, then release held frames.
-                logger.debug("ImageBeforeAudioReorderer: reordered — moved image before audio")
-                await self.push_frame(frame, direction)
-                for f in self._held_tts_frames:
-                    await self.push_frame(f, direction)
-                self._held_tts_frames = []
-            else:
-                logger.debug(
-                    "ImageBeforeAudioReorderer: no reorder needed — image was already first"
-                )
-                await self.push_frame(frame, direction)
-        elif isinstance(
-            frame,
-            (AggregatedTextFrame, TTSStartedFrame, TTSAudioRawFrame, TTSStoppedFrame, TTSTextFrame),
-        ):
-            # A new context_id means a new batch — reset image tracking.
-            context_id = frame.context_id
-            if context_id and context_id != self._current_context_id:
-                self._current_context_id = context_id
-                self._seen_image = False
-
-            if self._seen_image:
-                await self.push_frame(frame, direction)
-            else:
-                self._held_tts_frames.append(frame)
-        else:
-            await self.push_frame(frame, direction)
-
-
 class MonthPrepender(FrameProcessor):
     def __init__(self):
         super().__init__()
@@ -197,22 +137,27 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         # that, each pipeline runs concurrently and `SyncParallelPipeline` will
         # wait for the input frame to be processed.
         #
+        # We use `FrameOrder.PIPELINE` so that each synchronized batch of output
+        # frames is pushed in the order the pipelines are listed: image first,
+        # then audio. This ensures the transport receives the image before the
+        # audio frames it should accompany.
+        #
         # Note that `SyncParallelPipeline` requires the last processor in each
         # of the pipelines to be synchronous. In this case, we use
-        # `CartesiaHttpTTSService` and `FalImageGenService` which make HTTP
+        # `FalImageGenService` and `CartesiaHttpTTSService` which make HTTP
         # requests and wait for the response.
         pipeline = Pipeline(
             [
                 llm,  # LLM
                 sentence_aggregator,  # Aggregates LLM output into full sentences
                 SyncParallelPipeline(  # Run pipelines in parallel aggregating the result
-                    [month_prepender, tts],  # Create "Month: sentence" and output audio
                     [
                         imagegen,  # Generate image
                         MarkImageForPlaybackSync(),  # Mark image as needing sync w/audio during playback
                     ],
+                    [month_prepender, tts],  # Create "Month: sentence" and output audio
+                    frame_order=FrameOrder.PIPELINE,
                 ),
-                ImageBeforeAudioReorderer(),  # Ensure each image precedes its audio (important for playback)
                 transport.output(),  # Transport output
             ]
         )
diff --git a/src/pipecat/pipeline/sync_parallel_pipeline.py b/src/pipecat/pipeline/sync_parallel_pipeline.py
index 00af35bbf..9870f03f3 100644
--- a/src/pipecat/pipeline/sync_parallel_pipeline.py
+++ b/src/pipecat/pipeline/sync_parallel_pipeline.py
@@ -13,6 +13,7 @@ and prevent duplicate processing.
 
 import asyncio
 from dataclasses import dataclass
+from enum import Enum
 from itertools import chain
 from typing import List
 
@@ -24,6 +25,25 @@ from pipecat.pipeline.pipeline import Pipeline
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
 
 
+class FrameOrder(Enum):
+    """Controls the order in which synchronized frames are pushed downstream.
+
+    When multiple parallel pipelines produce output for the same input frame,
+    this setting determines the order in which those output frames are pushed.
+
+    Attributes:
+        ARRIVAL: Frames are pushed in the order they arrive from any pipeline.
+            This is the default and matches the behavior of prior versions.
+        PIPELINE: Frames are pushed in pipeline definition order — all frames
+            from the first pipeline are pushed, then all frames from the second
+            pipeline, and so on. Useful when the relative ordering between
+            pipelines matters (e.g. ensuring image frames precede audio frames).
+    """
+
+    ARRIVAL = "arrival"
+    PIPELINE = "pipeline"
+
+
 @dataclass
 class SyncFrame(ControlFrame):
     """Control frame used to synchronize parallel pipeline processing.
@@ -109,20 +129,30 @@ class SyncParallelPipeline(BasePipeline):
 
     The pipeline uses SyncFrame control frames to coordinate between parallel paths
     and ensure all paths have completed processing before moving to the next frame.
+
+    By default, output frames are pushed in the order they arrive from any pipeline
+    (``FrameOrder.ARRIVAL``). Set ``frame_order=FrameOrder.PIPELINE`` to push frames
+    in pipeline definition order instead — all output from the first pipeline, then
+    the second, and so on.
     """
 
-    def __init__(self, *args):
+    def __init__(self, *args, frame_order: FrameOrder = FrameOrder.ARRIVAL):
         """Initialize the synchronous parallel pipeline.
 
         Args:
-            *args: Variable number of processor lists, each representing a parallel pipeline path.
-                   Each argument should be a list of FrameProcessor instances.
+            *args: Variable number of processor lists, each representing a parallel
+                pipeline path. Each argument should be a list of FrameProcessor instances.
+            frame_order: Controls the order in which synchronized output frames are
+                pushed. ``FrameOrder.ARRIVAL`` (default) pushes frames in the order they arrive.
+                ``FrameOrder.PIPELINE`` pushes all frames from the first pipeline
+                before the second, and so on.
 
         Raises:
             Exception: If no arguments are provided.
             TypeError: If any argument is not a list of processors.
         """
         super().__init__()
+        self._frame_order = frame_order
 
         if len(args) == 0:
             raise Exception(f"SyncParallelPipeline needs at least one argument")
@@ -215,6 +245,11 @@ class SyncParallelPipeline(BasePipeline):
         to maintain proper ordering and prevent duplicate processing. Uses SyncFrame
         control frames to coordinate between parallel paths.
 
+        When ``frame_order`` is ``FrameOrder.ARRIVAL``, output frames are pushed in
+        the order they arrive from any pipeline (via a shared queue). When it is
+        ``FrameOrder.PIPELINE``, each pipeline collects its output into a separate
+        list and the lists are drained in pipeline definition order.
+
         Args:
             frame: The frame to process.
             direction: The direction of frame flow.
@@ -235,60 +270,88 @@ class SyncParallelPipeline(BasePipeline):
             await self.push_frame(frame, direction)
             return
 
+        use_pipeline_order = self._frame_order == FrameOrder.PIPELINE
+
         # The last processor of each pipeline needs to be synchronous otherwise
-        # this element won't work. Since, we know it should be synchronous we
+        # this element won't work. Since we know it should be synchronous we
         # push a SyncFrame. Since frames are ordered we know this frame will be
         # pushed after the synchronous processor has pushed its data allowing us
         # to synchronize all the internal pipelines by waiting for the
         # SyncFrame in all of them.
+        #
+        # In ARRIVAL mode, output frames are put onto a shared main_queue as
+        # they arrive. In PIPELINE mode, they are accumulated in a per-pipeline
+        # list and returned so the caller can drain them in definition order.
         async def wait_for_sync(
             obj, main_queue: asyncio.Queue, frame: Frame, direction: FrameDirection
-        ):
+        ) -> list[Frame]:
             processor = obj["processor"]
             queue = obj["queue"]
+            output_frames: list[Frame] = []
 
             await processor.process_frame(frame, direction)
 
             if isinstance(frame, EndFrame):
                 new_frame = await queue.get()
                 if isinstance(new_frame, EndFrame):
-                    await main_queue.put(new_frame)
+                    if use_pipeline_order:
+                        output_frames.append(new_frame)
+                    else:
+                        await main_queue.put(new_frame)
                 else:
                     while not isinstance(new_frame, EndFrame):
-                        await main_queue.put(new_frame)
+                        if use_pipeline_order:
+                            output_frames.append(new_frame)
+                        else:
+                            await main_queue.put(new_frame)
                         queue.task_done()
                         new_frame = await queue.get()
             else:
                 await processor.process_frame(SyncFrame(), direction)
                 new_frame = await queue.get()
                 while not isinstance(new_frame, SyncFrame):
-                    await main_queue.put(new_frame)
+                    if use_pipeline_order:
+                        output_frames.append(new_frame)
+                    else:
+                        await main_queue.put(new_frame)
                     queue.task_done()
                     new_frame = await queue.get()
 
+            return output_frames
+
         if direction == FrameDirection.UPSTREAM:
             # If we get an upstream frame we process it in each sink.
-            await asyncio.gather(
+            frames_per_pipeline = await asyncio.gather(
                 *[wait_for_sync(s, self._up_queue, frame, direction) for s in self._sinks]
             )
         elif direction == FrameDirection.DOWNSTREAM:
             # If we get a downstream frame we process it in each source.
-            await asyncio.gather(
+            frames_per_pipeline = await asyncio.gather(
                 *[wait_for_sync(s, self._down_queue, frame, direction) for s in self._sources]
             )
 
-        seen_ids = set()
-        while not self._up_queue.empty():
-            frame = await self._up_queue.get()
-            if frame.id not in seen_ids:
-                await self.push_frame(frame, FrameDirection.UPSTREAM)
-                seen_ids.add(frame.id)
-            self._up_queue.task_done()
+        if use_pipeline_order:
+            # Push frames in pipeline definition order, deduplicating by id.
+            seen_ids = set()
+            for pipeline_frames in frames_per_pipeline:
+                for f in pipeline_frames:
+                    if f.id not in seen_ids:
+                        await self.push_frame(f, direction)
+                        seen_ids.add(f.id)
+        else:
+            # ARRIVAL mode: drain the shared queues in the order frames arrived.
+            seen_ids = set()
+            while not self._up_queue.empty():
+                frame = await self._up_queue.get()
+                if frame.id not in seen_ids:
+                    await self.push_frame(frame, FrameDirection.UPSTREAM)
+                    seen_ids.add(frame.id)
+                self._up_queue.task_done()
 
-        seen_ids = set()
-        while not self._down_queue.empty():
-            frame = await self._down_queue.get()
-            if frame.id not in seen_ids:
-                await self.push_frame(frame, FrameDirection.DOWNSTREAM)
-                seen_ids.add(frame.id)
-            self._down_queue.task_done()
+            seen_ids = set()
+            while not self._down_queue.empty():
+                frame = await self._down_queue.get()
+                if frame.id not in seen_ids:
+                    await self.push_frame(frame, FrameDirection.DOWNSTREAM)
+                    seen_ids.add(frame.id)
+                self._down_queue.task_done()
diff --git a/tests/test_sync_parallel_pipeline.py b/tests/test_sync_parallel_pipeline.py
new file mode 100644
index 000000000..6c6faf7c7
--- /dev/null
+++ b/tests/test_sync_parallel_pipeline.py
@@ -0,0 +1,117 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import unittest
+from dataclasses import dataclass
+
+from pipecat.frames.frames import Frame, TextFrame
+from pipecat.pipeline.sync_parallel_pipeline import FrameOrder, SyncParallelPipeline
+from pipecat.processors.filters.identity_filter import IdentityFilter
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.tests.utils import run_test
+
+
+@dataclass
+class TaggedFrame(Frame):
+    """A simple tagged frame for testing pipeline ordering."""
+
+    tag: str = ""
+
+    def __str__(self):
+        return f"{self.name}(tag: {self.tag})"
+
+
+class EmitTaggedFrameProcessor(FrameProcessor):
+    """Emits a TaggedFrame for every TextFrame it receives.
+
+    Used to produce distinguishable output from different pipelines so tests
+    can verify ordering.
+    """
+
+    def __init__(self, tag: str, *, delay: float = 0, **kwargs):
+        super().__init__(**kwargs)
+        self._tag = tag
+        self._delay = delay
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, TextFrame):
+            if self._delay > 0:
+                await asyncio.sleep(self._delay)
+            await self.push_frame(TaggedFrame(tag=self._tag))
+        else:
+            await self.push_frame(frame, direction)
+
+
+class TestSyncParallelPipeline(unittest.IsolatedAsyncioTestCase):
+    async def test_dedup_multiple_frames(self):
+        """Identical frames from multiple paths should be deduplicated."""
+        pipeline = SyncParallelPipeline([IdentityFilter()], [IdentityFilter()])
+
+        frames_to_send = [TextFrame(text="one"), TextFrame(text="two")]
+        expected_down_frames = [TextFrame, TextFrame]
+        await run_test(
+            pipeline,
+            frames_to_send=frames_to_send,
+            expected_down_frames=expected_down_frames,
+        )
+
+    async def test_arrival_order(self):
+        """With FrameOrder.ARRIVAL, a slow first pipeline's frames should
+        arrive after a fast second pipeline's frames."""
+        pipeline = SyncParallelPipeline(
+            [EmitTaggedFrameProcessor("slow", delay=0.05)],
+            [EmitTaggedFrameProcessor("fast")],
+            frame_order=FrameOrder.ARRIVAL,
+        )
+
+        frames_to_send = [TextFrame(text="one"), TextFrame(text="two")]
+        (down_frames, _) = await run_test(
+            pipeline,
+            frames_to_send=frames_to_send,
+        )
+
+        tags = [f.tag for f in down_frames if isinstance(f, TaggedFrame)]
+        assert tags == [
+            "fast",
+            "slow",
+            "fast",
+            "slow",
+        ], f"Expected fast before slow in each batch, got {tags}"
+
+    async def test_pipeline_order(self):
+        """With FrameOrder.PIPELINE and multiple input frames, each batch
+        should follow pipeline definition order regardless of processing speed."""
+        pipeline = SyncParallelPipeline(
+            [EmitTaggedFrameProcessor("slow", delay=0.05)],
+            [EmitTaggedFrameProcessor("fast")],
+            frame_order=FrameOrder.PIPELINE,
+        )
+
+        frames_to_send = [TextFrame(text="one"), TextFrame(text="two")]
+        (down_frames, _) = await run_test(
+            pipeline,
+            frames_to_send=frames_to_send,
+        )
+
+        tags = [f.tag for f in down_frames if isinstance(f, TaggedFrame)]
+        assert tags == [
+            "slow",
+            "fast",
+            "slow",
+            "fast",
+        ], f"Expected pipeline definition order (slow, fast) in each batch, got {tags}"
+
+    async def test_default_is_arrival(self):
+        """The default frame_order should be ARRIVAL."""
+        pipeline = SyncParallelPipeline([IdentityFilter()])
+        assert pipeline._frame_order == FrameOrder.ARRIVAL
+
+
+if __name__ == "__main__":
+    unittest.main()

From 06f7da44f19b8fa949b8335273df3d3a970aabd8 Mon Sep 17 00:00:00 2001
From: Paul Kompfner <paul@daily.co>
Date: Wed, 18 Mar 2026 16:39:20 -0400
Subject: [PATCH 11/12] Clarify SyncParallelPipeline docstrings

Rewrite docstrings to more clearly explain what SyncParallelPipeline
does: hold all output until every parallel branch finishes, so frames
produced in response to a single input are released together.
---
 .../pipeline/sync_parallel_pipeline.py        | 65 ++++++++++---------
 1 file changed, 36 insertions(+), 29 deletions(-)

diff --git a/src/pipecat/pipeline/sync_parallel_pipeline.py b/src/pipecat/pipeline/sync_parallel_pipeline.py
index 9870f03f3..148d29b25 100644
--- a/src/pipecat/pipeline/sync_parallel_pipeline.py
+++ b/src/pipecat/pipeline/sync_parallel_pipeline.py
@@ -4,11 +4,16 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-"""Synchronous parallel pipeline implementation for concurrent frame processing.
+"""Synchronized parallel pipeline that holds output until all branches finish.
 
-This module provides a pipeline that processes frames through multiple parallel
-pipelines simultaneously, synchronizing their output to maintain frame ordering
-and prevent duplicate processing.
+A SyncParallelPipeline fans each inbound frame out to multiple parallel pipelines
+and waits for every pipeline to finish processing before releasing any of the
+resulting output frames. This ensures that all frames produced in response to a
+single input frame are emitted together.
+
+System frames (except EndFrame) are exempt from this synchronization — they pass
+straight through without waiting, since they are expected to race ahead of
+regular data frames.
 """
 
 import asyncio
@@ -46,20 +51,21 @@ class FrameOrder(Enum):
 
 @dataclass
 class SyncFrame(ControlFrame):
-    """Control frame used to synchronize parallel pipeline processing.
+    """Sentinel frame used to detect when a parallel pipeline has finished processing.
 
-    This frame is sent through parallel pipelines to determine when the
-    internal pipelines have finished processing a batch of frames.
+    After sending a real frame into a parallel pipeline, a SyncFrame is sent
+    behind it. When the SyncFrame emerges from the pipeline's output, we know
+    all output frames for the preceding input have been produced.
     """
 
     pass
 
 
 class SyncParallelPipelineSource(FrameProcessor):
-    """Source processor for synchronous parallel pipeline processing.
+    """Bookend processor placed at the start of each parallel pipeline.
 
-    Routes frames to parallel pipelines and collects upstream responses
-    for synchronization purposes.
+    Forwards downstream frames into the pipeline and captures upstream frames
+    into a queue so the parent SyncParallelPipeline can release them later.
     """
 
     def __init__(self, upstream_queue: asyncio.Queue):
@@ -88,10 +94,11 @@ class SyncParallelPipelineSource(FrameProcessor):
 
 
 class SyncParallelPipelineSink(FrameProcessor):
-    """Sink processor for synchronous parallel pipeline processing.
+    """Bookend processor placed at the end of each parallel pipeline.
 
-    Collects downstream frames from parallel pipelines and routes
-    upstream frames back through the pipeline.
+    Captures downstream output frames into a queue so the parent
+    SyncParallelPipeline can release them later, and forwards upstream
+    frames back through the pipeline.
     """
 
     def __init__(self, downstream_queue: asyncio.Queue):
@@ -120,15 +127,20 @@ class SyncParallelPipelineSink(FrameProcessor):
 
 
 class SyncParallelPipeline(BasePipeline):
-    """Pipeline that processes frames through multiple parallel pipelines synchronously.
+    """Fans each input frame to parallel pipelines then holds output until every pipeline finishes.
 
-    Creates multiple parallel processing paths that all receive the same input frames
-    and produces synchronized output. Each parallel path is a separate pipeline that
-    processes frames independently, with synchronization points to ensure consistent
-    ordering and prevent duplicate frame processing.
+    For each inbound frame the pipeline:
 
-    The pipeline uses SyncFrame control frames to coordinate between parallel paths
-    and ensure all paths have completed processing before moving to the next frame.
+    1. Sends the frame into every parallel pipeline.
+    2. Sends a ``SyncFrame`` sentinel behind it in each pipeline.
+    3. Waits until every pipeline has produced its ``SyncFrame``, meaning all
+       output for that input is ready.
+    4. Releases the collected output frames (deduplicating by frame id, since
+       the same frame may emerge from more than one branch).
+
+    System frames (except ``EndFrame``) bypass this mechanism entirely — they are
+    forwarded through each pipeline and pushed immediately, since system frames
+    are expected to race ahead of regular data frames.
 
     By default, output frames are pushed in the order they arrive from any pipeline
     (``FrameOrder.ARRIVAL``). Set ``frame_order=FrameOrder.PIPELINE`` to push frames
@@ -239,16 +251,11 @@ class SyncParallelPipeline(BasePipeline):
         await asyncio.gather(*[p.cleanup() for p in self._pipelines])
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
-        """Process frames through all parallel pipelines with synchronization.
+        """Send a frame through all parallel pipelines and release output once all finish.
 
-        Distributes frames to all parallel pipelines and synchronizes their output
-        to maintain proper ordering and prevent duplicate processing. Uses SyncFrame
-        control frames to coordinate between parallel paths.
-
-        When ``frame_order`` is ``FrameOrder.ARRIVAL``, output frames are pushed in
-        the order they arrive from any pipeline (via a shared queue). When it is
-        ``FrameOrder.PIPELINE``, each pipeline collects its output into a separate
-        list and the lists are drained in pipeline definition order.
+        System frames (except EndFrame) skip synchronization and pass straight
+        through. All other frames are fanned out to every pipeline, and output is
+        held until every pipeline signals completion (via SyncFrame).
 
         Args:
             frame: The frame to process.

From 57fd29f0c4204ca5df58468108325394cb211a95 Mon Sep 17 00:00:00 2001
From: Paul Kompfner <paul@daily.co>
Date: Thu, 19 Mar 2026 09:57:26 -0400
Subject: [PATCH 12/12] Remove changelog fragment that no longer applies after
 a rebase

---
 changelog/4029.fixed.2.md | 1 -
 1 file changed, 1 deletion(-)
 delete mode 100644 changelog/4029.fixed.2.md

diff --git a/changelog/4029.fixed.2.md b/changelog/4029.fixed.2.md
deleted file mode 100644
index faf54592e..000000000
--- a/changelog/4029.fixed.2.md
+++ /dev/null
@@ -1 +0,0 @@
-- Fixed TTS frame ordering when an audio context is active. Pass-through frames are now routed through the audio context queue so they stay properly ordered with TTS audio frames. Previously, a frame that preceded some incoming text to a TTS service could "jump the queue", getting ahead of the outgoing audio.