PipelineTask: set finished_event InterruptionFrame is received

CHANGELOG.md

@@ -9,6 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added an asyncio event `finished_event` field to `InterruptionFrame`. When
+  assigned, the asyncio event will be set when the frame reaches the end of the
+  pipeline. You can use this field to know when an interruption made it all the
+  way to the end of a pipeline. The field has been also added to
+  `InterruptionTaskFrame`.
+
 - Added support for loading external observers. You can now register custom
   pipeline observers by setting the `PIPECAT_OBSERVER_FILES` environment
   variable. This variable should contain a colon-separated list of Python files
@@ -28,8 +34,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - `CancelFrame` and `CancelTaskFrame` have an optional `reason` field to
   indicate why the pipeline is being canceled. This can be also specified when
-  you cancel a task with `PipelineTask.cancel(reason="cancellation your
-reason")`.
+  you cancel a task with `PipelineTask.cancel(reason="cancellation reason")`.
 
 - Added `include_prob_metrics` parameter to Whisper STT services to enable access
   to probability metrics from transcription results.

src/pipecat/frames/frames.py

@@ -862,7 +862,7 @@ class InterruptionFrame(SystemFrame):
     with other frames (so the order is not guaranteed).
 
     Parameters:
-        wait_event: If not None, the event will be set when the frame
+        finished_event: If not None, the event will be set when the frame
         reaches the end of the pipeline.
     """
 

src/pipecat/pipeline/task.py

@@ -745,7 +745,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(finished_event=frame.finished_event))
         elif isinstance(frame, ErrorFrame):
             await self._call_event_handler("on_pipeline_error", frame)
             if frame.fatal:
@@ -786,6 +786,10 @@ class PipelineTask(BasePipelineTask):
             self._pipeline_end_event.set()
         elif isinstance(frame, HeartbeatFrame):
             await self._heartbeat_queue.put(frame)
+        elif isinstance(frame, InterruptionFrame) and frame.finished_event:
+            # This should unblock any code waiting for the interruption to
+            # complete.
+            frame.finished_event.set()
 
     async def _heartbeat_push_handler(self):
         """Push heartbeat frames at regular intervals."""