From 49a5a1e375fe0eca0e345c36bbbd94a85e56c91b Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= <aleix@daily.co>
Date: Tue, 5 Aug 2025 11:25:29 -0700
Subject: [PATCH] PipelineTask: improve task cancellation

---
 src/pipecat/pipeline/task.py              | 33 +++++++++++++++--------
 src/pipecat/processors/frame_processor.py |  2 +-
 src/pipecat/services/heygen/video.py      |  4 +--
 3 files changed, 25 insertions(+), 14 deletions(-)

diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py
index 8b745c2a7..1640d589c 100644
--- a/src/pipecat/pipeline/task.py
+++ b/src/pipecat/pipeline/task.py
@@ -459,15 +459,20 @@ class PipelineTask(BasePipelineTask):
             # awaiting a task.
             pass
         finally:
-            # It's possibe that we get an asyncio.CancelledError from the
-            # outside, if so we need to make sure everything gets cancelled
-            # properly.
-            if cleanup_pipeline:
-                await self._cancel()
-            await self._cancel_tasks()
-            await self._cleanup(cleanup_pipeline)
-            if self._check_dangling_tasks:
-                self._print_dangling_tasks()
+            # We only cancel things cleanly if we know we are the ones
+            # cancelling. It's possibe that we get an asyncio.CancelledError
+            # from the outside, in which case it is very likely other tasks have
+            # been already cancelled (e.g. when python is shutting down) so we
+            # can't assume things are being cancelled nicely.
+            if self._cancelled:
+                await self._cancel_tasks()
+                await self._cleanup(cleanup_pipeline)
+                if self._check_dangling_tasks:
+                    self._print_dangling_tasks()
+            else:
+                logger.warning(
+                    f"Pipeline task {self} is not being cancelled properly (use cancel() method)"
+                )
             self._finished = True
 
     async def queue_frame(self, frame: Frame):
@@ -494,7 +499,7 @@ class PipelineTask(BasePipelineTask):
     async def _cancel(self):
         """Internal cancellation logic for the pipeline task."""
         if not self._cancelled:
-            logger.debug(f"Canceling pipeline task {self}")
+            logger.debug(f"Cancelling pipeline task {self}")
             self._cancelled = True
             # Make sure everything is cleaned up downstream. This is sent
             # out-of-band from the main streaming task which is what we want since
@@ -502,7 +507,9 @@ class PipelineTask(BasePipelineTask):
             await self._source.push_frame(CancelFrame())
             # Wait for CancelFrame to make it throught the pipeline.
             await self._wait_for_pipeline_end()
-            # Only cancel the push task. Everything else will be cancelled in run().
+            # Only cancel the push task, we don't want to be able to process any
+            # other frame after cancel. Everything else will be cancelled in
+            # run().
             if self._process_push_task:
                 await self._task_manager.cancel_task(self._process_push_task)
                 self._process_push_task = None
@@ -544,6 +551,10 @@ class PipelineTask(BasePipelineTask):
         """Cancel all running pipeline tasks."""
         await self._observer.stop()
 
+        if self._process_push_task:
+            await self._task_manager.cancel_task(self._process_push_task)
+            self._process_push_task = None
+
         if self._process_up_task:
             await self._task_manager.cancel_task(self._process_up_task)
             self._process_up_task = None
diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py
index e6611a161..fda818572 100644
--- a/src/pipecat/processors/frame_processor.py
+++ b/src/pipecat/processors/frame_processor.py
@@ -556,7 +556,7 @@ class FrameProcessor(BaseObject):
     #
 
     async def _start_interruption(self):
-        """Start handling an interruption by canceling current tasks."""
+        """Start handling an interruption by cancelling current tasks."""
         try:
             # Cancel the input task. This will stop processing queued frames.
             await self.__cancel_input_task()
diff --git a/src/pipecat/services/heygen/video.py b/src/pipecat/services/heygen/video.py
index 237dfc36f..f32c8f392 100644
--- a/src/pipecat/services/heygen/video.py
+++ b/src/pipecat/services/heygen/video.py
@@ -183,7 +183,7 @@ class HeyGenVideoService(AIService):
     async def stop(self, frame: EndFrame):
         """Stop the HeyGen video service gracefully.
 
-        Performs cleanup by ending the conversation and canceling ongoing tasks
+        Performs cleanup by ending the conversation and cancelling ongoing tasks
         in a controlled manner.
 
         Args:
@@ -241,7 +241,7 @@ class HeyGenVideoService(AIService):
         Manages the interruption flow by:
         1. Setting the interruption flag
         2. Signaling the client to interrupt current speech
-        3. Canceling ongoing audio sending tasks
+        3. Cancelling ongoing audio sending tasks
         4. Creating a new send task
         5. Activating the avatar's listening animation
         """