transports: push EndFrame/CancelFrame before stopping push task

CHANGELOG.md

@@ -17,6 +17,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed an issue where `BaseInputTransport` and `BaseOutputTransport` where
+  stopping push tasks before pushing `EndFrame` frames.
+
 - Fixed an error closing local audio transports.
 
 - Fixed an issue with Deepgram TTS that was introduced in the previous release.

src/pipecat/transports/base_input.py

@@ -87,16 +87,16 @@ class BaseInputTransport(FrameProcessor):
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         if isinstance(frame, CancelFrame):
-            await self.stop()
             # We don't queue a CancelFrame since we want to stop ASAP.
             await self.push_frame(frame, direction)
+            await self.stop()
         elif isinstance(frame, StartFrame):
             self._allow_interruption = frame.allow_interruptions
             await self.start(frame)
             await self._internal_push_frame(frame, direction)
         elif isinstance(frame, EndFrame):
-            await self.stop()
             await self._internal_push_frame(frame, direction)
+            await self.stop()
         else:
             await self._internal_push_frame(frame, direction)
 

src/pipecat/transports/base_output.py

@@ -121,11 +121,11 @@ class BaseOutputTransport(FrameProcessor):
             self._sink_queue.put_nowait(frame)
         # EndFrame is managed in the queue handler.
         elif isinstance(frame, CancelFrame):
+            await self.push_frame(frame, direction)
             await self.stop()
-            await self.push_frame(frame, direction)
         elif isinstance(frame, StartInterruptionFrame) or isinstance(frame, StopInterruptionFrame):
-            await self._handle_interruptions(frame)
             await self.push_frame(frame, direction)
+            await self._handle_interruptions(frame)
         else:
             self._sink_queue.put_nowait(frame)