Merge pull request #2796 from pipecat-ai/aleix/fix-interruption-strategies-context-order

fix context order when using interruption strategies

CHANGELOG.md

@@ -27,6 +27,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed an issue that would cause wrong user/assistant context ordering when
+  using interruption strategies.
+
 - Fixed RTVI incoming message handling, broken in 0.0.87.
 
 ## [0.0.87] - 2025-10-02

src/pipecat/frames/frames.py

@@ -849,13 +849,15 @@ class FrameProcessorResumeUrgentFrame(SystemFrame):
 class InterruptionFrame(SystemFrame):
     """Frame indicating user started speaking (interruption detected).
 
-    Emitted by the BaseInputTransport to indicate that a user has started
+    Usually emitted by the pipeline task to indicate that all frame processors
-    speaking (i.e. is interrupting). This is similar to
+    should be interrrupted.
-    UserStartedSpeakingFrame except that it should be pushed concurrently
+
-    with other frames (so the order is not guaranteed).
+    Parameters:
+        pushed_by_task: Whether this interruption was pushed from the pipeline
+        task.
     """
 
-    pass
+    pushed_by_task: bool = False
 
 
 @dataclass
@@ -886,6 +888,17 @@ class StartInterruptionFrame(InterruptionFrame):
             )
 
 
+@dataclass
+class InterruptionCompletedFrame(SystemFrame):
+    """Frame indicating that the whole pipeline has been interrupted.
+
+    This is emitted by the pipeline task when an InterruptionFrame has made it
+    all the way to the end of pipeline, interrupting all the frame processors.
+    """
+
+    pass
+
+
 @dataclass
 class UserStartedSpeakingFrame(SystemFrame):
     """Frame indicating user has started speaking.

src/pipecat/pipeline/task.py

@@ -30,6 +30,7 @@ from pipecat.frames.frames import (
     ErrorFrame,
     Frame,
     HeartbeatFrame,
+    InterruptionCompletedFrame,
     InterruptionFrame,
     InterruptionTaskFrame,
     MetricsFrame,
@@ -691,8 +692,11 @@ class PipelineTask(BasePipelineTask):
             # bypassing the push queue and directly queue into the
             # 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}")
+            interruption_frame = InterruptionFrame(pushed_by_task=True)
-            await self._pipeline.queue_frame(InterruptionFrame())
+            logger.debug(
+                f"{self}: received interruption task frame {frame}, pushing {interruption_frame}"
+            )
+            await self._pipeline.queue_frame(interruption_frame)
         elif isinstance(frame, ErrorFrame):
             if frame.fatal:
                 logger.error(f"A fatal error occurred: {frame}")
@@ -736,6 +740,14 @@ class PipelineTask(BasePipelineTask):
             self._pipeline_end_event.set()
         elif isinstance(frame, CancelFrame):
             self._pipeline_end_event.set()
+        elif isinstance(frame, InterruptionFrame) and frame.pushed_by_task:
+            # If an interruption frame made it all the way to the end of the
+            # pipeline, send an InterruptionCompleteFrame. Note that we are
+            # bypassing the push queue and directly queue into the
+            # 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}: interruption completed with {frame}")
+            await self._pipeline.queue_frame(InterruptionCompletedFrame())
         elif isinstance(frame, HeartbeatFrame):
             await self._heartbeat_queue.put(frame)
 

src/pipecat/processors/frame_processor.py

@@ -28,6 +28,7 @@ from pipecat.frames.frames import (
     FrameProcessorPauseUrgentFrame,
     FrameProcessorResumeFrame,
     FrameProcessorResumeUrgentFrame,
+    InterruptionCompletedFrame,
     InterruptionFrame,
     InterruptionTaskFrame,
     StartFrame,
@@ -571,7 +572,9 @@ class FrameProcessor(BaseObject):
         # 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 self._wait_for_interruption and isinstance(frame, InterruptionFrame):
+        if self._wait_for_interruption and isinstance(
+            frame, (InterruptionFrame, InterruptionCompletedFrame)
+        ):
             await self.__process_frame(frame, direction, callback)
             return
 
@@ -661,18 +664,17 @@ 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
+        # If we are waiting for an interruption and one completed, then we can
-        # we can unblock `push_interruption_task_frame_and_wait()`.
+        # unblock `push_interruption_task_frame_and_wait()`.
-        if self._wait_for_interruption and isinstance(frame, InterruptionFrame):
+        if self._wait_for_interruption and isinstance(frame, InterruptionCompletedFrame):
             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
+        task and waits to receive an `InterruptionCompletedFrame`. This
-        the function finishes it is guaranteed that the `InterruptionFrame` has
+        guarantees the whole pipeline has been interrupted.
-        been pushed downstream.
         """
         self._wait_for_interruption = True
 

tests/test_context_aggregators.py

@@ -5,6 +5,7 @@
 #
 
 import json
+import sys
 import unittest
 from typing import Any
 
@@ -20,6 +21,7 @@ from pipecat.frames.frames import (
     FunctionCallResultFrame,
     FunctionCallResultProperties,
     InterimTranscriptionFrame,
+    InterruptionCompletedFrame,
     InterruptionFrame,
     InterruptionTaskFrame,
     LLMFullResponseEndFrame,
@@ -568,6 +570,7 @@ class BaseTestUserContextAggregator:
             BotStartedSpeakingFrame,
             UserStartedSpeakingFrame,
             InterruptionFrame,
+            InterruptionCompletedFrame,
             UserStoppedSpeakingFrame,
             *self.EXPECTED_CONTEXT_FRAMES,
         ]

tests/test_dtmf_aggregator.py

@@ -10,6 +10,7 @@ from pipecat.audio.dtmf.types import KeypadEntry
 from pipecat.frames.frames import (
     EndFrame,
     InputDTMFFrame,
+    InterruptionCompletedFrame,
     InterruptionFrame,
     TranscriptionFrame,
 )
@@ -30,6 +31,7 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase):
         expected_down_frames = [
             InputDTMFFrame,
             InterruptionFrame,
+            InterruptionCompletedFrame,
             InputDTMFFrame,
             InputDTMFFrame,
             InputDTMFFrame,
@@ -62,10 +64,12 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase):
         expected_down_frames = [
             InputDTMFFrame,
             InterruptionFrame,
+            InterruptionCompletedFrame,
             InputDTMFFrame,
             TranscriptionFrame,  # First aggregation "12"
             InputDTMFFrame,
             InterruptionFrame,
+            InterruptionCompletedFrame,
             TranscriptionFrame,  # Second aggregation "3"
         ]
 
@@ -98,11 +102,13 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase):
         expected_down_frames = [
             InputDTMFFrame,
             InterruptionFrame,
+            InterruptionCompletedFrame,
             InputDTMFFrame,
             InputDTMFFrame,
             TranscriptionFrame,  # "12#"
             InputDTMFFrame,
             InterruptionFrame,
+            InterruptionCompletedFrame,
             InputDTMFFrame,
             TranscriptionFrame,  # "45"
         ]
@@ -132,6 +138,7 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase):
         expected_down_frames = [
             InputDTMFFrame,
             InterruptionFrame,
+            InterruptionCompletedFrame,
             InputDTMFFrame,
             TranscriptionFrame,  # Should flush before EndFrame
             EndFrame,
@@ -160,6 +167,7 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase):
         expected_down_frames = [
             InputDTMFFrame,
             InterruptionFrame,
+            InterruptionCompletedFrame,
             InputDTMFFrame,
             TranscriptionFrame,
         ]
@@ -187,6 +195,7 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase):
         expected_down_frames = [
             InputDTMFFrame,
             InterruptionFrame,
+            InterruptionCompletedFrame,
             InputDTMFFrame,
             InputDTMFFrame,
             TranscriptionFrame,
@@ -224,7 +233,7 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase):
 
         # All the InputDTMFFrames plus one TranscriptionFrame
         expected_down_frames = (
-            [InputDTMFFrame, InterruptionFrame]
+            [InputDTMFFrame, InterruptionFrame, InterruptionCompletedFrame]
             + [InputDTMFFrame] * (len(frames_to_send) - 1)
             + [TranscriptionFrame]
         )

tests/test_frame_processor.py

@@ -10,6 +10,7 @@ import unittest
 from pipecat.frames.frames import (
     EndFrame,
     Frame,
+    InterruptionCompletedFrame,
     InterruptionFrame,
     OutputTransportMessageUrgentFrame,
     TextFrame,
@@ -101,6 +102,7 @@ class TestFrameProcessor(unittest.IsolatedAsyncioTestCase):
         expected_down_frames = [
             InterruptionFrame,
             InterruptionFrame,
+            InterruptionCompletedFrame,
             OutputTransportMessageUrgentFrame,
             EndFrame,
         ]