From 2345090b10541adbb52c4395b336ccdfed1fcec2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 5 Feb 2026 22:14:50 -0800 Subject: [PATCH] Attach asyncio.Event to InterruptionFrame for completion signaling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 43 +++++++++---- src/pipecat/pipeline/task.py | 4 +- .../aggregators/llm_response_universal.py | 6 ++ src/pipecat/processors/frame_processor.py | 60 ++++++++++--------- 4 files changed, 73 insertions(+), 40 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index b0e196a22..b7be8045e 100644 --- a/src/pipecat/frames/frames.py +++ b/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 diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index e643164eb..8bda19b18 100644 --- a/src/pipecat/pipeline/task.py +++ b/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) diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index db28b85c1..cf65c6443 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/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) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 740c19ae6..f0c9e7183 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/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. - # Then we wait for the corresponding `InterruptionFrame` to travel from - # the start of the pipeline back to the processor that sent the - # `InterruptionTaskFrame`. This wait is handled using the following - # event. + # Set while awaiting push_interruption_task_frame_and_wait() so that + # _start_interruption() knows not to cancel the process task. 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 - # frames and we will process the frame right away. This is because a - # previous system frame might be waiting for the interruption frame and - # it's blocking the input task. + # If we are waiting for an interruption, bypass all queued system frames + # and process the frame right away. This is because a previous system + # frame might be waiting for the interruption frame blocking the input + # 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 - task and waits to receive the corresponding `InterruptionFrame`. When - the function finishes it is guaranteed that the `InterruptionFrame` has - been pushed downstream. + This function sends an `InterruptionTaskFrame` upstream to the + pipeline task. The task creates a corresponding `InterruptionFrame` + and sends it downstream through the pipeline. An `asyncio.Event` is + 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 - # 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() + await self.push_frame(InterruptionTaskFrame(event=event), FrameDirection.UPSTREAM) - # Clean the event. - self._wait_interruption_event.clear() + # Wait for the `InterruptionFrame` to complete and log a warning + # 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