Merge pull request #4414 from pipecat-ai/mb/fix-ttsspeakframe-assistant-turn-stopped

changelog/4414.fixed.md

@@ -0,0 +1 @@
+- Fixed `TTSSpeakFrame(append_to_context=True)` greetings sometimes splitting across two assistant messages in the LLM context and not surfacing in `on_assistant_turn_stopped`. The `LLMAssistantPushAggregationFrame` emitted at the end of a TTS context now carries a PTS just past the last word so it can't overtake clock-queued `TTSTextFrame`s in the transport's output, and `LLMAssistantAggregator` now triggers `on_assistant_turn_started`/`on_assistant_turn_stopped` when it receives the frame outside an LLM response cycle (restoring v0.0.104 behavior for greeting transcripts).

src/pipecat/processors/aggregators/llm_response_universal.py

@@ -927,7 +927,7 @@ class LLMAssistantAggregator(LLMContextAggregator):
             await self._handle_end_or_cancel(frame)
             await self.push_frame(frame, direction)
         elif isinstance(frame, LLMAssistantPushAggregationFrame):
-            await self.push_aggregation()
+            await self._handle_push_aggregation()
         elif isinstance(frame, LLMFullResponseStartFrame):
             await self._handle_llm_start(frame)
         elif isinstance(frame, LLMFullResponseEndFrame):
@@ -1309,6 +1309,17 @@ class LLMAssistantAggregator(LLMContextAggregator):
     async def _handle_llm_end(self, _: LLMFullResponseEndFrame):
         await self._trigger_assistant_turn_stopped()
 
+    async def _handle_push_aggregation(self):
+        # LLMAssistantPushAggregationFrame is emitted by TTSService at the end
+        # of a TTSSpeakFrame-driven utterance (no surrounding LLM response
+        # cycle), so no LLMFullResponseStartFrame ever set the turn-start
+        # timestamp. Open a turn now so on_assistant_turn_stopped fires for the
+        # greeting text the same way it did before LLMAssistantPushAggregationFrame
+        # was introduced.
+        if not self._assistant_turn_start_timestamp:
+            await self._trigger_assistant_turn_started()
+        await self._trigger_assistant_turn_stopped()
+
     async def _handle_text(self, frame: TextFrame):
         # Skip TextFrame types not intended to build the assistant context
         if isinstance(frame, (TranscriptionFrame, TranslationFrame, InterimTranscriptionFrame)):

src/pipecat/services/tts_service.py

@@ -771,7 +771,17 @@ class TTSService(AIService):
         if isinstance(frame, TTSStoppedFrame) and frame.context_id:
             if frame.context_id in self._tts_contexts:
                 if self._tts_contexts[frame.context_id].push_assistant_aggregation:
-                    await self.push_frame(LLMAssistantPushAggregationFrame())
+                    aggregation_frame = LLMAssistantPushAggregationFrame()
+                    # When word-level TTSTextFrames are routed through the
+                    # transport's clock queue (PTS-based), the aggregation frame
+                    # would otherwise take the audio (sync) queue path and
+                    # could overtake the final word frames. Stamping it with a
+                    # PTS just past the last word forces it through the clock
+                    # queue too, so the assistant aggregator sees every word
+                    # before flushing.
+                    if self._word_last_pts:
+                        aggregation_frame.pts = self._word_last_pts + 1
+                    await self.push_frame(aggregation_frame)
                 logger.debug(f"{self} cleaning up TTS context {frame.context_id}")
                 del self._tts_contexts[frame.context_id]
 

tests/test_context_aggregators_universal.py

@@ -17,6 +17,7 @@ from pipecat.frames.frames import (
     FunctionCallsStartedFrame,
     InterimTranscriptionFrame,
     InterruptionFrame,
+    LLMAssistantPushAggregationFrame,
     LLMContextAssistantTimestampFrame,
     LLMContextFrame,
     LLMFullResponseEndFrame,
@@ -34,6 +35,7 @@ from pipecat.frames.frames import (
     TextFrame,
     TranscriptionFrame,
     TranslationFrame,
+    TTSTextFrame,
     UserMuteStartedFrame,
     UserStartedSpeakingFrame,
     UserStoppedSpeakingFrame,
@@ -58,6 +60,7 @@ from pipecat.turns.user_mute import (
 )
 from pipecat.turns.user_stop import SpeechTimeoutUserTurnStopStrategy
 from pipecat.turns.user_turn_strategies import UserTurnStrategies
+from pipecat.utils.text.base_text_aggregator import AggregationType
 
 USER_TURN_STOP_TIMEOUT = 0.2
 TRANSCRIPTION_TIMEOUT = 0.1
@@ -975,6 +978,83 @@ class TestLLMAssistantAggregator(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(len(stop_messages), 1)
         self.assertEqual(stop_messages[0].content, "Hello from Pipecat!")
 
+    async def test_push_aggregation_fires_turn_stopped_for_tts_speak(self):
+        """LLMAssistantPushAggregationFrame must fire on_assistant_turn_stopped.
+
+        Mirrors the TTSSpeakFrame(append_to_context=True) greeting flow: TTS-driven
+        TTSTextFrames accumulate without an LLMFullResponseStartFrame, then the
+        TTS service emits LLMAssistantPushAggregationFrame to commit them.
+        """
+        context = LLMContext()
+        aggregator = LLMAssistantAggregator(context)
+
+        start_count = 0
+        stop_messages = []
+
+        @aggregator.event_handler("on_assistant_turn_started")
+        async def on_assistant_turn_started(aggregator):
+            nonlocal start_count
+            start_count += 1
+
+        @aggregator.event_handler("on_assistant_turn_stopped")
+        async def on_assistant_turn_stopped(aggregator, message: AssistantTurnStoppedMessage):
+            stop_messages.append(message)
+
+        frames_to_send = [
+            TTSTextFrame("Hello,", aggregated_by=AggregationType.WORD),
+            TTSTextFrame("how", aggregated_by=AggregationType.WORD),
+            TTSTextFrame("can I help?", aggregated_by=AggregationType.WORD),
+            LLMAssistantPushAggregationFrame(),
+        ]
+        expected_down_frames = [LLMContextFrame, LLMContextAssistantTimestampFrame]
+        await run_test(
+            aggregator,
+            frames_to_send=frames_to_send,
+            expected_down_frames=expected_down_frames,
+        )
+        self.assertEqual(start_count, 1)
+        self.assertEqual(len(stop_messages), 1)
+        self.assertFalse(stop_messages[0].interrupted)
+        self.assertEqual(stop_messages[0].content, "Hello, how can I help?")
+        self.assertEqual(
+            context.messages[-1],
+            {"role": "assistant", "content": "Hello, how can I help?"},
+        )
+
+    async def test_push_aggregation_does_not_double_fire_in_llm_response(self):
+        """LLMAssistantPushAggregationFrame mid-response must not double-fire turn events.
+
+        Inside an LLMFullResponseStart/End cycle, a stray LLMAssistantPushAggregationFrame
+        should flush whatever is buffered and consume the active turn (firing exactly
+        one stopped event). The closing LLMFullResponseEndFrame then has no pending
+        turn to stop.
+        """
+        context = LLMContext()
+        aggregator = LLMAssistantAggregator(context)
+
+        start_count = 0
+        stop_messages = []
+
+        @aggregator.event_handler("on_assistant_turn_started")
+        async def on_assistant_turn_started(aggregator):
+            nonlocal start_count
+            start_count += 1
+
+        @aggregator.event_handler("on_assistant_turn_stopped")
+        async def on_assistant_turn_stopped(aggregator, message: AssistantTurnStoppedMessage):
+            stop_messages.append(message)
+
+        frames_to_send = [
+            LLMFullResponseStartFrame(),
+            LLMTextFrame("Hello!"),
+            LLMAssistantPushAggregationFrame(),
+            LLMFullResponseEndFrame(),
+        ]
+        await run_test(aggregator, frames_to_send=frames_to_send)
+        self.assertEqual(start_count, 1)
+        self.assertEqual(len(stop_messages), 1)
+        self.assertEqual(stop_messages[0].content, "Hello!")
+
     async def test_turn_completion_markers_stripped_from_transcript(self):
         """Turn completion markers should be stripped from assistant transcript."""
         from pipecat.turns.user_turn_completion_mixin import (

tests/test_tts_frame_ordering.py

@@ -33,6 +33,7 @@ from pipecat.frames.frames import (
     AggregatedTextFrame,
     DataFrame,
     Frame,
+    LLMAssistantPushAggregationFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     TextFrame,
@@ -657,5 +658,55 @@ async def test_websocket_word_timestamps_punctuation_tokens():
     assert all(f.includes_inter_frame_spaces is True for f in text_frames)
 
 
+@pytest.mark.asyncio
+async def test_push_aggregation_pts_after_last_word():
+    """LLMAssistantPushAggregationFrame must carry PTS > last TTSTextFrame PTS.
+
+    Without a PTS the aggregation frame routes through the transport's audio
+    (sync) queue while word-level TTSTextFrames go through the clock queue, so
+    it can overtake the last words and leave the trailing text orphaned in the
+    aggregator buffer (issue #4264).
+    """
+    word_times = [("hello", 0.0), ("world", 0.2)]
+    tts = _MockWordTimestampHttpTTSService(word_times=word_times)
+    frames_received = await run_test(
+        tts,
+        frames_to_send=[TTSSpeakFrame(text="hello world", append_to_context=True)],
+    )
+    down = frames_received[0]
+    text_frames = [f for f in down if isinstance(f, TTSTextFrame)]
+    push_frames = [f for f in down if isinstance(f, LLMAssistantPushAggregationFrame)]
+
+    assert len(push_frames) == 1, (
+        f"Expected exactly one LLMAssistantPushAggregationFrame, got {len(push_frames)}"
+    )
+    assert text_frames, "Expected TTSTextFrames to be emitted"
+
+    last_word_pts = max(f.pts for f in text_frames)
+    assert push_frames[0].pts is not None and push_frames[0].pts > last_word_pts, (
+        f"LLMAssistantPushAggregationFrame.pts ({push_frames[0].pts}) must exceed "
+        f"the last TTSTextFrame PTS ({last_word_pts}) so it can't overtake it in the "
+        f"transport's clock queue"
+    )
+
+
+@pytest.mark.asyncio
+async def test_push_aggregation_no_pts_without_word_timestamps():
+    """Aggregation frame stays unstamped when no word timestamps were emitted.
+
+    Without word frames, both the aggregation frame and any other downstream
+    frames travel through the transport's sync queue in order, so adding a PTS
+    would needlessly route it through the clock queue.
+    """
+    tts = MockHttpTTSService()
+    frames_received = await run_test(
+        tts,
+        frames_to_send=[TTSSpeakFrame(text="hello world", append_to_context=True)],
+    )
+    push_frames = [f for f in frames_received[0] if isinstance(f, LLMAssistantPushAggregationFrame)]
+    assert len(push_frames) == 1
+    assert push_frames[0].pts is None or push_frames[0].pts == 0
+
+
 if __name__ == "__main__":
     unittest.main()