Merge pull request #2669 from pipecat-ai/aleix/interruption-task-frame-wait-fixes

interruption task frame wait fixes

CHANGELOG.md

@@ -21,7 +21,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
-- Update `livekit` to 1.0.13.
+- Updated `livekit` to 1.0.13.
 
 - `torch` and `torchaudio` are no longer required for running Smart Turn
   locally. This avoids gigabytes of dependencies being installed.
@@ -44,6 +44,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed an issue that would cause `push_interruption_task_frame_and_wait()` to
+  not wait if a previous interruption had already happened.
+
 - Fixed a couple of bugs in `ServiceSwitcher`:
   - Using multiple `ServiceSwitcher`s in a pipeline would result in an error.
   - `ServiceSwitcherFrame`s (such as `ManuallySwitchServiceFrame`s) were having

src/pipecat/processors/frame_processor.py

@@ -220,6 +220,11 @@ class FrameProcessor(BaseObject):
         self.__process_event: Optional[asyncio.Event] = None
         self.__process_frame_task: Optional[asyncio.Task] = 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.
         self._wait_for_interruption = False
         self._wait_interruption_event = asyncio.Event()
 
@@ -632,7 +637,9 @@ class FrameProcessor(BaseObject):
 
         await self.__internal_push_frame(frame, direction)
 
-        if isinstance(frame, InterruptionFrame):
+        # 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):

tests/test_frame_processor.py

@@ -0,0 +1,67 @@
+#
+# Copyright (c) 2024-2025 Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import unittest
+
+from pipecat.frames.frames import (
+    EndFrame,
+    Frame,
+    InterruptionFrame,
+    TextFrame,
+    TransportMessageUrgentFrame,
+)
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.tests.utils import SleepFrame, run_test
+
+
+class TestFrameProcessor(unittest.IsolatedAsyncioTestCase):
+    async def test_interruption_and_wait(self):
+        class DelayFrameProcessor(FrameProcessor):
+            """This processors just gives time to the event loop to change
+            between tasks. Otherwise things happen to fast."""
+
+            async def process_frame(self, frame: Frame, direction: FrameDirection):
+                await super().process_frame(frame, direction)
+                await asyncio.sleep(0.1)
+                await self.push_frame(frame, direction)
+
+        class InterruptFrameProcessor(FrameProcessor):
+            async def process_frame(self, frame: Frame, direction: FrameDirection):
+                await super().process_frame(frame, direction)
+
+                if isinstance(frame, TextFrame):
+                    await self.push_interruption_task_frame_and_wait()
+                    await self.push_frame(TransportMessageUrgentFrame(message=frame.text))
+                else:
+                    await self.push_frame(frame, direction)
+
+        pipeline = Pipeline([DelayFrameProcessor(), InterruptFrameProcessor()])
+
+        frames_to_send = [
+            # Just a random interruption to make sure we don't clear anything
+            # before the actual `InterruptionTaskFrame` interruption.
+            InterruptionFrame(),
+            # This will generate an `InterruptionTaskFrame` and will wait for an
+            # `InterruptionFrame`.
+            TextFrame(text="Hello from Pipecat!"),
+            # Just give time for everything to complete.
+            SleepFrame(sleep=0.5),
+            EndFrame(),
+        ]
+        expected_down_frames = [
+            InterruptionFrame,
+            InterruptionFrame,
+            TransportMessageUrgentFrame,
+            EndFrame,
+        ]
+        await run_test(
+            pipeline,
+            frames_to_send=frames_to_send,
+            expected_down_frames=expected_down_frames,
+            send_end_frame=False,
+        )