diff --git a/changelog/3956.fixed.md b/changelog/3956.fixed.md new file mode 100644 index 000000000..c8af8db91 --- /dev/null +++ b/changelog/3956.fixed.md @@ -0,0 +1 @@ +- Fixed turn completion state not resetting at end of LLM responses. `LLMFullResponseEndFrame` is pushed (not received) by the LLM service, so the mixin now handles it in `push_frame` instead of `process_frame`. diff --git a/src/pipecat/turns/user_turn_completion_mixin.py b/src/pipecat/turns/user_turn_completion_mixin.py index 3aea01be9..40ae551b4 100644 --- a/src/pipecat/turns/user_turn_completion_mixin.py +++ b/src/pipecat/turns/user_turn_completion_mixin.py @@ -332,14 +332,25 @@ class UserTurnCompletionLLMServiceMixin: if isinstance(frame, InterruptionFrame): await self._cancel_incomplete_timeout() await self._turn_reset() - # Reset turn state at end of LLM response (but don't cancel timeout - - # incomplete timeouts should continue running) - elif isinstance(frame, LLMFullResponseEndFrame): - await self._turn_reset() # Pass frame to parent await super().process_frame(frame, direction) + async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): + """Push a frame downstream, resetting turn state at end of each LLM response. + + ``LLMFullResponseEndFrame`` is generated by the LLM service itself (pushed, + not received), so it must be handled here rather than in ``process_frame``. + + Args: + frame: The frame to push downstream. + direction: The direction of frame flow. Defaults to downstream. + """ + if isinstance(frame, LLMFullResponseEndFrame): + await self._turn_reset() + + await super().push_frame(frame, direction) + async def _push_turn_text(self, text: str): """Push LLM text with turn completion detection. diff --git a/tests/test_user_turn_completion_mixin.py b/tests/test_user_turn_completion_mixin.py index d258cff9a..273d8cc90 100644 --- a/tests/test_user_turn_completion_mixin.py +++ b/tests/test_user_turn_completion_mixin.py @@ -5,9 +5,10 @@ # import unittest +import unittest.mock from unittest.mock import AsyncMock -from pipecat.frames.frames import LLMTextFrame +from pipecat.frames.frames import LLMFullResponseEndFrame, LLMTextFrame from pipecat.processors.frame_processor import FrameProcessor from pipecat.turns.user_turn_completion_mixin import ( USER_TURN_COMPLETE_MARKER, @@ -112,6 +113,32 @@ class TestUserUserTurnCompletionLLMServiceMixin(unittest.IsolatedAsyncioTestCase # Now frames should be pushed self.assertEqual(len(pushed_frames), 2) + async def test_turn_state_reset_after_llm_full_response_end_frame(self): + """Test that _turn_complete_found is reset when LLMFullResponseEndFrame is pushed.""" + processor = MockProcessor() + + # Mock push_frame on the instance so _push_turn_text can call it without + # a live pipeline, but keep _turn_reset as the real implementation. + processor.push_frame = AsyncMock() + + # Simulate first LLM response: complete marker sets _turn_complete_found = True + await processor._push_turn_text(f"{USER_TURN_COMPLETE_MARKER} Hello!") + self.assertTrue(processor._turn_complete_found) + + # Restore the real push_frame so the mixin override runs, then call it + # with LLMFullResponseEndFrame as the LLM service would. + del processor.push_frame # removes instance mock, restores class method + + # Patch only the FrameProcessor-level send so no live pipeline is needed. + with unittest.mock.patch.object(FrameProcessor, "push_frame", AsyncMock()): + end_frame = LLMFullResponseEndFrame() + await processor.push_frame(end_frame) + + # _turn_complete_found must now be False — ready for the next response + self.assertFalse(processor._turn_complete_found) + self.assertEqual(processor._turn_text_buffer, "") + self.assertFalse(processor._turn_suppressed) + if __name__ == "__main__": unittest.main()