Attach asyncio.Event to InterruptionFrame for completion signaling

Move the interruption wait event from per-processor instance state to
the frame itself. The event is created in
push_interruption_task_frame_and_wait(), threaded through
InterruptionTaskFrame → InterruptionFrame, and set when the frame
reaches the pipeline sink. This scopes the event to each interruption
flow rather than sharing mutable state on the processor.

Also adds a 2s timeout warning to help diagnose cases where
InterruptionFrame.complete() is never called.

src/pipecat/frames/frames.py

@@ -11,6 +11,7 @@ including data frames, system frames, and control frames for audio, video, text,
 and LLM processing.
 """
 
+import asyncio
 from dataclasses import dataclass, field
 from enum import Enum
 from typing import (
@@ -1118,15 +1119,29 @@ class FrameProcessorResumeUrgentFrame(SystemFrame):
 
 @dataclass
 class InterruptionFrame(SystemFrame):
-    """Frame indicating user started speaking (interruption detected).
+    """Frame pushed to interrupt the pipeline.
+
+    This frame is used to interrupt the pipeline. For example, when a user
+    starts speaking to cancel any in-progress bot output. It can also be pushed
+    by any processor.
+
+    Parameters:
+        event: Optional event set when the frame has fully traversed the
+            pipeline.
 
-    Emitted by the BaseInputTransport to indicate that a user has started
-    speaking (i.e. is interrupting). This is similar to
-    UserStartedSpeakingFrame except that it should be pushed concurrently
-    with other frames (so the order is not guaranteed).
     """
 
-    pass
+    event: Optional[asyncio.Event] = None
+
+    def complete(self):
+        """Signal that this interruption has been fully processed.
+
+        Called automatically when the frame reaches the pipeline sink, or
+        manually when the frame is consumed before reaching it (e.g. when
+        the user is muted).
+        """
+        if self.event:
+            self.event.set()
 
 
 @dataclass
@@ -1706,15 +1721,19 @@ class StopTaskFrame(TaskFrame):
 
 @dataclass
 class InterruptionTaskFrame(TaskFrame):
-    """Frame indicating the bot should be interrupted.
+    """Frame indicating the pipeline should be interrupted.
+
+    This frame should be pushed upstream to indicate the pipeline should be
+    interrupted. The pipeline task converts this into an `InterruptionFrame` and
+    sends it downstream. The `event` is passed to the `InterruptionFrame` so it
+    can signal when the interruption has fully traversed the pipeline.
+
+    Parameters:
+        event: Optional event passed to the corresponding `InterruptionFrame`.
 
-    Emitted when the bot should be interrupted. This will mainly cause the
-    same actions as if the user interrupted except that the
-    UserStartedSpeakingFrame and UserStoppedSpeakingFrame won't be generated.
-    This frame should be pushed upstream.
     """
 
-    pass
+    event: Optional[asyncio.Event] = None
 
 
 @dataclass

src/pipecat/pipeline/task.py

@@ -864,7 +864,7 @@ class PipelineTask(BasePipelineTask):
             # pipeline. This is in case the push task is blocked waiting for a
             # pipeline-ending frame to finish traversing the pipeline.
             logger.debug(f"{self}: received interruption task frame {frame}")
-            await self._pipeline.queue_frame(InterruptionFrame())
+            await self._pipeline.queue_frame(InterruptionFrame(event=frame.event))
         elif isinstance(frame, ErrorFrame):
             await self._call_event_handler("on_pipeline_error", frame)
             if frame.fatal:
@@ -903,6 +903,8 @@ class PipelineTask(BasePipelineTask):
             self._pipeline_end_event.set()
         elif isinstance(frame, CancelFrame):
             self._pipeline_end_event.set()
+        elif isinstance(frame, InterruptionFrame):
+            frame.complete()
         elif isinstance(frame, HeartbeatFrame):
             await self._heartbeat_queue.put(frame)
 

src/pipecat/processors/aggregators/llm_response_universal.py

@@ -552,6 +552,12 @@ class LLMUserAggregator(LLMContextAggregator):
         if should_mute_frame:
             logger.trace(f"{frame.name} suppressed - user currently muted")
 
+        # When muted, the InterruptionFrame won't propagate further and
+        # will never reach the pipeline sink. Complete it here so
+        # push_interruption_task_frame_and_wait() doesn't hang.
+        if should_mute_frame and isinstance(frame, InterruptionFrame):
+            frame.complete()
+
         should_mute_next_time = False
         for s in self._params.user_mute_strategies:
             should_mute_next_time |= await s.process_frame(frame)

src/pipecat/processors/frame_processor.py

@@ -52,6 +52,8 @@ from pipecat.processors.metrics.frame_processor_metrics import FrameProcessorMet
 from pipecat.utils.asyncio.task_manager import BaseTaskManager
 from pipecat.utils.base_object import BaseObject
 
+INTERRUPTION_COMPLETION_TIMEOUT = 2.0
+
 
 class FrameDirection(Enum):
     """Direction of frame flow in the processing pipeline.
@@ -240,13 +242,9 @@ class FrameProcessor(BaseObject):
         self.__process_frame_task: Optional[asyncio.Task] = None
         self.__process_current_frame: Optional[Frame] = None
 
-        # To interrupt a pipeline, we push an `InterruptionTaskFrame` upstream.
+        # Set while awaiting push_interruption_task_frame_and_wait() so that
-        # Then we wait for the corresponding `InterruptionFrame` to travel from
+        # _start_interruption() knows not to cancel the process task.
-        # the start of the pipeline back to the processor that sent the
-        # `InterruptionTaskFrame`. This wait is handled using the following
-        # event.
         self._wait_for_interruption = False
-        self._wait_interruption_event = asyncio.Event()
 
         # Frame processor events.
         self._register_event_handler("on_before_process_frame", sync=True)
@@ -600,10 +598,11 @@ class FrameProcessor(BaseObject):
         if self._cancelling:
             return
 
-        # If we are waiting for an interruption we will bypass all queued system
+        # If we are waiting for an interruption, bypass all queued system frames
-        # frames and we will process the frame right away. This is because a
+        # and process the frame right away. This is because a previous system
-        # previous system frame might be waiting for the interruption frame and
+        # frame might be waiting for the interruption frame blocking the input
-        # it's blocking the input task.
+        # task, so this InterruptionFrame would never be dequeued and we'd
+        # deadlock.
         if self._wait_for_interruption and isinstance(frame, InterruptionFrame):
             await self.__process_frame(frame, direction, callback)
             return
@@ -742,31 +741,38 @@ class FrameProcessor(BaseObject):
 
         await self._call_event_handler("on_after_push_frame", frame)
 
-        # If we are waiting for an interruption and we get an interruption, then
-        # we can unblock `push_interruption_task_frame_and_wait()`.
-        if self._wait_for_interruption and isinstance(frame, InterruptionFrame):
-            self._wait_interruption_event.set()
-
     async def push_interruption_task_frame_and_wait(self):
         """Push an interruption task frame upstream and wait for the interruption.
 
-        This function sends an `InterruptionTaskFrame` upstream to the pipeline
+        This function sends an `InterruptionTaskFrame` upstream to the
-        task and waits to receive the corresponding `InterruptionFrame`. When
+        pipeline task. The task creates a corresponding `InterruptionFrame`
-        the function finishes it is guaranteed that the `InterruptionFrame` has
+        and sends it downstream through the pipeline. An `asyncio.Event` is
-        been pushed downstream.
+        attached to both frames so the caller can wait until the interruption
+        has fully traversed the pipeline. The event is set when the
+        `InterruptionFrame` reaches the pipeline sink. If the frame does
+        not complete within `INTERRUPTION_COMPLETION_TIMEOUT` seconds, a
+        warning is logged periodically until it completes.
+
         """
         self._wait_for_interruption = True
 
-        await self.push_frame(InterruptionTaskFrame(), FrameDirection.UPSTREAM)
+        event = asyncio.Event()
 
-        # Wait for an `InterruptionFrame` to come to this processor and be
+        await self.push_frame(InterruptionTaskFrame(event=event), FrameDirection.UPSTREAM)
-        # pushed. Take a look at `push_frame()` to see how we first push the
-        # `InterruptionFrame` and then we set the event in order to maintain
-        # frame ordering.
-        await self._wait_interruption_event.wait()
 
-        # Clean the event.
+        # Wait for the `InterruptionFrame` to complete and log a warning
-        self._wait_interruption_event.clear()
+        # periodically if it takes too long.
+        while not event.is_set():
+            try:
+                await asyncio.wait_for(event.wait(), timeout=INTERRUPTION_COMPLETION_TIMEOUT)
+            except asyncio.TimeoutError:
+                logger.warning(
+                    f"{self}: InterruptionFrame has not completed after"
+                    f" {INTERRUPTION_COMPLETION_TIMEOUT}s. Make sure"
+                    " InterruptionFrame.complete() is being called (e.g. if the"
+                    " frame is being blocked or consumed before reaching the"
+                    " pipeline sink)."
+                )
 
         self._wait_for_interruption = False