Pair force-completed text with a progress frame, drop unmatched buffered words

changelog/5681.fixed.2.md

@@ -0,0 +1 @@
+Fixed a word-timestamp event that belongs to no sentence being discarded with nothing logged, in streaming mode (a TTS service using `TextAggregationMode.TOKEN`). This comes up when a provider reports a punctuation mark as its own event after the preceding word already took it. Such an event used to sit in a buffer until the next turn began and was thrown away there; it is now logged and dropped when its own turn ends.

changelog/5681.fixed.md

@@ -0,0 +1 @@
+Fixed `AggregatedFrameSequencer.force_complete` emitting the text it force-completes as a `TTSTextFrame` without a matching `AggregatedTextProgressFrame`, so the progress view stopped where the TTS provider stopped reporting words while the word frames carried the rest of the turn.

src/pipecat/utils/context/aggregated_frame_sequencer.py

@@ -639,10 +639,13 @@ def force_complete(self, context_id: str, last_word_pts: int) -> list[Frame]:
                 force-completed frames and forwarded to :meth:`flush`.
 
         Returns:
-            Combined list of TTSTextFrames (for incomplete spoken slots) and
+            Combined list of TTSTextFrames (for incomplete spoken slots),
+            AggregatedTextProgressFrames pairing the forced text, and
             AggregatedTextFrames (skipped slots now unblocked), in emission order.
         """
         frames: list[Frame] = []
+        if self._streaming:
+            self._discard_buffered_words(context_id)
         for slot in self._slots:
             if slot.spoken and not slot.complete and slot.context_id == context_id:
                 if slot.tracker:
@@ -668,6 +671,10 @@ def force_complete(self, context_id: str, last_word_pts: int) -> list[Frame]:
                                 includes_inter_frame_spaces=slot.includes_inter_frame_spaces,
                             )
                         )
+                        # That frame carries the rest of the text, so the progress view
+                        # has to reach the end alongside it.
+                        slot.tracker.take_remaining_as_spoken()
+                        frames.append(self._build_progress_frame(slot, last_word_pts))
                 slot.complete = True
         frames.extend(self.flush(last_word_pts=last_word_pts))
         # Context is fully done: forget it so any later word is dropped as stale.
@@ -811,6 +818,22 @@ def _drain_buffered_words(self) -> list[Frame]:
             )
         return frames
 
+    def _discard_buffered_words(self, context_id: str) -> None:
+        """Drop this context's still-buffered words, which no slot ever matched.
+
+        Sentence mode drops an unrecognised word where it arrives; streaming only knows
+        one will never match once the context it belongs to has ended.
+        """
+        keep: list[_BufferedWord] = []
+        for w in self._buffered_words:
+            if w.context_id != context_id:
+                keep.append(w)
+                continue
+            logger.warning(
+                f"{self._name} Dropping buffered word '{w.word}' not recognised by any slot."
+            )
+        self._buffered_words = keep
+
     def _slot_matches_context(self, slot: _AggregatedFrameSlot, context_id: str | None) -> bool:
         """Whether *slot* is an eligible active slot for *context_id*.
 

src/pipecat/utils/context/word_completion_tracker.py

@@ -223,6 +223,16 @@ def _force_complete(self, word: str) -> bool:
         self._overflow_word = word
         return True
 
+    def take_remaining_as_spoken(self) -> None:
+        """Move the cursors to the end, for a caller that emits the remainder itself.
+
+        The accumulated and remaining views then agree with the text that went out.
+        """
+        self._user_facing_pos = len(self._user_facing_text)
+        if self._llm_text is not None:
+            self._llm_pos = len(self._llm_text)
+            self._llm_spoken_pos = len(self._llm_text)
+
     def _record_llm_span(self, word: str, llm_pos_before: int, spoken_before: int) -> None:
         """Record which part of ``llm_text`` the word just added stands for.
 

tests/test_aggregated_frame_sequencer.py

@@ -2223,5 +2223,51 @@ async def test_slot_completes_after_the_resync(self):
         self.assertEqual(seq._slots, [])
 
 
+class TestForceCompleteWordStream(unittest.IsolatedAsyncioTestCase):
+    """What a context ending has to account for: text no word arrived for, which the
+    progress view has to reach, and a buffered word no slot ever matched, which is
+    dropped where sentence mode would have dropped it on arrival.
+    """
+
+    async def _streamed_slot(self, *tokens: str) -> AggregatedFrameSequencer:
+        seq = _seq(streaming=True)
+        for t in tokens:
+            await seq.register_spoken(_spoken_frame(t, raw_text=t), "ctx1", t, True)
+        return seq
+
+    async def test_buffered_word_is_dropped_at_context_end(self):
+        # "。" is taken by "好" as trailing punctuation, so the slot is already
+        # complete when the provider reports the mark on its own, so it is buffered.
+        seq = await self._streamed_slot("您好", "。", "谢谢")
+        for i, ch in enumerate(["您", "好", "。"]):
+            seq.process_word(ch, pts=(i + 1) * 10, context_id="ctx1")
+        self.assertEqual([w.word for w in seq._buffered_words], ["。"])
+
+        words = [f for f in seq.force_complete("ctx1", 30) if isinstance(f, TTSTextFrame)]
+        self.assertEqual(words, [])
+        self.assertEqual(seq._buffered_words, [])
+
+    async def test_buffered_word_for_another_context_is_left_alone(self):
+        seq = await self._streamed_slot("您好", "。", "谢谢")
+        for i, ch in enumerate(["您", "好", "。"]):
+            seq.process_word(ch, pts=(i + 1) * 10, context_id="ctx1")
+        seq._buffered_words[0].context_id = "ctx2"
+        seq.force_complete("ctx1", 30)
+        self.assertEqual([w.word for w in seq._buffered_words], ["。"])
+
+    async def test_forced_tail_reports_progress_to_the_end(self):
+        seq = _seq()
+        text = "Hello there friend"
+        await seq.register_spoken(_spoken_frame(text, raw_text=text), "ctx1", text, True)
+        seq.process_word("Hello", pts=10, context_id="ctx1")
+
+        frames = seq.force_complete("ctx1", 20)
+        words = [f for f in frames if isinstance(f, TTSTextFrame)]
+        progress = [f for f in frames if isinstance(f, AggregatedTextProgressFrame)]
+        self.assertEqual([f.text for f in words], ["there friend"])
+        self.assertEqual(progress[-1].accumulated_text, text)
+        self.assertEqual(progress[-1].remaining_text, "")
+
+
 if __name__ == "__main__":
     unittest.main()