From 86315183888e98c500b2120659f7bd940ac4434e Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Fri, 15 May 2026 14:08:50 -0400 Subject: [PATCH] test: cover wait_for_transcript_to_end_user_turn=False aggregator behavior MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds five tests for the delayed-transcript flow on `LLMUserAggregator`: - basic flow: `on_user_turn_stopped` fires fast with empty content; `on_user_turn_message_finalized` fires later with the populated transcript; user message lands in context. - backstop with no transcript: backstop timer still finalizes the turn; message_finalized fires with empty content; no user message added to context. - next-turn precondition violation: a new VAD start fires while the previous turn is still pending; the previous turn is force-flushed before the new turn begins. - context-order with assistant response: paired aggregators with a late user transcript arriving before the assistant content streams; verifies the user message lands in context before the assistant message (the conversational-order invariant the design relies on). - strategy mutation: explicit start/stop strategies are mutated by the bundle — `TranscriptionUserTurnStartStrategy` is dropped from start, `wait_for_transcript=False` is flipped on the stop strategy that had it explicitly set to True. Tests patch `DEFAULT_TTFS_P99` to keep the backstop fast. --- tests/test_context_aggregators_universal.py | 296 ++++++++++++++++++++ 1 file changed, 296 insertions(+) diff --git a/tests/test_context_aggregators_universal.py b/tests/test_context_aggregators_universal.py index a8ecace9d..62eb46e66 100644 --- a/tests/test_context_aggregators_universal.py +++ b/tests/test_context_aggregators_universal.py @@ -763,6 +763,302 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase): user_messages = [m for m in context.get_messages() if m.get("role") == "user"] self.assertEqual([m["content"] for m in user_messages], ["I'm thinking", "about pizza"]) + async def test_no_wait_for_transcript_basic_flow(self): + """``wait_for_transcript_to_end_user_turn=False`` splits the lifecycle: + + - ``on_user_turn_stopped`` fires on the audible end of the turn, + with empty content (the transcript hasn't arrived yet). + - The late transcript is captured into ``_aggregation``. + - When the backstop timer fires (no further transcripts), + ``on_user_turn_message_finalized`` fires with the populated + transcript and the user message lands in context. + """ + from unittest.mock import patch + + from pipecat.processors.aggregators import llm_response_universal + + # Shrink the backstop so the test runs quickly. + with patch.object(llm_response_universal, "DEFAULT_TTFS_P99", TRANSCRIPTION_TIMEOUT): + context = LLMContext() + user_aggregator = LLMUserAggregator( + context, + params=LLMUserAggregatorParams( + user_turn_strategies=UserTurnStrategies( + stop=[ + SpeechTimeoutUserTurnStopStrategy( + user_speech_timeout=TRANSCRIPTION_TIMEOUT + ) + ], + ), + wait_for_transcript_to_end_user_turn=False, + ), + ) + + events: list[tuple[str, str]] = [] + + @user_aggregator.event_handler("on_user_turn_stopped") + async def on_stopped(aggregator, strategy, message): + events.append(("stopped", message.content)) + + @user_aggregator.event_handler("on_user_turn_message_finalized") + async def on_finalized(aggregator, strategy, message): + events.append(("finalized", message.content)) + + pipeline = Pipeline([user_aggregator]) + + frames_to_send = [ + VADUserStartedSpeakingFrame(), + SleepFrame(), + VADUserStoppedSpeakingFrame(), + # Let the user_speech_timeout fire so the strategy + # fires turn-stopped. + SleepFrame(sleep=TRANSCRIPTION_TIMEOUT + 0.05), + # Late transcript arrives after the audible stop. + TranscriptionFrame(text="Hello!", user_id="", timestamp="now"), + # Wait for the backstop to fire. + SleepFrame(sleep=TRANSCRIPTION_TIMEOUT + 0.05), + ] + await run_test(pipeline, frames_to_send=frames_to_send) + + # Two events fired in order: audible stop first (empty), + # message finalized later (populated). + self.assertEqual(events, [("stopped", ""), ("finalized", "Hello!")]) + + # Context contains the user message. + user_messages = [m for m in context.get_messages() if m.get("role") == "user"] + self.assertEqual([m["content"] for m in user_messages], ["Hello!"]) + + async def test_no_wait_for_transcript_backstop_no_transcript_arrives(self): + """When no transcript arrives, the backstop timer still finalizes + the turn — ``on_user_turn_message_finalized`` fires with empty + content and nothing is written to context. + """ + from unittest.mock import patch + + from pipecat.processors.aggregators import llm_response_universal + + with patch.object(llm_response_universal, "DEFAULT_TTFS_P99", TRANSCRIPTION_TIMEOUT): + context = LLMContext() + user_aggregator = LLMUserAggregator( + context, + params=LLMUserAggregatorParams( + user_turn_strategies=UserTurnStrategies( + stop=[ + SpeechTimeoutUserTurnStopStrategy( + user_speech_timeout=TRANSCRIPTION_TIMEOUT + ) + ], + ), + wait_for_transcript_to_end_user_turn=False, + ), + ) + + events: list[tuple[str, str]] = [] + + @user_aggregator.event_handler("on_user_turn_stopped") + async def on_stopped(aggregator, strategy, message): + events.append(("stopped", message.content)) + + @user_aggregator.event_handler("on_user_turn_message_finalized") + async def on_finalized(aggregator, strategy, message): + events.append(("finalized", message.content)) + + pipeline = Pipeline([user_aggregator]) + + frames_to_send = [ + VADUserStartedSpeakingFrame(), + SleepFrame(), + VADUserStoppedSpeakingFrame(), + # Strategy fires turn-stopped. + SleepFrame(sleep=TRANSCRIPTION_TIMEOUT + 0.05), + # Backstop fires without any transcript arriving. + SleepFrame(sleep=TRANSCRIPTION_TIMEOUT + 0.05), + ] + await run_test(pipeline, frames_to_send=frames_to_send) + + self.assertEqual(events, [("stopped", ""), ("finalized", "")]) + + # No user message added to context (empty aggregation). + user_messages = [m for m in context.get_messages() if m.get("role") == "user"] + self.assertEqual(user_messages, []) + + async def test_no_wait_for_transcript_next_turn_force_flushes_previous(self): + """If a new user turn starts while the previous turn's transcript + is still pending (precondition violation), the previous turn's + finalization fires before the new turn's start. Whatever + transcript was captured by then is what lands in context. + """ + from unittest.mock import patch + + from pipecat.processors.aggregators import llm_response_universal + + with patch.object( + llm_response_universal, + "DEFAULT_TTFS_P99", + TRANSCRIPTION_TIMEOUT * 10, # backstop should NOT fire during the test + ): + context = LLMContext() + user_aggregator = LLMUserAggregator( + context, + params=LLMUserAggregatorParams( + user_turn_strategies=UserTurnStrategies( + stop=[ + SpeechTimeoutUserTurnStopStrategy( + user_speech_timeout=TRANSCRIPTION_TIMEOUT + ) + ], + ), + wait_for_transcript_to_end_user_turn=False, + ), + ) + + events: list[tuple[str, str]] = [] + + @user_aggregator.event_handler("on_user_turn_stopped") + async def on_stopped(aggregator, strategy, message): + events.append(("stopped", message.content)) + + @user_aggregator.event_handler("on_user_turn_message_finalized") + async def on_finalized(aggregator, strategy, message): + events.append(("finalized", message.content)) + + @user_aggregator.event_handler("on_user_turn_started") + async def on_started(aggregator, strategy): + events.append(("started", "")) + + pipeline = Pipeline([user_aggregator]) + + frames_to_send = [ + # Turn 1 + VADUserStartedSpeakingFrame(), + SleepFrame(), + VADUserStoppedSpeakingFrame(), + SleepFrame(sleep=TRANSCRIPTION_TIMEOUT + 0.05), + # Late transcript for turn 1 arrives. + TranscriptionFrame(text="Hello!", user_id="", timestamp="now"), + SleepFrame(), + # Turn 2 starts before turn 1's backstop fires — precondition + # violation. The aggregator should force-flush turn 1 first. + VADUserStartedSpeakingFrame(), + SleepFrame(), + ] + await run_test(pipeline, frames_to_send=frames_to_send) + + # The sequence must show turn 1's audible stop and finalization + # firing before turn 2's start event. + self.assertEqual( + events, + [ + ("started", ""), # turn 1 starts + ("stopped", ""), # turn 1 audible stop + ("finalized", "Hello!"), # forced flush before turn 2 starts + ("started", ""), # turn 2 starts + ], + ) + + user_messages = [m for m in context.get_messages() if m.get("role") == "user"] + self.assertEqual([m["content"] for m in user_messages], ["Hello!"]) + + async def test_no_wait_for_transcript_context_order_with_assistant_response(self): + """End-to-end ordering test: with both aggregators, verify the user + message lands in context *before* the assistant message, even though + the user transcript arrives after the audible turn end. + + Models the documented precondition: the transcript arrives before + the assistant content begins streaming. The user aggregator must + flush before the assistant aggregator does. + """ + from unittest.mock import patch + + from pipecat.processors.aggregators import llm_response_universal + + # Short backstop so the user flush fires while the assistant + # response is still streaming. + with patch.object(llm_response_universal, "DEFAULT_TTFS_P99", 0.05): + context = LLMContext() + user_aggregator, assistant_aggregator = LLMContextAggregatorPair( + context, + user_params=LLMUserAggregatorParams( + user_turn_strategies=UserTurnStrategies( + stop=[ + SpeechTimeoutUserTurnStopStrategy( + user_speech_timeout=TRANSCRIPTION_TIMEOUT + ) + ], + ), + wait_for_transcript_to_end_user_turn=False, + ), + ) + + pipeline = Pipeline([user_aggregator, assistant_aggregator]) + + frames_to_send = [ + VADUserStartedSpeakingFrame(), + SleepFrame(), + VADUserStoppedSpeakingFrame(), + # Strategy fires turn-stopped (audible). + SleepFrame(sleep=TRANSCRIPTION_TIMEOUT + 0.05), + # User transcript arrives after audible stop (the realtime + # service has finally emitted it). + TranscriptionFrame(text="What's the weather?", user_id="", timestamp="now"), + # Bot starts responding. By the precondition, transcripts + # arrive before assistant content begins streaming. + LLMFullResponseStartFrame(), + LLMTextFrame("It's sunny."), + # Allow time for the user backstop to fire (flushing the + # user message to context) before the assistant turn ends. + SleepFrame(sleep=0.1), + LLMFullResponseEndFrame(), + SleepFrame(), + ] + await run_test(pipeline, frames_to_send=frames_to_send) + + # Context must contain the user message before the assistant message. + roles_and_content = [(m.get("role"), m.get("content")) for m in context.get_messages()] + self.assertEqual( + roles_and_content, + [ + ("user", "What's the weather?"), + ("assistant", "It's sunny."), + ], + ) + + async def test_no_wait_for_transcript_strategies_are_mutated(self): + """``wait_for_transcript_to_end_user_turn=False`` mutates the + provided strategies: drops ``TranscriptionUserTurnStartStrategy`` + from start, flips ``wait_for_transcript=False`` on stop. + """ + from pipecat.turns.user_start import ( + TranscriptionUserTurnStartStrategy, + VADUserTurnStartStrategy, + ) + + context = LLMContext() + stop = SpeechTimeoutUserTurnStopStrategy( + user_speech_timeout=TRANSCRIPTION_TIMEOUT, + wait_for_transcript=True, # explicitly True; bundle should flip to False + ) + user_aggregator = LLMUserAggregator( + context, + params=LLMUserAggregatorParams( + user_turn_strategies=UserTurnStrategies( + start=[ + VADUserTurnStartStrategy(), + TranscriptionUserTurnStartStrategy(), + ], + stop=[stop], + ), + wait_for_transcript_to_end_user_turn=False, + ), + ) + + # Start strategies: TranscriptionUserTurnStartStrategy dropped. + start_types = [type(s) for s in (user_aggregator._params.user_turn_strategies.start or [])] + self.assertEqual(start_types, [VADUserTurnStartStrategy]) + + # Stop strategy: wait_for_transcript flipped to False. + self.assertFalse(stop._wait_for_transcript) + class TestLLMAssistantAggregator(unittest.IsolatedAsyncioTestCase): async def test_empty(self):