wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: preserve frame insertion order in BaseOutputTransport for equal PTS values

Browse Source 
Frames sharing the same presentation timestamp were being reordered by the
priority queue. Adds a monotonic counter as a tiebreaker so frames with equal
PTS are always emitted in insertion order, preventing subtle audio/text
sequencing bugs.
This commit is contained in:
filipi87
2026-05-20 10:03:08 -03:00
parent deefc32faf
commit 6b9deefbe3
1 changed files with 5 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
src/pipecat/transports/base_output.py
View File

@@ -448,6 +448,9 @@ class BaseOutputTransport(FrameProcessor):
self._video_task: asyncio.Task | None = None self._video_task: asyncio.Task | None = None
self._clock_task: asyncio.Task | None = None self._clock_task: asyncio.Task | None = None
# If timestamps are equal, use this count to preserve the insertion order
self._clock_queue_counter = itertools.count()
@property @property
def sample_rate(self) -> int: def sample_rate(self) -> int:
"""Get the audio sample rate. """Get the audio sample rate.
@@ -498,7 +501,7 @@ class BaseOutputTransport(FrameProcessor):
frame: The end frame signaling sender shutdown. frame: The end frame signaling sender shutdown.
""" """
# Let the sink tasks process the queue until they reach this EndFrame. # Let the sink tasks process the queue until they reach this EndFrame.
await self._clock_queue.put((float("inf"), frame.id, frame)) await self._clock_queue.put((float("inf"), next(self._clock_queue_counter), frame))
await self._audio_queue.put(frame) await self._audio_queue.put(frame)
# At this point we have enqueued an EndFrame and we need to wait for # At this point we have enqueued an EndFrame and we need to wait for
@@ -610,7 +613,7 @@ class BaseOutputTransport(FrameProcessor):
Args: Args:
frame: The frame with timing information to handle. frame: The frame with timing information to handle.
""" """
await self._clock_queue.put((frame.pts, frame.id, frame)) await self._clock_queue.put((frame.pts, next(self._clock_queue_counter), frame))
async def handle_sync_frame(self, frame: Frame): async def handle_sync_frame(self, frame: Frame):
"""Handle frames that need synchronized processing. """Handle frames that need synchronized processing.