diff --git a/changelog/3737.fixed.md b/changelog/3737.fixed.md new file mode 100644 index 000000000..6dee96f82 --- /dev/null +++ b/changelog/3737.fixed.md @@ -0,0 +1 @@ +- Fixed `LLMUserAggregator` broadcasting mute events before `StartFrame` reaches downstream processors. diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index 6b28b5bf2..8450842c8 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -549,6 +549,15 @@ class LLMUserAggregator(LLMContextAggregator): await s.cleanup() async def _maybe_mute_frame(self, frame: Frame): + # Control frames must flow unconditionally — never feed them to mute + # strategies. Without this guard, strategies like + # MuteUntilFirstBotCompleteUserMuteStrategy fire on_user_mute_started + # and broadcast UserMuteStartedFrame before StartFrame is pushed + # downstream, causing downstream processors to receive frames before + # StartFrame and log errors. + if isinstance(frame, (StartFrame, EndFrame, CancelFrame)): + return False + should_mute_frame = self._user_is_muted and isinstance( frame, ( diff --git a/tests/test_context_aggregators_universal.py b/tests/test_context_aggregators_universal.py index 7e09a5449..1bba463b0 100644 --- a/tests/test_context_aggregators_universal.py +++ b/tests/test_context_aggregators_universal.py @@ -24,7 +24,9 @@ from pipecat.frames.frames import ( LLMThoughtEndFrame, LLMThoughtStartFrame, LLMThoughtTextFrame, + StartFrame, TranscriptionFrame, + UserMuteStartedFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, VADUserStartedSpeakingFrame, @@ -40,7 +42,11 @@ from pipecat.processors.aggregators.llm_response_universal import ( LLMUserAggregatorParams, ) from pipecat.tests.utils import SleepFrame, run_test -from pipecat.turns.user_mute import FirstSpeechUserMuteStrategy, FunctionCallUserMuteStrategy +from pipecat.turns.user_mute import ( + FirstSpeechUserMuteStrategy, + FunctionCallUserMuteStrategy, + MuteUntilFirstBotCompleteUserMuteStrategy, +) from pipecat.turns.user_stop import SpeechTimeoutUserTurnStopStrategy from pipecat.turns.user_turn_strategies import UserTurnStrategies @@ -386,6 +392,42 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase): self.assertIsNone(strategy) # strategy is None for end/cancel self.assertEqual(message.content, "Hello!") + async def test_start_frame_before_mute_event(self): + """StartFrame must reach downstream before mute events are broadcast. + + With MuteUntilFirstBotCompleteUserMuteStrategy, the mute logic should + not run on control frames (StartFrame, EndFrame, CancelFrame). This + ensures StartFrame reaches downstream processors before + UserMuteStartedFrame is broadcast. + + The default TurnAnalyzerUserTurnStopStrategy broadcasts a + SpeechControlParamsFrame when it processes StartFrame, which gets + re-queued to the aggregator. That non-control frame legitimately + triggers the mute state change, so UserMuteStartedFrame follows + StartFrame — but crucially, after it. + """ + context = LLMContext() + + user_aggregator = LLMUserAggregator( + context, + params=LLMUserAggregatorParams( + user_mute_strategies=[MuteUntilFirstBotCompleteUserMuteStrategy()], + ), + ) + + pipeline = Pipeline([user_aggregator]) + + # run_test internally sends StartFrame via PipelineRunner. With + # ignore_start=False we can verify ordering: StartFrame must arrive + # before UserMuteStartedFrame. Before the fix, UserMuteStartedFrame + # was broadcast before StartFrame reached downstream processors. + (down_frames, _) = await run_test( + pipeline, + frames_to_send=[], + expected_down_frames=[StartFrame, UserMuteStartedFrame], + ignore_start=False, + ) + class TestLLMAssistantAggregator(unittest.IsolatedAsyncioTestCase): async def test_empty(self):