diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py
index 13c3f24fb..9a7ee9c3e 100644
--- a/src/pipecat/processors/aggregators/llm_response_universal.py
+++ b/src/pipecat/processors/aggregators/llm_response_universal.py
@@ -64,7 +64,7 @@ from pipecat.processors.aggregators.llm_context import (
     LLMSpecificMessage,
     NotGiven,
 )
-from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.processors.frame_processor import FrameCallback, FrameDirection, FrameProcessor
 from pipecat.turns.user_idle_controller import UserIdleController
 from pipecat.turns.user_mute import BaseUserMuteStrategy
 from pipecat.turns.user_start import BaseUserTurnStartStrategy, UserTurnStartedParams
@@ -452,7 +452,7 @@ class LLMUserAggregator(LLMContextAggregator):
             await self.push_frame(frame, direction)
         elif isinstance(frame, LLMSetToolChoiceFrame):
             self.set_tool_choice(frame.tool_choice)
-        elif isinstance(frame, SpeechControlParamsFrame):
+        elif isinstance(frame, SpeechControlParamsFrame) and not self._vad_controller:
             await self._handle_speech_control_params(frame)
         else:
             await self.push_frame(frame, direction)
@@ -587,25 +587,36 @@ class LLMUserAggregator(LLMContextAggregator):
             )
         )
 
+    async def _queued_broadcast_frame(self, frame_cls: Type[Frame], **kwargs):
+        """Broadcasts a frame upstream and queues it downstream for internal processing.
+
+        Queues the frame downstream so it flows through ``process_frame`` and
+        is handled internally (e.g. by the ``UserTurnController``). The
+        upstream frame is pushed directly.
+
+        Args:
+            frame_cls: The class of the frame to be broadcasted.
+            **kwargs: Keyword arguments to be passed to the frame's constructor.
+        """
+        await self.queue_frame(frame_cls(**kwargs))
+        await self.push_frame(frame_cls(**kwargs), FrameDirection.UPSTREAM)
+
     async def _on_push_frame(
         self, controller, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM
     ):
-        await self.push_frame(frame, direction)
+        await self.queue_frame(frame, direction)
 
     async def _on_broadcast_frame(self, controller, frame_cls: Type[Frame], **kwargs):
-        await self.broadcast_frame(frame_cls, **kwargs)
+        await self._queued_broadcast_frame(frame_cls, **kwargs)
 
     async def _on_vad_speech_started(self, controller):
-        await self.queue_frame(VADUserStartedSpeakingFrame())
-        await self.broadcast_frame(VADUserStartedSpeakingFrame)
+        await self._queued_broadcast_frame(VADUserStartedSpeakingFrame)
 
     async def _on_vad_speech_stopped(self, controller):
-        await self.queue_frame(VADUserStoppedSpeakingFrame())
-        await self.broadcast_frame(VADUserStoppedSpeakingFrame)
+        await self._queued_broadcast_frame(VADUserStoppedSpeakingFrame)
 
     async def _on_vad_speech_activity(self, controller):
-        await self.queue_frame(UserSpeakingFrame())
-        await self.broadcast_frame(UserSpeakingFrame)
+        await self._queued_broadcast_frame(UserSpeakingFrame)
 
     async def _on_user_turn_started(
         self,