Merge pull request #323 from pipecat-ai/aleix/base-input-handle-incoming-interruptions

transports(inputs): handle start/stop interruption frames

CHANGELOG.md

@@ -16,6 +16,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Fixed a `BaseInputTransport` issue that was causing incoming system frames to
   be queued instead of being pushed immediately.
 
+- Fixed a `BaseInputTransport` issue that was causing start/stop interruptions
+  incoming frames to not cancel tasks and be processed properly.
+
 ## [0.0.39] - 2024-07-23
 
 ### Fixed

src/pipecat/transports/base_input.py

@@ -77,6 +77,10 @@ class BaseInputTransport(FrameProcessor):
             await self.push_frame(frame, direction)
         elif isinstance(frame, BotInterruptionFrame):
             await self._handle_interruptions(frame, False)
+        elif isinstance(frame, StartInterruptionFrame):
+            await self._start_interruption()
+        elif isinstance(frame, StopInterruptionFrame):
+            await self._stop_interruption()
         # All other system frames
         elif isinstance(frame, SystemFrame):
             await self.push_frame(frame, direction)
@@ -120,6 +124,9 @@ class BaseInputTransport(FrameProcessor):
     #
 
     async def _start_interruption(self):
+        if not self.interruptions_allowed:
+            return
+
         # Cancel the task. This will stop pushing frames downstream.
         self._push_frame_task.cancel()
         await self._push_frame_task
@@ -131,6 +138,9 @@ class BaseInputTransport(FrameProcessor):
         self._create_push_task()
 
     async def _stop_interruption(self):
+        if not self.interruptions_allowed:
+            return
+
         await self.push_frame(StopInterruptionFrame())
 
     async def _handle_interruptions(self, frame: Frame, push_frame: bool):