From 4945cfbd8fe34d16efe324c12cbc9852756da77b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 6 Feb 2026 15:12:48 -0800 Subject: [PATCH 1/3] Buffer internal frames during ParallelPipeline lifecycle synchronization Processors inside parallel sub-pipelines can push frames during StartFrame/EndFrame/CancelFrame processing. Previously these frames could escape the ParallelPipeline before all branches finished processing the lifecycle frame. Now they are buffered and flushed after synchronization completes. --- changelog/3668.fixed.md | 1 + src/pipecat/pipeline/parallel_pipeline.py | 24 +++++++++++++++++-- tests/test_pipeline.py | 28 +++++++++++++++++++++++ 3 files changed, 51 insertions(+), 2 deletions(-) create mode 100644 changelog/3668.fixed.md diff --git a/changelog/3668.fixed.md b/changelog/3668.fixed.md new file mode 100644 index 000000000..6885a7591 --- /dev/null +++ b/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. diff --git a/src/pipecat/pipeline/parallel_pipeline.py b/src/pipecat/pipeline/parallel_pipeline.py index 81beeead8..88ea04638 100644 --- a/src/pipecat/pipeline/parallel_pipeline.py +++ b/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) diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index dda9e583d..71121a3fc 100644 --- a/tests/test_pipeline.py +++ b/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): From 57df03aade6e050a9984d3a541bd9af8119e74d1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 6 Feb 2026 15:12:53 -0800 Subject: [PATCH 2/3] Update CLAUDE.md with PR workflow instructions --- CLAUDE.md | 26 ++++++++------------------ 1 file changed, 8 insertions(+), 18 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 162615ab2..d7ef2619c 100644 --- a/CLAUDE.md +++ b/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/..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 ` to generate the changelog file and `/pr-description ` to update the PR description. From cd03d449cb9cfb48592cdfef9c77efa73fdb15bf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 6 Feb 2026 15:12:58 -0800 Subject: [PATCH 3/3] Update changelog skill with skip rules and allowed types --- .claude/skills/changelog/SKILL.md | 19 +++++++++++++------ 1 file changed, 13 insertions(+), 6 deletions(-) diff --git a/.claude/skills/changelog/SKILL.md b/.claude/skills/changelog/SKILL.md index 0feb6b92c..89e13a40e 100644 --- a/.claude/skills/changelog/SKILL.md +++ b/.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