Merge pull request #3668 from pipecat-ai/aleix/parallel-pipeline-buffering

Buffer internal frames during ParallelPipeline lifecycle sync

.claude/skills/changelog/SKILL.md

@@ -7,23 +7,30 @@ Create changelog files for the important commits in this PR. The PR number is pr
 
 ## Instructions
 
-1. First, check what commits are on the current branch compared to main:
+1. Skip changelog for: documentation-only, internal refactoring, test-only, CI changes.
+
+2. First, check what commits are on the current branch compared to main:
    ```
    git log main..HEAD --oneline
    ```
 
-2. For each significant change, create a changelog file in the `changelog/` folder using the format:
+3. For each significant change, create a changelog file in the `changelog/` folder using the format:
+   Allowed types: `added`, `changed`, `deprecated`, `removed`, `fixed`, `security`, `performance`, `other`
    - `{PR_NUMBER}.added.md` - for new features
-   - `{PR_NUMBER}.added.2.md`, `{PR_NUMBER}.added.3.md` - for additional new features
+   - `{PR_NUMBER}.added.2.md`, `{PR_NUMBER}.added.3.md` - for additional entries of the same type
    - `{PR_NUMBER}.changed.md` - for changes to existing functionality
    - `{PR_NUMBER}.fixed.md` - for bug fixes
    - `{PR_NUMBER}.deprecated.md` - for deprecations
+   - `{PR_NUMBER}.removed.md` - for removed features
+   - `{PR_NUMBER}.security.md` - for security fixes
+   - `{PR_NUMBER}.performance.md` - for performance improvements
+   - `{PR_NUMBER}.other.md` - for other changes
 
-3. Each changelog file should at least contain a main single line starting with `- ` followed by a clear description of the change.
+4. Each changelog file should at least contain a main single line starting with `- ` followed by a clear description of the change.
 
-4. If the change is complicated, changelog files can have indented lines after the main line with additional details or code samples.
+5. If the change is complicated, changelog files can have indented lines after the main line with additional details or code samples.
 
-5. Use ⚠️ emoji prefix for breaking changes.
+6. Use ⚠️ emoji prefix for breaking changes.
 
 ## Example
 

CLAUDE.md

@@ -27,6 +27,10 @@ uv run pytest tests/test_name.py::test_function_name
 # Preview changelog
 towncrier build --draft --version Unreleased
 
+# Lint and format check
+uv run ruff check
+uv run ruff format --check
+
 # Update dependencies (after editing pyproject.toml)
 uv lock && uv sync
 ```
@@ -122,24 +126,6 @@ class MyService(LLMService):
         super().__init__(**kwargs)
 ```
 
-## Changelog
-
-Every user-facing PR needs a changelog fragment in `changelog/`:
-
-```
-changelog/<PR_number>.<type>.md
-```
-
-Types: `added`, `changed`, `deprecated`, `removed`, `fixed`, `security`, `other`
-
-Content format (include the `-`):
-
-```markdown
-- Added support for new feature X.
-```
-
-Skip changelog for: documentation-only, internal refactoring, test-only, CI changes.
-
 ## Service Implementation
 
 When adding a new service:
@@ -151,3 +137,7 @@ When adding a new service:
 5. Push `ErrorFrame` on failures
 6. Add metrics tracking via `MetricsData` if relevant
 7. Follow the pattern of existing services in `src/pipecat/services/`
+
+## Pull Requests
+
+After creating a PR, use `/changelog <pr_number>` to generate the changelog file and `/pr-description <pr_number>` to update the PR description.

changelog/3668.fixed.md

@@ -0,0 +1 @@
+- Fixed `ParallelPipeline` allowing frames pushed by internal processors to escape during lifecycle frame (`StartFrame`/`EndFrame`/`CancelFrame`) synchronization. These frames are now buffered and flushed after all branches complete.

src/pipecat/pipeline/parallel_pipeline.py

@@ -52,6 +52,8 @@ class ParallelPipeline(BasePipeline):
 
         self._seen_ids = set()
         self._frame_counter: Dict[int, int] = {}
+        self._synchronizing: bool = False
+        self._buffered_frames: list[tuple[Frame, FrameDirection]] = []
 
         logger.debug(f"Creating {self} pipelines")
 
@@ -143,6 +145,7 @@ class ParallelPipeline(BasePipeline):
         # Parallel pipeline synchronized frames.
         if isinstance(frame, (StartFrame, EndFrame, CancelFrame)):
             self._frame_counter[frame.id] = len(self._pipelines)
+            self._synchronizing = True
             await self.pause_processing_system_frames()
             await self.pause_processing_frames()
 
@@ -151,10 +154,18 @@ class ParallelPipeline(BasePipeline):
             await p.queue_frame(frame, direction)
 
     async def _parallel_push_frame(self, frame: Frame, direction: FrameDirection):
-        """Push frames while avoiding duplicates using frame ID tracking."""
+        """Push frames while avoiding duplicates using frame ID tracking.
+
+        During lifecycle frame synchronization, non-lifecycle frames are buffered
+        to prevent them from escaping the parallel pipeline before all branches
+        have finished processing the lifecycle frame.
+        """
         if frame.id not in self._seen_ids:
             self._seen_ids.add(frame.id)
-            await self.push_frame(frame, direction)
+            if self._synchronizing:
+                self._buffered_frames.append((frame, direction))
+            else:
+                await self.push_frame(frame, direction)
 
     async def _pipeline_sink_push_frame(self, frame: Frame, direction: FrameDirection):
         # Parallel pipeline synchronized frames.
@@ -167,8 +178,17 @@ 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()
                 await self.resume_processing_system_frames()
                 await self.resume_processing_frames()
         else:
             await self._parallel_push_frame(frame, direction)
+
+    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:
+            await self.push_frame(frame, direction)

tests/test_pipeline.py

@@ -96,6 +96,34 @@ class TestParallelPipeline(unittest.IsolatedAsyncioTestCase):
             expected_down_frames=expected_down_frames,
         )
 
+    async def test_parallel_internal_frames_buffered_during_start(self):
+        """Frames pushed by internal processors during StartFrame processing
+        should be buffered and only released after StartFrame synchronization
+        completes."""
+
+        class EmitOnStartProcessor(FrameProcessor):
+            """Pushes a TextFrame when it receives a StartFrame."""
+
+            async def process_frame(self, frame: Frame, direction: FrameDirection):
+                await super().process_frame(frame, direction)
+                await self.push_frame(frame, direction)
+                if isinstance(frame, StartFrame):
+                    await self.push_frame(TextFrame(text="from start"))
+
+        pipeline = ParallelPipeline([EmitOnStartProcessor()], [IdentityFilter()])
+
+        frames_to_send = [TextFrame(text="Hello!")]
+
+        # StartFrame should come first, then the TextFrame emitted during
+        # StartFrame processing, then the regular TextFrame.
+        expected_down_frames = [StartFrame, TextFrame, TextFrame]
+        await run_test(
+            pipeline,
+            frames_to_send=frames_to_send,
+            expected_down_frames=expected_down_frames,
+            ignore_start=False,
+        )
+
 
 class TestPipelineTask(unittest.IsolatedAsyncioTestCase):
     async def test_task_single(self):