From 901899aa19acbb6da938d70292740994654ffe93 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= <aleix@daily.co>
Date: Wed, 27 Aug 2025 18:17:29 -0700
Subject: [PATCH] PipelineTask: handle BotInterruptionFrame in PipelineTask

This allows having processors before the input transport or even not having an
input transport.
---
 CHANGELOG.md                         |  4 ++++
 src/pipecat/pipeline/task.py         |  8 +++++++-
 src/pipecat/transports/base_input.py | 10 ----------
 tests/test_dtmf_aggregator.py        |  7 +++++++
 4 files changed, 18 insertions(+), 11 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index bc1aa0e68..518988297 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -97,6 +97,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- `BotInterruptionFrame` frame handling has been moved from `BaseInputTransport`
+  to `PipelineTask`. This allows any type of pipeline to handle
+  `BotInterruptionFrame` without the need of an input transport.
+
 - `pipeline.tests.utils.run_test()` now allows passing `PipelineParams` instead
   of individual parameters.
 
diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py
index 23b8c5cc2..c4af5ee93 100644
--- a/src/pipecat/pipeline/task.py
+++ b/src/pipecat/pipeline/task.py
@@ -23,6 +23,7 @@ from pipecat.audio.interruptions.base_interruption_strategy import BaseInterrupt
 from pipecat.clocks.base_clock import BaseClock
 from pipecat.clocks.system_clock import SystemClock
 from pipecat.frames.frames import (
+    BotInterruptionFrame,
     BotSpeakingFrame,
     CancelFrame,
     CancelTaskFrame,
@@ -36,6 +37,7 @@ from pipecat.frames.frames import (
     LLMFullResponseEndFrame,
     MetricsFrame,
     StartFrame,
+    StartInterruptionFrame,
     StopFrame,
     StopTaskFrame,
     TranscriptionFrame,
@@ -632,7 +634,11 @@ class PipelineTask(BasePipelineTask):
         if isinstance(frame, self._reached_upstream_types):
             await self._call_event_handler("on_frame_reached_upstream", frame)
 
-        if isinstance(frame, EndTaskFrame):
+        if isinstance(frame, BotInterruptionFrame) and self.params.allow_interruptions:
+            # Tell the pipeline we should interrupt.
+            logger.debug("Bot interruption")
+            await self.queue_frame(StartInterruptionFrame())
+        elif isinstance(frame, EndTaskFrame):
             # Tell the task we should end nicely.
             await self.queue_frame(EndFrame())
         elif isinstance(frame, CancelTaskFrame):
diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py
index aad2fb842..a6ece651e 100644
--- a/src/pipecat/transports/base_input.py
+++ b/src/pipecat/transports/base_input.py
@@ -22,7 +22,6 @@ from pipecat.audio.turn.base_turn_analyzer import (
 )
 from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADState
 from pipecat.frames.frames import (
-    BotInterruptionFrame,
     BotStartedSpeakingFrame,
     BotStoppedSpeakingFrame,
     CancelFrame,
@@ -289,8 +288,6 @@ class BaseInputTransport(FrameProcessor):
         elif isinstance(frame, CancelFrame):
             await self.cancel(frame)
             await self.push_frame(frame, direction)
-        elif isinstance(frame, BotInterruptionFrame):
-            await self._handle_bot_interruption(frame)
         elif isinstance(frame, BotStartedSpeakingFrame):
             await self._handle_bot_started_speaking(frame)
             await self.push_frame(frame, direction)
@@ -335,13 +332,6 @@ class BaseInputTransport(FrameProcessor):
     # Handle interruptions
     #
 
-    async def _handle_bot_interruption(self, frame: BotInterruptionFrame):
-        """Handle bot interruption frames."""
-        logger.debug("Bot interruption")
-        if self.interruptions_allowed:
-            await self._start_interruption()
-            await self.push_frame(StartInterruptionFrame())
-
     async def _handle_user_interruption(self, frame: Frame):
         """Handle user interruption events based on speaking state."""
         if isinstance(frame, UserStartedSpeakingFrame):
diff --git a/tests/test_dtmf_aggregator.py b/tests/test_dtmf_aggregator.py
index 697d50d27..0d2acb514 100644
--- a/tests/test_dtmf_aggregator.py
+++ b/tests/test_dtmf_aggregator.py
@@ -12,6 +12,7 @@ from pipecat.frames.frames import (
     KeypadEntry,
     TranscriptionFrame,
 )
+from pipecat.pipeline.task import PipelineParams
 from pipecat.processors.aggregators.dtmf_aggregator import DTMFAggregator
 from pipecat.tests.utils import SleepFrame, run_test
 
@@ -69,6 +70,8 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase):
             aggregator,
             frames_to_send=frames_to_send,
             expected_down_frames=expected_down_frames,
+            # TODO(aleix): we should handle StartInterruptionFrame
+            pipeline_params=PipelineParams(allow_interruptions=False),
         )
 
         # Find the TranscriptionFrames
@@ -105,6 +108,8 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase):
             aggregator,
             frames_to_send=frames_to_send,
             expected_down_frames=expected_down_frames,
+            # TODO(aleix): we should handle StartInterruptionFrame
+            pipeline_params=PipelineParams(allow_interruptions=False),
         )
 
         transcription_frames = [
@@ -134,6 +139,8 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase):
             aggregator,
             frames_to_send=frames_to_send,
             expected_down_frames=expected_down_frames,
+            # TODO(aleix): we should handle StartInterruptionFrame
+            pipeline_params=PipelineParams(allow_interruptions=False),
             send_end_frame=False,  # We're sending one in the test to test EndFrame logic
         )