diff --git a/CHANGELOG.md b/CHANGELOG.md index f596c3462..c2a6660a5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -53,6 +53,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 an issue where the `SmallWebRTCRequest` dataclass in runner would scrub arbitrary request data from client due to camelCase typing. This fixes data passthrough for JS clients where `APIRequest` is used. diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 1ca3333b5..225629ad9 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -93,7 +93,7 @@ class FrameProcessorQueue(asyncio.PriorityQueue): self.__high_counter = 0 self.__low_counter = 0 - async def put(self, item: Tuple[Frame, FrameDirection, FrameCallback]): + async def put(self, item: Tuple[Frame, FrameDirection, Optional[FrameCallback]]): """Put an item into the priority queue. System frames (`SystemFrame`) have higher priority than any other @@ -228,11 +228,9 @@ class FrameProcessor(BaseObject): # 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. + # start to end of the pipeline. When it reaches the end we will be + # notified through the assigned event. self._wait_for_interruption = False - self._wait_interruption_event = asyncio.Event() # Frame processor events. self._register_event_handler("on_before_process_frame", sync=True) @@ -567,10 +565,6 @@ class FrameProcessor(BaseObject): if self._cancelling: return - # If we are waiting for an interruption we will bypass all queued system - # 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): await self.__process_frame(frame, direction, callback) return @@ -661,31 +655,27 @@ 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 - # 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): - """Push an interruption task frame upstream and wait for the interruption. + """Interrupt the pipeline and wait for the interruption to complete. + + This function sends an `InterruptionTaskFrame` upstream with an + associated asyncio event. It then waits for the generated + `InterruptionFrame` to reach the end of the pipeline where the event + will be set. - This function sends an `InterruptionTaskFrame` upstream to the pipeline - task and waits to receive the corresponding `InterruptionFrame`. When - the function finishes it is guaranteed that the `InterruptionFrame` has - been pushed downstream. """ self._wait_for_interruption = True - await self.push_frame(InterruptionTaskFrame(), FrameDirection.UPSTREAM) + finished_event = asyncio.Event() - # Wait for an `InterruptionFrame` to come to this processor and be - # pushed. Take a look at `push_frame()` to see how we first push the - # `InterruptionFrame` and then we set the event in order to maintain - # frame ordering. - await self._wait_interruption_event.wait() + await self.push_frame( + InterruptionTaskFrame(finished_event=finished_event), FrameDirection.UPSTREAM + ) - # Clean the event. - self._wait_interruption_event.clear() + # Wait for the event to be set. This event is set when the + # `InterruptionFrame` pushed by the pipeline task reaches the end of the + # pipeline. + await finished_event.wait() self._wait_for_interruption = False