From d2341e01993434cec32e5194a128c4aa840d5053 Mon Sep 17 00:00:00 2001
From: Paul Kompfner <paul@daily.co>
Date: Fri, 13 Mar 2026 22:09:44 -0400
Subject: [PATCH] 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
             ]
         )