PipelineTask: add on_pipeline_finished event

This deprecates `on_pipeline_stopped`, `on_pipeline_ended` and
`on_pipeline_cancelled`.

CHANGELOG.md

@@ -7,6 +7,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Added `on_pipeline_finished` event to `PipelineTask`. This event will get
+  fired when the pipeline is done running. This can be the result of a
+  `StopFrame`, `CancelFrame` or `EndFrame`.
+
+  ```python
+  @task.event_handler("on_pipeline_finished")
+  async def on_pipeline_finished(task: PipelineTask, frame: Frame):
+      ...
+  ```
+
+### Deprecated
+
+- `PipelineTask` events `on_pipeline_stopped`, `on_pipeline_ended` and
+  `on_pipeline_cancelled` are now deprecated. Use `on_pipeline_finished`
+  instead.
+
 ### Fixed
 
 - Fixed a `FastAPIWebsocketTransport` and `SmallWebRTCTransport` issue where

src/pipecat/pipeline/task.py

@@ -115,9 +115,28 @@ class PipelineTask(BasePipelineTask):
     - on_frame_reached_downstream: Called when downstream frames reach the sink
     - on_idle_timeout: Called when pipeline is idle beyond timeout threshold
     - on_pipeline_started: Called when pipeline starts with StartFrame
-    - on_pipeline_stopped: Called when pipeline stops with StopFrame
+    - on_pipeline_stopped: [deprecated] Called when pipeline stops with StopFrame
-    - on_pipeline_ended: Called when pipeline ends with EndFrame
+
-    - on_pipeline_cancelled: Called when pipeline is cancelled
+            .. deprecated:: 0.0.86
+                Use `on_pipeline_finished` instead.
+
+    - on_pipeline_ended: [deprecated] Called when pipeline ends with EndFrame
+
+            .. deprecated:: 0.0.86
+                Use `on_pipeline_finished` instead.
+
+    - on_pipeline_cancelled: [deprecated] Called when pipeline is cancelled with CancelFrame
+
+            .. deprecated:: 0.0.86
+                Use `on_pipeline_finished` instead.
+
+    - on_pipeline_finished: Called after the pipeline has reached any terminal state.
+          This includes:
+              - StopFrame: pipeline was stopped (processors keep connections open)
+              - EndFrame: pipeline ended normally
+              - CancelFrame: pipeline was cancelled
+          Use this event for cleanup, logging, or post-processing tasks. Users can inspect
+          the frame if they need to handle specific cases.
 
     Example::
 
@@ -128,6 +147,10 @@ class PipelineTask(BasePipelineTask):
         @task.event_handler("on_idle_timeout")
         async def on_pipeline_idle_timeout(task):
             ...
+
+        @task.event_handler("on_pipeline_finished")
+        async def on_pipeline_finished(task, frame):
+            ...
     """
 
     def __init__(
@@ -264,6 +287,7 @@ class PipelineTask(BasePipelineTask):
         self._register_event_handler("on_pipeline_stopped")
         self._register_event_handler("on_pipeline_ended")
         self._register_event_handler("on_pipeline_cancelled")
+        self._register_event_handler("on_pipeline_finished")
 
     @property
     def params(self) -> PipelineParams:
@@ -292,6 +316,27 @@ class PipelineTask(BasePipelineTask):
         """
         return self._turn_trace_observer
 
+    def event_handler(self, event_name: str):
+        """Decorator for registering event handlers.
+
+        Args:
+            event_name: The name of the event to handle.
+
+        Returns:
+            The decorator function that registers the handler.
+        """
+        if event_name in ["on_pipeline_stopped", "on_pipeline_ended", "on_pipeline_cancelled"]:
+            import warnings
+
+            with warnings.catch_warnings():
+                warnings.simplefilter("always")
+                warnings.warn(
+                    f"Event '{event_name}' is deprecated, use 'on_pipeline_finished' instead.",
+                    DeprecationWarning,
+                )
+
+        return super().event_handler(event_name)
+
     def add_observer(self, observer: BaseObserver):
         """Add an observer to monitor pipeline execution.
 
@@ -534,6 +579,7 @@ class PipelineTask(BasePipelineTask):
                 )
             finally:
                 await self._call_event_handler("on_pipeline_cancelled", frame)
+                await self._call_event_handler("on_pipeline_finished", frame)
 
         logger.debug(f"{self}: Closing. Waiting for {frame} to reach the end of the pipeline...")
 
@@ -681,9 +727,11 @@ class PipelineTask(BasePipelineTask):
             self._pipeline_start_event.set()
         elif isinstance(frame, EndFrame):
             await self._call_event_handler("on_pipeline_ended", frame)
+            await self._call_event_handler("on_pipeline_finished", frame)
             self._pipeline_end_event.set()
         elif isinstance(frame, StopFrame):
             await self._call_event_handler("on_pipeline_stopped", frame)
+            await self._call_event_handler("on_pipeline_finished", frame)
             self._pipeline_end_event.set()
         elif isinstance(frame, CancelFrame):
             self._pipeline_end_event.set()

tests/test_pipeline.py

@@ -196,10 +196,10 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase):
             nonlocal start_received
             start_received = True
 
-        @task.event_handler("on_pipeline_ended")
+        @task.event_handler("on_pipeline_finished")
-        async def on_pipeline_ended(task, frame: EndFrame):
+        async def on_pipeline_finished(task, frame: Frame):
             nonlocal end_received
-            end_received = True
+            end_received = isinstance(frame, EndFrame)
 
         await task.queue_frame(EndFrame())
         await task.run(PipelineTaskParams(loop=asyncio.get_event_loop()))
@@ -214,10 +214,10 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase):
         pipeline = Pipeline([identity])
         task = PipelineTask(pipeline)
 
-        @task.event_handler("on_pipeline_stopped")
+        @task.event_handler("on_pipeline_finished")
-        async def on_pipeline_ended(task, frame: StopFrame):
+        async def on_pipeline_finished(task, frame: Frame):
             nonlocal stop_received
-            stop_received = True
+            stop_received = isinstance(frame, StopFrame)
 
         await task.queue_frame(StopFrame())
         await task.run(PipelineTaskParams(loop=asyncio.get_event_loop()))
@@ -441,10 +441,10 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase):
         async def on_pipeline_started(task: PipelineTask, frame: StartFrame):
             await task.cancel()
 
-        @task.event_handler("on_pipeline_cancelled")
+        @task.event_handler("on_pipeline_finished")
-        async def on_pipeline_cancelled(task: PipelineTask, frame: CancelFrame):
+        async def on_pipeline_finished(task: PipelineTask, frame: Frame):
             nonlocal cancelled
-            cancelled = True
+            cancelled = isinstance(frame, CancelFrame)
 
         try:
             await task.run(PipelineTaskParams(loop=asyncio.get_event_loop()))