From ba358a4f0acd98f51e3ecb5ef473ddaae3b8dc91 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= <aleix@daily.co>
Date: Tue, 28 Jan 2025 15:01:20 -0800
Subject: [PATCH] task: cleanup processors after task finishes running

---
 CHANGELOG.md                 | 3 +++
 src/pipecat/pipeline/task.py | 2 +-
 2 files changed, 4 insertions(+), 1 deletion(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 6886f37d7..8c26602d8 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -12,6 +12,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Fixed a `DailyTransport` issue that would cause events to be triggered before
   join finished.
 
+- Fixed a `PipelineTask` issue that was preventing processors to be cleaned up
+  after cancelling the task.
+
 ## [0.0.54] - 2025-01-27
 
 ### Added
diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py
index 45596980f..9f59e8177 100644
--- a/src/pipecat/pipeline/task.py
+++ b/src/pipecat/pipeline/task.py
@@ -160,7 +160,6 @@ class PipelineTask(BaseTask):
         await self._source.push_frame(CancelFrame())
         # Only cancel the push task. Everything else will be cancelled in run().
         await cancel_task(self._process_push_task)
-        await self._cleanup()
 
     async def run(self):
         """
@@ -176,6 +175,7 @@ class PipelineTask(BaseTask):
             # awaiting a task.
             pass
         await self._cancel_tasks()
+        await self._cleanup()
         self._finished = True
 
     async def queue_frame(self, frame: Frame):