diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index bf910a0c4..560a3dd67 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/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)): diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index 2230e9948..a77a0f837 100644 --- a/src/pipecat/services/tts_service.py +++ b/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] diff --git a/tests/test_context_aggregators_universal.py b/tests/test_context_aggregators_universal.py index 4cd195bee..7065e2ef3 100644 --- a/tests/test_context_aggregators_universal.py +++ b/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 ( diff --git a/tests/test_tts_frame_ordering.py b/tests/test_tts_frame_ordering.py index acc1cd2fd..70d2a0323 100644 --- a/tests/test_tts_frame_ordering.py +++ b/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()