Merge pull request #3789 from pipecat-ai/aleix/fix-missing-await-and-interruption-hang

Fix missing await and interruption timeout hang

changelog/3789.fixed.md

@@ -0,0 +1 @@
+- Fixed `push_interruption_task_frame_and_wait()` hanging indefinitely when the `InterruptionFrame` does not reach the pipeline sink within the timeout. Added a `timeout` keyword argument to customize the wait duration.

examples/foundational/07s-interruptible-google-audio-in.py

@@ -96,7 +96,7 @@ class UserAudioCollector(FrameProcessor):
             self._user_speaking = True
         elif isinstance(frame, UserStoppedSpeakingFrame):
             self._user_speaking = False
-            self._context.add_audio_frames_message(audio_frames=self._audio_frames)
+            await self._context.add_audio_frames_message(audio_frames=self._audio_frames)
             await self._user_context_aggregator.push_frame(LLMRunFrame())
 
         elif isinstance(frame, InputAudioRawFrame):

examples/foundational/25-google-audio-in.py

@@ -98,7 +98,7 @@ class UserAudioCollector(FrameProcessor):
             self._user_speaking = True
         elif isinstance(frame, UserStoppedSpeakingFrame):
             self._user_speaking = False
-            self._context.add_audio_frames_message(audio_frames=self._audio_frames)
+            await self._context.add_audio_frames_message(audio_frames=self._audio_frames)
             await self._user_context_aggregator.push_frame(LLMContextFrame(context=self._context))
         elif isinstance(frame, InputAudioRawFrame):
             if self._user_speaking:

src/pipecat/processors/frame_processor.py

@@ -52,8 +52,6 @@ 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.
@@ -763,7 +761,7 @@ class FrameProcessor(BaseObject):
 
         await self._call_event_handler("on_after_push_frame", frame)
 
-    async def push_interruption_task_frame_and_wait(self):
+    async def push_interruption_task_frame_and_wait(self, *, timeout: float = 5.0):
         """Push an interruption task frame upstream and wait for the interruption.
 
         This function sends an `InterruptionTaskFrame` upstream to the
@@ -772,9 +770,11 @@ class FrameProcessor(BaseObject):
         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
+        not complete within the given timeout, a warning is logged and the
-        warning is logged periodically until it completes.
+        event is forcibly set so the caller is unblocked.
 
+        Args:
+            timeout: Maximum seconds to wait for the interruption to complete.
         """
         self._wait_for_interruption = True
 
@@ -782,19 +782,20 @@ class FrameProcessor(BaseObject):
 
         await self.push_frame(InterruptionTaskFrame(event=event), FrameDirection.UPSTREAM)
 
-        # Wait for the `InterruptionFrame` to complete and log a warning
+        # Wait for the `InterruptionFrame` to complete and log a warning if it
-        # periodically if it takes too long.
+        # takes too long. If it does take too long make sure we unblock it,
+        # otherwise we will hang here forever.
         while not event.is_set():
             try:
-                await asyncio.wait_for(event.wait(), timeout=INTERRUPTION_COMPLETION_TIMEOUT)
+                await asyncio.wait_for(event.wait(), timeout=timeout)
             except asyncio.TimeoutError:
                 logger.warning(
                     f"{self}: InterruptionFrame has not completed after"
-                    f" {INTERRUPTION_COMPLETION_TIMEOUT}s. Make sure"
+                    f" {timeout}s. Make sure InterruptionFrame.complete()"
-                    " InterruptionFrame.complete() is being called (e.g. if the"
+                    " is being called (e.g. if the frame is being blocked"
-                    " frame is being blocked or consumed before reaching the"
+                    " or consumed before reaching the pipeline sink)."
-                    " pipeline sink)."
                 )
+                event.set()
 
         self._wait_for_interruption = False
 

tests/test_frame_processor.py

@@ -25,7 +25,6 @@ from pipecat.frames.frames import (
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.processors.filters.identity_filter import IdentityFilter
 from pipecat.processors.frame_processor import (
-    INTERRUPTION_COMPLETION_TIMEOUT,
     FrameDirection,
     FrameProcessor,
 )
@@ -521,7 +520,7 @@ class TestFrameProcessor(unittest.IsolatedAsyncioTestCase):
                         # Complete after the timeout so the warning fires
                         # but the test doesn't hang.
                         async def delayed_complete():
-                            await asyncio.sleep(INTERRUPTION_COMPLETION_TIMEOUT + 1.0)
+                            await asyncio.sleep(1.0)
                             frame.complete()
 
                         asyncio.create_task(delayed_complete())
@@ -532,7 +531,7 @@ class TestFrameProcessor(unittest.IsolatedAsyncioTestCase):
                 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_interruption_task_frame_and_wait(timeout=0.5)
                         await self.push_frame(OutputTransportMessageUrgentFrame(message="done"))
                     else:
                         await self.push_frame(frame, direction)