From e85e93b9b18ac16eafb3ae65b835bb9a3820133d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Sat, 1 Nov 2025 14:46:46 -0700 Subject: [PATCH 1/2] CancelFrame: add reason field to indicate why pipeline is being cancelled --- CHANGELOG.md | 9 +++++++++ src/pipecat/frames/frames.py | 16 ++++++++++++++-- src/pipecat/pipeline/task.py | 22 +++++++++++++++------- 3 files changed, 38 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 75dcc1b31..2a6ccc102 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,15 @@ All notable changes to **Pipecat** will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +### Added + +- `CancelFrame` and `CancelTaskFrame` have an optional `reason` field to + indicate why the pipeline is being canceled. This can be also specified when + you cancel a task with `PipelineTask.cancel(reason="cancellation your + reason")`. + ## [0.0.92] - 2025-10-31 🎃 "The Haunted Edition" 👻 ### Added diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 6577ae7a4..fd02ba52a 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -773,9 +773,15 @@ class CancelFrame(SystemFrame): Indicates that a pipeline needs to stop right away without processing remaining queued frames. + + Parameters: + reason: Optional reason for pushing a cancel frame. """ - pass + reason: Optional[str] = None + + def __str__(self): + return f"{self.name}(reason: {self.reason})" @dataclass @@ -1378,9 +1384,15 @@ class CancelTaskFrame(TaskFrame): This is used to notify the pipeline task that the pipeline should be stopped immediately by pushing a CancelFrame downstream. This frame should be pushed upstream. + + Parameters: + reason: Optional reason for pushing a cancel frame. """ - pass + reason: Optional[str] = None + + def __str__(self): + return f"{self.name}(reason: {self.reason})" @dataclass diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 1f9cbf63a..d2451bfd0 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -446,10 +446,14 @@ class PipelineTask(BasePipelineTask): logger.debug(f"Task {self} scheduled to stop when done") await self.queue_frame(EndFrame()) - async def cancel(self): - """Request the running pipeline to cancel.""" + async def cancel(self, *, reason: Optional[str] = None): + """Request the running pipeline to cancel. + + Args: + reason: Optional reason to indicate why the pipeline is being cancelled. + """ if not self._finished: - await self._cancel() + await self._cancel(reason=reason) async def run(self, params: PipelineTaskParams): """Start and manage the pipeline execution until completion or cancellation. @@ -513,12 +517,16 @@ class PipelineTask(BasePipelineTask): for frame in frames: await self.queue_frame(frame) - async def _cancel(self): - """Internal cancellation logic for the pipeline task.""" + async def _cancel(self, *, reason: Optional[str] = None): + """Internal cancellation logic for the pipeline task. + + Args: + reason: Optional reason to indicate why the pipeline is being cancelled. + """ if not self._cancelled: logger.debug(f"Cancelling pipeline task {self}") self._cancelled = True - await self.queue_frame(CancelFrame()) + await self.queue_frame(CancelFrame(reason=reason)) async def _create_tasks(self): """Create and start all pipeline processing tasks.""" @@ -720,7 +728,7 @@ class PipelineTask(BasePipelineTask): elif isinstance(frame, CancelTaskFrame): # Tell the task we should end right away. logger.debug(f"{self}: received cancel task frame {frame}") - await self.queue_frame(CancelFrame()) + await self.queue_frame(CancelFrame(reason=frame.reason)) elif isinstance(frame, StopTaskFrame): # Tell the task we should stop nicely. logger.debug(f"{self}: received stop task frame {frame}") From 62e45f466aa7abd6744c1898e965b25731630a7a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Sun, 2 Nov 2025 00:45:27 -0700 Subject: [PATCH 2/2] EndFrame: add reason field to indicate why pipeline is being ended --- CHANGELOG.md | 3 +++ src/pipecat/frames/frames.py | 16 ++++++++++++++-- src/pipecat/pipeline/task.py | 2 +- 3 files changed, 18 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2a6ccc102..93d43a82f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- `EndFrame` and `EndTaskFrame` have an optional `reason` field to indicate why + the pipeline is being ended. + - `CancelFrame` and `CancelTaskFrame` have an optional `reason` field to indicate why the pipeline is being canceled. This can be also specified when you cancel a task with `PipelineTask.cancel(reason="cancellation your diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index fd02ba52a..5db02c856 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -1372,9 +1372,15 @@ class EndTaskFrame(TaskFrame): This is used to notify the pipeline task that the pipeline should be closed nicely (flushing all the queued frames) by pushing an EndFrame downstream. This frame should be pushed upstream. + + Parameters: + reason: Optional reason for pushing an end frame. """ - pass + reason: Optional[str] = None + + def __str__(self): + return f"{self.name}(reason: {self.reason})" @dataclass @@ -1463,9 +1469,15 @@ class EndFrame(ControlFrame): sending frames to its output channel(s) and close all its threads. Note, that this is a control frame, which means it will be received in the order it was sent. + + Parameters: + reason: Optional reason for pushing an end frame. """ - pass + reason: Optional[str] = None + + def __str__(self): + return f"{self.name}(reason: {self.reason})" @dataclass diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index d2451bfd0..ebcb9146d 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -724,7 +724,7 @@ class PipelineTask(BasePipelineTask): if isinstance(frame, EndTaskFrame): # Tell the task we should end nicely. logger.debug(f"{self}: received end task frame {frame}") - await self.queue_frame(EndFrame()) + await self.queue_frame(EndFrame(reason=frame.reason)) elif isinstance(frame, CancelTaskFrame): # Tell the task we should end right away. logger.debug(f"{self}: received cancel task frame {frame}")