fix: preserve queued transcriptions across interruptions

changelog/5702.fixed.md

@@ -0,0 +1 @@
+- Final user transcriptions queued during turn-start interruptions are now preserved, preventing batched STT results or simultaneous speech and DTMF input from being dropped.

src/pipecat/frames/frames.py

@@ -447,11 +447,13 @@ class AggregatedTextProgressFrame(DataFrame):
 
 
 @dataclass
-class TranscriptionFrame(TextFrame):
+class TranscriptionFrame(TextFrame, UninterruptibleFrame):
     """Text frame containing speech transcription data.
 
     A text frame with transcription-specific data. The `result` field
-    contains the result from the STT service if available.
+    contains the result from the STT service if available. Transcription
+    frames are uninterruptible so received user input is preserved when a
+    user turn interrupts queued pipeline work.
 
     Parameters:
         user_id: Identifier for the user who spoke.

tests/test_context_aggregators_universal.py

@@ -98,6 +98,48 @@
 
 
 class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
+    async def test_queued_transcriptions_survive_turn_start_interruption(self):
+        async def committed_user_messages(frames: list[Frame], *, use_interim: bool) -> list[str]:
+            context = LLMContext()
+            user_aggregator = LLMUserAggregator(
+                context,
+                params=LLMUserAggregatorParams(
+                    user_turn_strategies=UserTurnStrategies(
+                        start=[TranscriptionUserTurnStartStrategy(use_interim=use_interim)],
+                        stop=[
+                            SpeechTimeoutUserTurnStopStrategy(
+                                user_speech_timeout=TRANSCRIPTION_TIMEOUT
+                            )
+                        ],
+                    )
+                ),
+            )
+            await run_test(user_aggregator, frames_to_send=frames)
+            return [
+                message["content"]
+                for message in context.get_messages()
+                if message.get("role") == "user"
+            ]
+
+        timestamp = "2026-01-01T00:00:00Z"
+        interim_then_final = await committed_user_messages(
+            [
+                InterimTranscriptionFrame(text="what is", user_id="user", timestamp=timestamp),
+                TranscriptionFrame(text="what is my balance", user_id="user", timestamp=timestamp),
+            ],
+            use_interim=True,
+        )
+        consecutive_finals = await committed_user_messages(
+            [
+                TranscriptionFrame(text="one", user_id="user", timestamp=timestamp),
+                TranscriptionFrame(text="two", user_id="user", timestamp=timestamp),
+            ],
+            use_interim=False,
+        )
+
+        self.assertEqual(interim_then_final, ["what is my balance"])
+        self.assertEqual(consecutive_finals, ["one two"])
+
     async def test_llm_run(self):
         context = LLMContext()
 

tests/test_dtmf_aggregator.py

@@ -13,11 +13,36 @@
     InterruptionFrame,
     TranscriptionFrame,
 )
+from pipecat.pipeline.pipeline import Pipeline
 from pipecat.processors.aggregators.dtmf_aggregator import DTMFAggregator
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import LLMUserAggregator
 from pipecat.tests.utils import SleepFrame, run_test
 
 
 class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase):
+    async def test_transcription_overlap_survives_dtmf_interruption(self):
+        context = LLMContext()
+        pipeline = Pipeline([DTMFAggregator(), LLMUserAggregator(context)])
+
+        await run_test(
+            pipeline,
+            frames_to_send=[
+                TranscriptionFrame(text="what is my balance", user_id="user", timestamp="now"),
+                InputDTMFFrame(button=KeypadEntry.ONE),
+                InputDTMFFrame(button=KeypadEntry.POUND),
+            ],
+        )
+
+        user_messages = [
+            message["content"]
+            for message in context.get_messages()
+            if message.get("role") == "user"
+        ]
+        self.assertEqual(len(user_messages), 1)
+        self.assertIn("what is my balance", user_messages[0])
+        self.assertIn("DTMF: 1#", user_messages[0])
+
     async def test_basic_aggregation_with_pound(self):
         """Test basic DTMF aggregation ending with pound key."""
         aggregator = DTMFAggregator()