FrameProcessor: ignore other frames during CancelFrame

CHANGELOG.md

@@ -43,6 +43,9 @@ async def on_audio_data(processor, audio, sample_rate, num_channels):
 
 ### Fixed
 
+- Fixed an issue where other frames were being processed while a `CancelFrame`
+  was being pushed down the pipeline.
+
 - `AudioBufferProcessor` now handles interruptions properly.
 
 - Fixed a `WebsocketServerTransport` issue that would prevent interruptions with

src/pipecat/processors/frame_processor.py

@@ -13,6 +13,7 @@ from loguru import logger
 
 from pipecat.clocks.base_clock import BaseClock
 from pipecat.frames.frames import (
+    CancelFrame,
     EndFrame,
     ErrorFrame,
     Frame,
@@ -58,6 +59,13 @@ class FrameProcessor:
         self._enable_usage_metrics = False
         self._report_only_initial_ttfb = False
 
+        # Cancellation is done through CancelFrame (a system frame). This could
+        # cause other events being triggered (e.g. closing a transport) which
+        # could also cause other frames to be pushed from other tasks
+        # (e.g. EndFrame). So, when we are cancelling we don't want anything
+        # else to be pushed.
+        self._cancelling = False
+
         # Metrics
         self._metrics = metrics or FrameProcessorMetrics()
         self._metrics.set_processor_name(self.name)
@@ -161,6 +169,10 @@ class FrameProcessor:
             Callable[["FrameProcessor", Frame, FrameDirection], Awaitable[None]]
         ] = None,
     ):
+        # If we are cancelling we don't want to process any other frame.
+        if self._cancelling:
+            return
+
         if isinstance(frame, SystemFrame):
             # We don't want to queue system frames.
             await self.process_frame(frame, direction)
@@ -187,6 +199,8 @@ class FrameProcessor:
             await self.stop_all_metrics()
         elif isinstance(frame, StopInterruptionFrame):
             self._should_report_ttfb = True
+        elif isinstance(frame, CancelFrame):
+            self._cancelling = True
 
     async def push_error(self, error: ErrorFrame):
         await self.push_frame(error, FrameDirection.UPSTREAM)