Add RealtimeServiceModeConfig to LLMContextAggregatorPair

Decouple context management from turn frames and transcripts when a
realtime LLM service drives the conversation. Three problems with today's
behavior:

  - Some realtime services (Gemini Live, AWS Nova Sonic, Ultravox) emit
    no UserStarted/StoppedSpeakingFrame at all, so the aggregator — which
    writes user messages on those frames — doesn't write to context
    correctly without them.
  - The workaround (local VAD on the aggregator) generates turn
    boundaries that don't match the provider's server-side ground truth,
    and the per-service "do I need it?" rule is hard to keep straight.
  - When local turn detection is the intended driver, turn-end strategies
    still wait for transcripts on the latency critical path.

Add a realtime_service_mode: RealtimeServiceModeConfig | None = None
kwarg on LLMContextAggregatorPair. When set, the pair switches both
halves to trailing context writes: user messages are flushed on the first
assistant content frame, assistant messages on the next user transcript,
both halves on EndFrame. Turn-end strategies stop waiting for transcripts
by default. Two fine-grained boolean fields (context_writes_await_turns,
turns_await_transcripts) let callers dial back to cascade-style behavior
selectively; their invalid combination is rejected in __post_init__.

The bifurcation is dispatch-only: seven branch points across the two
halves, each at method entry, each delegating to a mode-pure private
method. Cross-half coordination uses an asyncio.Lock and a back-reference
shared by both halves; the assistant signals user.flush() on
LLMFullResponseStartFrame, and the user signals assistant.flush() on the
first new transcript after the assistant turn. The mechanism reuses the
existing push_aggregation() — no parallel write path.

Two new events fire when messages are flushed to context:
on_user_message_added and on_assistant_message_added. In cascade mode
they coincide with the existing turn-stopped events; in realtime mode
(where the turn-stopped event fires before the message is finalized)
they're the canonical way to subscribe to "context just updated, here's
the text."

UserTurnStoppedMessage.content is now typed str | None to reflect that
realtime mode fires the event with None.

When a RealtimeServiceMetadataFrame arrives and realtime_service_mode is
None, the aggregator logs a one-time INFO recommendation pointing users
at the option.

src/pipecat/processors/aggregators/llm_response_universal.py

@@ -55,6 +55,7 @@ from pipecat.frames.frames import (
     LLMThoughtEndFrame,
     LLMThoughtStartFrame,
     LLMThoughtTextFrame,
+    RealtimeServiceMetadataFrame,
     StartFrame,
     TextFrame,
     TranscriptionFrame,
@@ -83,7 +84,11 @@ from pipecat.processors.aggregators.llm_context_summarizer import (
 from pipecat.processors.frame_processor import 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
+from pipecat.turns.user_start import (
+    BaseUserTurnStartStrategy,
+    TranscriptionUserTurnStartStrategy,
+    UserTurnStartedParams,
+)
 from pipecat.turns.user_stop import BaseUserTurnStopStrategy, UserTurnStoppedParams
 from pipecat.turns.user_turn_completion_mixin import UserTurnCompletionConfig
 from pipecat.turns.user_turn_controller import UserTurnController
@@ -258,6 +263,43 @@ class LLMAssistantAggregatorParams:
             self.context_summarization_config = None
 
 
+@dataclass
+class RealtimeServiceModeConfig:
+    """Configure an ``LLMContextAggregatorPair`` for use with a realtime LLM service.
+
+    Both fields default to False (the recommended realtime behavior, dropping
+    transcript-related waits at both points in the flow). Override individual
+    fields to dial back to cascade-style behavior selectively.
+
+    Parameters:
+        context_writes_await_turns: When False (default), context writes are
+            triggered by the content stream itself (transcripts and assistant
+            text frames), making writes independent of turn-frame availability
+            and timing. When True, user messages are written to context on
+            user-turn-end frames (cascade behavior).
+        turns_await_transcripts: When False (default), turn-end fires as soon
+            as VAD signals end of speech, avoiding latency on the critical
+            path when local turn detection drives a realtime conversation.
+            When True, turn-end strategies wait for transcripts to arrive
+            before signalling end-of-turn.
+    """
+
+    context_writes_await_turns: bool = False
+    turns_await_transcripts: bool = False
+
+    def __post_init__(self):
+        """Validate the field combination."""
+        if not self.turns_await_transcripts and self.context_writes_await_turns:
+            raise ValueError(
+                "Invalid combination: turns fire early (without transcripts) "
+                "but context writes wait on those turn frames — context would "
+                "be written with incomplete user messages. Either set "
+                "turns_await_transcripts=True (preserve transcript-aware "
+                "turn-end timing) or context_writes_await_turns=False "
+                "(decouple writes from turn frames)."
+            )
+
+
 @dataclass
 class UserTurnStoppedMessage:
     """A user turn stopped message containing a user transcript update.
@@ -266,13 +308,18 @@ class UserTurnStoppedMessage:
     the aggregated transcript that is then used in the context.
 
     Parameters:
-        content: The message content/text.
+        content: The message content/text. ``None`` in realtime mode
+            (``RealtimeServiceModeConfig(context_writes_await_turns=False)``)
+            when fired from a user-turn-stop frame, since the user message
+            hasn't been finalized at that point. Subscribers that need the
+            finalized text should listen to ``on_user_message_added``
+            instead.
         timestamp: When the user turn started.
         user_id: Optional identifier for the user.
 
     """
 
-    content: str
+    content: str | None
     timestamp: str
     user_id: str | None = None
 
@@ -567,6 +614,9 @@ class LLMUserAggregator(LLMContextAggregator):
         context: LLMContext,
         *,
         params: LLMUserAggregatorParams | None = None,
+        _realtime_service_mode: RealtimeServiceModeConfig | None = None,
+        _paired_half: "LLMAssistantAggregator | None" = None,
+        _pair_lock: asyncio.Lock | None = None,
         **kwargs,
     ):
         """Initialize the user context aggregator.
@@ -574,6 +624,14 @@ class LLMUserAggregator(LLMContextAggregator):
         Args:
             context: The LLM context for conversation storage.
             params: Configuration parameters for aggregation behavior.
+            _realtime_service_mode: Pair-internal. Realtime-mode
+                configuration propagated from
+                ``LLMContextAggregatorPair``. Not intended for direct use —
+                construct the aggregators via the pair.
+            _paired_half: Pair-internal. Back-reference to the paired
+                assistant aggregator for cross-half coordination.
+            _pair_lock: Pair-internal. Shared asyncio lock serializing
+                cross-half flushes.
             **kwargs: Additional arguments.
         """
         params = params or LLMUserAggregatorParams()
@@ -590,9 +648,23 @@ class LLMUserAggregator(LLMContextAggregator):
         self._register_event_handler("on_user_turn_stop_timeout")
         self._register_event_handler("on_user_turn_idle")
         self._register_event_handler("on_user_turn_inference_triggered")
+        self._register_event_handler("on_user_message_added")
         self._register_event_handler("on_user_mute_started")
         self._register_event_handler("on_user_mute_stopped")
 
+        # Realtime-mode wiring. Defaults (no config) preserve cascade
+        # behavior: context writes happen on turn frames, turns wait
+        # for transcripts.
+        self._realtime_service_mode = _realtime_service_mode
+        self._paired_half = _paired_half
+        self._pair_lock = _pair_lock
+        if _realtime_service_mode is not None:
+            self._context_writes_await_turns = _realtime_service_mode.context_writes_await_turns
+            self._turns_await_transcripts = _realtime_service_mode.turns_await_transcripts
+        else:
+            self._context_writes_await_turns = True
+            self._turns_await_transcripts = True
+
         user_turn_strategies = self._params.user_turn_strategies or UserTurnStrategies()
 
         # Deprecated path: translate filter_incomplete_user_turns into
@@ -606,8 +678,19 @@ class LLMUserAggregator(LLMContextAggregator):
             )
             self._params.user_turn_strategies = user_turn_strategies
 
+        # Realtime mutation: when turns shouldn't wait for transcripts,
+        # drop the transcription-based start strategy and flip the
+        # wait_for_transcript flag on stop strategies that expose it. The
+        # set of strategies that support it intentionally stays narrow —
+        # the flag was reintroduced specifically for this realtime path.
+        if not self._turns_await_transcripts:
+            self._apply_realtime_strategy_mutations(user_turn_strategies)
+
         self._user_is_muted = False
         self._user_turn_start_timestamp = ""
+        # Tracks whether the §3.6 recommendation log has already fired
+        # for this session — see _handle_realtime_service_metadata.
+        self._realtime_recommendation_logged = False
         # Full transcript across the user turn. Each
         # `_on_user_turn_inference_triggered` push captures only the
         # new segment since the previous push (push_aggregation resets
@@ -717,6 +800,9 @@ class LLMUserAggregator(LLMContextAggregator):
             await self.push_frame(frame, direction)
         elif isinstance(frame, LLMSetToolChoiceFrame):
             self.set_tool_choice(frame.tool_choice)
+        elif isinstance(frame, RealtimeServiceMetadataFrame):
+            await self._handle_realtime_service_metadata(frame)
+            await self.push_frame(frame, direction)
         else:
             await self.push_frame(frame, direction)
 
@@ -734,9 +820,16 @@ class LLMUserAggregator(LLMContextAggregator):
         self._context.add_message({"role": self.role, "content": aggregation})
         await self.push_context_frame()
 
+        message = UserTurnStoppedMessage(
+            content=aggregation, timestamp=self._user_turn_start_timestamp
+        )
+        await self._call_event_handler("on_user_message_added", message)
+
         return aggregation
 
     async def _start(self, frame: StartFrame):
+        self._validate_realtime_pairing()
+
         if self._vad_controller:
             await self._vad_controller.setup(self.task_manager)
 
@@ -748,13 +841,138 @@ class LLMUserAggregator(LLMContextAggregator):
             await s.setup(self.task_manager)
 
     async def _stop(self, frame: EndFrame):
-        await self._maybe_emit_user_turn_stopped(on_session_end=True)
+        if not self._context_writes_await_turns:
+            # Realtime: flush trailing user content directly. The
+            # on_user_turn_stopped event already fired (if turn frames
+            # were emitted), so don't re-fire it from session end.
+            await self.push_aggregation()
+        else:
+            await self._maybe_emit_user_turn_stopped(on_session_end=True)
         await self._cleanup()
 
     async def _cancel(self, frame: CancelFrame):
-        await self._maybe_emit_user_turn_stopped(on_session_end=True)
+        if not self._context_writes_await_turns:
+            await self.push_aggregation()
+        else:
+            await self._maybe_emit_user_turn_stopped(on_session_end=True)
         await self._cleanup()
 
+    def _validate_realtime_pairing(self):
+        """Validate the realtime-mode wiring set by ``LLMContextAggregatorPair``.
+
+        Realtime mode requires both halves to be paired through the
+        ``LLMContextAggregatorPair`` so cross-half flushes can find each
+        other. Direct construction of a half with the private realtime
+        kwargs is not supported.
+        """
+        if not self._context_writes_await_turns:
+            if self._paired_half is None:
+                raise RuntimeError(
+                    f"{self}: realtime_service_mode is configured but this user "
+                    "aggregator has no paired assistant aggregator. Construct "
+                    "the pair via LLMContextAggregatorPair("
+                    "context, realtime_service_mode=RealtimeServiceModeConfig())."
+                )
+        if self._paired_half is not None:
+            if (
+                self._context_writes_await_turns != self._paired_half._context_writes_await_turns
+                or self._turns_await_transcripts != self._paired_half._turns_await_transcripts
+            ):
+                raise RuntimeError(
+                    f"{self}: realtime-mode config mismatch between user and "
+                    "assistant halves. Use LLMContextAggregatorPair to construct "
+                    "the pair so both halves share the same configuration."
+                )
+
+    def _apply_realtime_strategy_mutations(self, user_turn_strategies: UserTurnStrategies) -> None:
+        """Mutate turn strategies for the realtime ``turns_await_transcripts=False`` path.
+
+        Drops ``TranscriptionUserTurnStartStrategy`` from the start strategies
+        (transcripts shouldn't start a turn when the realtime service drives
+        the conversation) and flips ``wait_for_transcript=False`` on stop
+        strategies that expose the flag, so end-of-turn fires as soon as VAD /
+        the turn analyzer reports end-of-speech.
+        """
+        custom_strategies = self._params.user_turn_strategies is not None
+
+        start_strategies = user_turn_strategies.start or []
+        dropped: list[str] = []
+        kept_start: list[BaseUserTurnStartStrategy] = []
+        for s in start_strategies:
+            if isinstance(s, TranscriptionUserTurnStartStrategy):
+                dropped.append(s.__class__.__name__)
+            else:
+                kept_start.append(s)
+        user_turn_strategies.start = kept_start
+
+        flipped: list[str] = []
+        for s in user_turn_strategies.stop or []:
+            if hasattr(s, "wait_for_transcript"):
+                try:
+                    s.wait_for_transcript = False
+                    flipped.append(s.__class__.__name__)
+                except AttributeError:
+                    # Strategy exposes the property but no setter — skip.
+                    pass
+
+        if not dropped and not flipped:
+            return
+
+        msg = (
+            f"{self}: realtime_service_mode(turns_await_transcripts=False) — "
+            f"dropped {dropped or 'no'} start strategy(ies); set "
+            f"wait_for_transcript=False on {flipped or 'no'} stop strategy(ies)."
+        )
+        if custom_strategies:
+            logger.warning(msg)
+        else:
+            logger.debug(msg)
+
+    async def _handle_realtime_service_metadata(self, frame: RealtimeServiceMetadataFrame):
+        """Handle a ``RealtimeServiceMetadataFrame`` broadcast by a realtime LLM service.
+
+        When ``realtime_service_mode`` isn't configured, log a one-time INFO
+        recommendation pointing the user at the option and warning about the
+        timing change on ``on_user_turn_stopped``. When it is configured, log
+        a confirming debug message. Fires at most once per session.
+        """
+        if self._realtime_recommendation_logged:
+            return
+        self._realtime_recommendation_logged = True
+
+        if self._realtime_service_mode is None:
+            logger.info(
+                f"{self}: detected realtime service `{frame.service_name}` in the "
+                "pipeline. For correct context-write semantics with realtime "
+                "services, consider passing "
+                "realtime_service_mode=RealtimeServiceModeConfig() to "
+                "LLMContextAggregatorPair. Note: this changes when user messages "
+                "are written to context — they're written when the assistant "
+                "response starts rather than when the user-turn-end frame fires. "
+                "Subscribe to `on_user_message_added` instead of "
+                "`on_user_turn_stopped` if you need post-write semantics."
+            )
+        else:
+            logger.debug(
+                f"{self}: detected realtime service `{frame.service_name}`; "
+                "realtime_service_mode is configured."
+            )
+
+    async def _realtime_handoff_flush(self) -> None:
+        """Flush pending user aggregation to context.
+
+        Called by the paired assistant half from
+        ``_realtime_handle_llm_start`` (i.e. on ``LLMFullResponseStartFrame``)
+        to commit the in-flight user message before the assistant starts
+        its own turn. No-op when there's no pending content.
+        """
+        if not self._aggregation:
+            return
+        # push_aggregation writes the message to context, pushes
+        # LLMContextFrame, and emits on_user_message_added.
+        await self.push_aggregation()
+        self._user_turn_start_timestamp = ""
+
     async def _cleanup(self):
         if self._vad_controller:
             await self._vad_controller.cleanup()
@@ -826,6 +1044,10 @@ class LLMUserAggregator(LLMContextAggregator):
             await self.push_context_frame()
 
     async def _handle_transcription(self, frame: TranscriptionFrame):
+        if not self._context_writes_await_turns:
+            await self._realtime_handle_transcription(frame)
+            return
+
         text = frame.text
 
         # Make sure we really have some text.
@@ -839,6 +1061,30 @@ class LLMUserAggregator(LLMContextAggregator):
             )
         )
 
+    async def _realtime_handle_transcription(self, frame: TranscriptionFrame):
+        """Realtime variant: signal the paired assistant half to flush, then append.
+
+        The first new user transcript after an assistant turn ends is what
+        commits the assistant's pending message to context. The flush is
+        idempotent (no-op when nothing pending), so it's safe to call on
+        every chunk.
+        """
+        if not frame.text.strip():
+            return
+
+        if self._paired_half is not None and self._pair_lock is not None:
+            async with self._pair_lock:
+                await self._paired_half._realtime_handoff_flush()
+
+        if not self._user_turn_start_timestamp:
+            self._user_turn_start_timestamp = time_now_iso8601()
+
+        self._aggregation.append(
+            TextPartForConcatenation(
+                frame.text, includes_inter_part_spaces=frame.includes_inter_frame_spaces
+            )
+        )
+
     async def _queued_broadcast_frame(self, frame_cls: type[Frame], **kwargs):
         """Broadcasts a frame upstream and queues it for internal processing.
 
@@ -903,6 +1149,17 @@ class LLMUserAggregator(LLMContextAggregator):
         controller: UserTurnController,
         strategy: BaseUserTurnStopStrategy,
     ):
+        if not self._context_writes_await_turns:
+            # Realtime: turn frames are supplemental — they don't drive
+            # context writes. Fire the event without pushing aggregation;
+            # the trailing-write path commits the user message instead.
+            logger.debug(
+                f"{self}: User turn inference triggered (strategy: {strategy}) "
+                "[realtime: event-only, no context push]"
+            )
+            await self._call_event_handler("on_user_turn_inference_triggered", strategy)
+            return
+
         logger.debug(f"{self}: User turn inference triggered (strategy: {strategy})")
 
         # Push aggregation now: this writes the user message segment to
@@ -935,6 +1192,17 @@ class LLMUserAggregator(LLMContextAggregator):
 
         await self._user_idle_controller.process_frame(UserStoppedSpeakingFrame())
 
+        if not self._context_writes_await_turns:
+            # Realtime: turn frames are supplemental. The user message
+            # isn't finalized at turn-stop time — content is None.
+            # Subscribers wanting the finalized text use
+            # on_user_message_added instead.
+            message = UserTurnStoppedMessage(
+                content=None, timestamp=self._user_turn_start_timestamp
+            )
+            await self._call_event_handler("on_user_turn_stopped", strategy, message)
+            return
+
         await self._maybe_emit_user_turn_stopped(strategy)
 
     async def _on_reset_aggregation(
@@ -1030,6 +1298,9 @@ class LLMAssistantAggregator(LLMContextAggregator):
         context: LLMContext,
         *,
         params: LLMAssistantAggregatorParams | None = None,
+        _realtime_service_mode: RealtimeServiceModeConfig | None = None,
+        _paired_half: "LLMUserAggregator | None" = None,
+        _pair_lock: asyncio.Lock | None = None,
         **kwargs,
     ):
         """Initialize the assistant context aggregator.
@@ -1037,6 +1308,14 @@ class LLMAssistantAggregator(LLMContextAggregator):
         Args:
             context: The OpenAI LLM context for conversation storage.
             params: Configuration parameters for aggregation behavior.
+            _realtime_service_mode: Pair-internal. Realtime-mode
+                configuration propagated from
+                ``LLMContextAggregatorPair``. Not intended for direct use —
+                construct the aggregators via the pair.
+            _paired_half: Pair-internal. Back-reference to the paired
+                user aggregator for cross-half coordination.
+            _pair_lock: Pair-internal. Shared asyncio lock serializing
+                cross-half flushes.
             **kwargs: Additional arguments.
         """
         params = params or LLMAssistantAggregatorParams()
@@ -1048,6 +1327,24 @@ class LLMAssistantAggregator(LLMContextAggregator):
         )
         self._params = params
 
+        # Realtime-mode wiring. Defaults (no config) preserve cascade
+        # behavior: write to context on LLMFullResponseEndFrame.
+        self._realtime_service_mode = _realtime_service_mode
+        self._paired_half = _paired_half
+        self._pair_lock = _pair_lock
+        if _realtime_service_mode is not None:
+            self._context_writes_await_turns = _realtime_service_mode.context_writes_await_turns
+            self._turns_await_transcripts = _realtime_service_mode.turns_await_transcripts
+        else:
+            self._context_writes_await_turns = True
+            self._turns_await_transcripts = True
+
+        # Realtime mode only. Holds the assistant turn's content between
+        # LLMFullResponseEndFrame (the moment we mark it ready to flush)
+        # and the next user transcript (the moment we actually write it
+        # to context).
+        self._pending_assistant_message_to_flush: dict | None = None
+
         self._function_calls_in_progress: dict[str, FunctionCallInProgressFrame | None] = {}
         self._function_calls_image_results: dict[str, UserImageRawFrame] = {}
         self._context_updated_tasks: set[asyncio.Task] = set()
@@ -1084,6 +1381,7 @@ class LLMAssistantAggregator(LLMContextAggregator):
 
         self._register_event_handler("on_assistant_turn_started")
         self._register_event_handler("on_assistant_turn_stopped")
+        self._register_event_handler("on_assistant_message_added")
         self._register_event_handler("on_assistant_thought")
         self._register_event_handler("on_summary_applied")
 
@@ -1184,6 +1482,10 @@ class LLMAssistantAggregator(LLMContextAggregator):
             if self._push_context_on_bot_stopped_speaking and not self._user_speaking:
                 logger.debug(f"{self}: Bot stopped speaking — pushing deferred context frame!")
                 await self.push_context_frame(FrameDirection.UPSTREAM)
+        elif isinstance(frame, RealtimeServiceMetadataFrame):
+            # The user half logs the §3.6 recommendation; the assistant
+            # half just passes the frame through.
+            await self.push_frame(frame, direction)
         else:
             await self.push_frame(frame, direction)
 
@@ -1192,9 +1494,37 @@ class LLMAssistantAggregator(LLMContextAggregator):
             await self._summarizer.process_frame(frame)
 
     async def _start(self, frame: StartFrame):
+        self._validate_realtime_pairing()
         if self._summarizer:
             await self._summarizer.setup(self.task_manager)
 
+    def _validate_realtime_pairing(self):
+        """Validate the realtime-mode wiring set by ``LLMContextAggregatorPair``.
+
+        Realtime mode requires both halves to be paired through the
+        ``LLMContextAggregatorPair`` so cross-half flushes can find each
+        other. Direct construction of a half with the private realtime
+        kwargs is not supported.
+        """
+        if not self._context_writes_await_turns:
+            if self._paired_half is None:
+                raise RuntimeError(
+                    f"{self}: realtime_service_mode is configured but this assistant "
+                    "aggregator has no paired user aggregator. Construct the pair "
+                    "via LLMContextAggregatorPair("
+                    "context, realtime_service_mode=RealtimeServiceModeConfig())."
+                )
+        if self._paired_half is not None:
+            if (
+                self._context_writes_await_turns != self._paired_half._context_writes_await_turns
+                or self._turns_await_transcripts != self._paired_half._turns_await_transcripts
+            ):
+                raise RuntimeError(
+                    f"{self}: realtime-mode config mismatch between user and "
+                    "assistant halves. Use LLMContextAggregatorPair to construct "
+                    "the pair so both halves share the same configuration."
+                )
+
     async def push_aggregation(self) -> str:
         """Push the current assistant aggregation with timestamp."""
         if not self._aggregation:
@@ -1247,6 +1577,12 @@ class LLMAssistantAggregator(LLMContextAggregator):
 
     async def _handle_end_or_cancel(self, frame: Frame):
         await self._trigger_assistant_turn_stopped(interrupted=isinstance(frame, CancelFrame))
+        if not self._context_writes_await_turns:
+            # Flush any pending assistant content parked by
+            # _realtime_trigger_assistant_turn_stopped (i.e. the bot
+            # finished its last reply but no follow-up user transcript
+            # arrived before the session ended).
+            await self._realtime_handoff_flush()
         if self._summarizer:
             await self._summarizer.cleanup()
 
@@ -1349,26 +1685,7 @@ class LLMAssistantAggregator(LLMContextAggregator):
                     run_llm = True
 
         if run_llm and not self._user_speaking:
-            if self.has_queued_frame(FunctionCallResultFrame):
+            await self._maybe_push_context_after_function_result()
-                # Another FunctionCallResultFrame is already queued. Defer the context push
-                # to bundle all results into a single LLM call instead of triggering one
-                # inference pass per result. The context will be pushed once the last
-                # function call in the queue is processed.
-                logger.debug(
-                    f"{self}: More FunctionCallResultFrames queued — deferring context frame push."
-                )
-            elif self._bot_speaking:
-                # Defer the context frame push until the bot finishes speaking. If multiple
-                # function call results arrive while the bot is speaking, they all accumulate
-                # in the context and a single push is performed once speaking stops, preventing
-                # the LLM from running multiple times and producing duplicated responses.
-                # This should be an edge case, since it would require a FunctionCallResultFrame
-                # being queued between an LLM response start and end frame.
-                logger.debug(f"{self}: Bot is speaking — deferring context frame push.")
-                self._push_context_on_bot_stopped_speaking = True
-            else:
-                logger.debug(f"{self}: Pushing context frame!")
-                await self.push_context_frame(FrameDirection.UPSTREAM)
 
         # Call the `on_context_updated` callback once the function call result
         # is added to the context. Also, run this in a separate task to make
@@ -1379,6 +1696,42 @@ class LLMAssistantAggregator(LLMContextAggregator):
             self._context_updated_tasks.add(task)
             task.add_done_callback(self._context_updated_task_finished)
 
+    async def _maybe_push_context_after_function_result(self) -> None:
+        """Decide whether to push a context frame after a function-call result.
+
+        Dispatched by mode. Cascade re-runs LLM inference by pushing an
+        ``LLMContextFrame`` upstream (with care to avoid duplicate pushes
+        while results are queued or the bot is still speaking). Realtime
+        services consume function results directly via
+        ``FunctionCallResultFrame``, so the context-driven re-inference
+        cycle is unnecessary.
+        """
+        if not self._context_writes_await_turns:
+            # Realtime: the realtime service has the result via
+            # FunctionCallResultFrame. No context push needed.
+            return
+
+        if self.has_queued_frame(FunctionCallResultFrame):
+            # Another FunctionCallResultFrame is already queued. Defer the context push
+            # to bundle all results into a single LLM call instead of triggering one
+            # inference pass per result. The context will be pushed once the last
+            # function call in the queue is processed.
+            logger.debug(
+                f"{self}: More FunctionCallResultFrames queued — deferring context frame push."
+            )
+        elif self._bot_speaking:
+            # Defer the context frame push until the bot finishes speaking. If multiple
+            # function call results arrive while the bot is speaking, they all accumulate
+            # in the context and a single push is performed once speaking stops, preventing
+            # the LLM from running multiple times and producing duplicated responses.
+            # This should be an edge case, since it would require a FunctionCallResultFrame
+            # being queued between an LLM response start and end frame.
+            logger.debug(f"{self}: Bot is speaking — deferring context frame push.")
+            self._push_context_on_bot_stopped_speaking = True
+        else:
+            logger.debug(f"{self}: Pushing context frame!")
+            await self.push_context_frame(FrameDirection.UPSTREAM)
+
     async def _handle_function_call_intermediate_result(
         self, frame: FunctionCallResultFrame, in_progress_frame: FunctionCallInProgressFrame
     ):
@@ -1469,6 +1822,20 @@ class LLMAssistantAggregator(LLMContextAggregator):
             )
 
     async def _handle_llm_start(self, _: LLMFullResponseStartFrame):
+        if not self._context_writes_await_turns:
+            await self._realtime_handle_llm_start()
+            return
+        await self._trigger_assistant_turn_started()
+
+    async def _realtime_handle_llm_start(self):
+        """Realtime: flush the paired user half before starting the assistant turn.
+
+        The first content frame of an assistant turn is the trigger to
+        commit any in-flight user transcript to context.
+        """
+        if self._paired_half is not None and self._pair_lock is not None:
+            async with self._pair_lock:
+                await self._paired_half._realtime_handoff_flush()
         await self._trigger_assistant_turn_started()
 
     async def _handle_llm_end(self, _: LLMFullResponseEndFrame):
@@ -1606,6 +1973,10 @@ class LLMAssistantAggregator(LLMContextAggregator):
         await self._call_event_handler("on_assistant_turn_started")
 
     async def _trigger_assistant_turn_stopped(self, *, interrupted: bool = False):
+        if not self._context_writes_await_turns:
+            await self._realtime_trigger_assistant_turn_stopped(interrupted=interrupted)
+            return
+
         if not self._assistant_turn_start_timestamp:
             return
 
@@ -1620,9 +1991,86 @@ class LLMAssistantAggregator(LLMContextAggregator):
             timestamp=self._assistant_turn_start_timestamp,
         )
         await self._call_event_handler("on_assistant_turn_stopped", message)
+        if aggregation:
+            await self._call_event_handler("on_assistant_message_added", message)
 
         self._assistant_turn_start_timestamp = ""
 
+    async def _realtime_trigger_assistant_turn_stopped(self, *, interrupted: bool):
+        """Realtime variant: defer the context write or flush on interruption.
+
+        Normal end-of-turn (``interrupted=False``, from
+        ``LLMFullResponseEndFrame``) parks the message text in a pending
+        slot — it isn't written to context until the next user transcript
+        arrives or the session ends. Interruption (``interrupted=True``)
+        commits immediately, matching today's
+        ``AssistantTurnStoppedMessage.interrupted`` semantics.
+        """
+        if not self._assistant_turn_start_timestamp:
+            return
+
+        timestamp = self._assistant_turn_start_timestamp
+        self._assistant_turn_start_timestamp = ""
+
+        if interrupted:
+            aggregation = await self.push_aggregation()
+            if aggregation:
+                aggregation = self._maybe_strip_turn_completion_markers(aggregation)
+            message = AssistantTurnStoppedMessage(
+                content=aggregation, interrupted=True, timestamp=timestamp
+            )
+            await self._call_event_handler("on_assistant_turn_stopped", message)
+            if aggregation:
+                await self._call_event_handler("on_assistant_message_added", message)
+            return
+
+        # Normal end. Park the message for trailing write.
+        raw_aggregation = self.aggregation_string()
+        if raw_aggregation:
+            self._pending_assistant_message_to_flush = {
+                "raw": raw_aggregation,
+                "timestamp": timestamp,
+            }
+        await self.reset()
+        stripped = (
+            self._maybe_strip_turn_completion_markers(raw_aggregation) if raw_aggregation else ""
+        )
+        message = AssistantTurnStoppedMessage(
+            content=stripped, interrupted=False, timestamp=timestamp
+        )
+        await self._call_event_handler("on_assistant_turn_stopped", message)
+
+    async def _realtime_handoff_flush(self) -> None:
+        """Flush pending assistant aggregation to context.
+
+        Called by the paired user half from
+        ``_realtime_handle_transcription`` when a new transcript arrives,
+        committing the assistant's deferred message before the user
+        starts a new turn. No-op when nothing is pending.
+        """
+        if self._pending_assistant_message_to_flush is None:
+            return
+        pending = self._pending_assistant_message_to_flush
+        self._pending_assistant_message_to_flush = None
+
+        raw = pending["raw"]
+        timestamp = pending["timestamp"]
+
+        # Mirror push_aggregation: write the raw aggregation (with any
+        # turn-completion markers intact) to context, emit LLMContextFrame
+        # and the timestamp frame. Markers are stripped only from the
+        # event-carried text.
+        self._context.add_message({"role": "assistant", "content": raw})
+        await self.push_context_frame()
+        timestamp_frame = LLMContextAssistantTimestampFrame(timestamp=time_now_iso8601())
+        await self.push_frame(timestamp_frame)
+
+        stripped = self._maybe_strip_turn_completion_markers(raw)
+        message = AssistantTurnStoppedMessage(
+            content=stripped, interrupted=False, timestamp=timestamp
+        )
+        await self._call_event_handler("on_assistant_message_added", message)
+
     def _maybe_strip_turn_completion_markers(self, text: str) -> str:
         """Strip turn completion markers from assistant transcript.
 
@@ -1685,6 +2133,7 @@ class LLMContextAggregatorPair:
         user_params: LLMUserAggregatorParams | None = None,
         assistant_params: LLMAssistantAggregatorParams | None = None,
         add_tool_change_messages: bool | None = None,
+        realtime_service_mode: RealtimeServiceModeConfig | None = None,
     ):
         """Initialize the LLM context aggregator pair.
 
@@ -1702,14 +2151,38 @@ class LLMContextAggregatorPair:
                 announcement is added exactly once (the second aggregator's
                 diff is empty by the time it sees the frame). Leave as
                 ``None`` to respect per-params settings.
+            realtime_service_mode: When provided, configures the pair for
+                use with a realtime (speech-to-speech) LLM service.
+                Context writes become trailing — driven by the content
+                stream itself (transcripts, ``LLMFullResponseStartFrame``)
+                rather than turn frames — and, by default, turn-end
+                strategies stop waiting for transcripts. Both halves share
+                this configuration via a private channel; mismatched
+                halves are rejected at ``StartFrame``. Defaults to
+                ``None``, which preserves cascade behavior.
         """
         user_params = user_params or LLMUserAggregatorParams()
         assistant_params = assistant_params or LLMAssistantAggregatorParams()
         if add_tool_change_messages is not None:
             user_params.add_tool_change_messages = add_tool_change_messages
             assistant_params.add_tool_change_messages = add_tool_change_messages
-        self._user = LLMUserAggregator(context, params=user_params)
+
-        self._assistant = LLMAssistantAggregator(context, params=assistant_params)
+        pair_lock = asyncio.Lock() if realtime_service_mode is not None else None
+        self._user = LLMUserAggregator(
+            context,
+            params=user_params,
+            _realtime_service_mode=realtime_service_mode,
+            _pair_lock=pair_lock,
+        )
+        self._assistant = LLMAssistantAggregator(
+            context,
+            params=assistant_params,
+            _realtime_service_mode=realtime_service_mode,
+            _pair_lock=pair_lock,
+        )
+        # Wire the cross-half back-references after both halves exist.
+        self._user._paired_half = self._assistant
+        self._assistant._paired_half = self._user
 
     def user(self) -> LLMUserAggregator:
         """Get the user context aggregator.

tests/test_context_aggregators_universal.py

@@ -4,6 +4,7 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
+import asyncio
 import json
 import unittest
 
@@ -33,6 +34,7 @@ from pipecat.frames.frames import (
     LLMThoughtEndFrame,
     LLMThoughtStartFrame,
     LLMThoughtTextFrame,
+    RealtimeServiceMetadataFrame,
     SpeechControlParamsFrame,
     StartFrame,
     TextFrame,
@@ -55,6 +57,8 @@ from pipecat.processors.aggregators.llm_response_universal import (
     LLMContextAggregatorPair,
     LLMUserAggregator,
     LLMUserAggregatorParams,
+    RealtimeServiceModeConfig,
+    UserTurnStoppedMessage,
 )
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.tests.utils import SleepFrame, run_test
@@ -63,6 +67,10 @@ from pipecat.turns.user_mute import (
     FunctionCallUserMuteStrategy,
     MuteUntilFirstBotCompleteUserMuteStrategy,
 )
+from pipecat.turns.user_start import (
+    TranscriptionUserTurnStartStrategy,
+    VADUserTurnStartStrategy,
+)
 from pipecat.turns.user_stop import SpeechTimeoutUserTurnStopStrategy
 from pipecat.turns.user_turn_strategies import (
     FilterIncompleteUserTurnStrategies,
@@ -1651,5 +1659,314 @@ class TestToolChangeMessages(unittest.IsolatedAsyncioTestCase):
         self.assertFalse(pair.assistant()._add_tool_change_messages)
 
 
+class TestRealtimeServiceModeConfig(unittest.TestCase):
+    def test_default_fields_are_realtime(self):
+        cfg = RealtimeServiceModeConfig()
+        self.assertFalse(cfg.context_writes_await_turns)
+        self.assertFalse(cfg.turns_await_transcripts)
+
+    def test_keep_transcripts_keep_writes_on_turn(self):
+        cfg = RealtimeServiceModeConfig(
+            turns_await_transcripts=True, context_writes_await_turns=True
+        )
+        self.assertTrue(cfg.context_writes_await_turns)
+        self.assertTrue(cfg.turns_await_transcripts)
+
+    def test_keep_transcripts_trailing_writes(self):
+        # Valid third row: turns wait on transcripts but context writes
+        # are trailing. The plan calls this out as the explicit fine-grained
+        # case (downstream consumers of user-turn frames want transcripts).
+        cfg = RealtimeServiceModeConfig(turns_await_transcripts=True)
+        self.assertFalse(cfg.context_writes_await_turns)
+        self.assertTrue(cfg.turns_await_transcripts)
+
+    def test_invalid_combination_rejected(self):
+        # turns fire early but context writes wait → incomplete messages.
+        with self.assertRaises(ValueError):
+            RealtimeServiceModeConfig(
+                turns_await_transcripts=False, context_writes_await_turns=True
+            )
+
+
+class TestRealtimeServiceModeAggregator(unittest.IsolatedAsyncioTestCase):
+    """End-to-end tests for the trailing-write realtime mode."""
+
+    def _build_pair(
+        self,
+        *,
+        realtime_service_mode: RealtimeServiceModeConfig | None = None,
+        user_params: LLMUserAggregatorParams | None = None,
+    ) -> tuple[LLMContext, LLMContextAggregatorPair]:
+        context = LLMContext()
+        pair = LLMContextAggregatorPair(
+            context,
+            user_params=user_params,
+            realtime_service_mode=realtime_service_mode,
+        )
+        return context, pair
+
+    async def test_pair_propagates_realtime_mode_to_halves(self):
+        _, pair = self._build_pair(realtime_service_mode=RealtimeServiceModeConfig())
+        # The pair wires shared state into both halves.
+        self.assertIs(pair.user()._paired_half, pair.assistant())
+        self.assertIs(pair.assistant()._paired_half, pair.user())
+        self.assertIs(pair.user()._pair_lock, pair.assistant()._pair_lock)
+        self.assertFalse(pair.user()._context_writes_await_turns)
+        self.assertFalse(pair.user()._turns_await_transcripts)
+        self.assertFalse(pair.assistant()._context_writes_await_turns)
+        self.assertFalse(pair.assistant()._turns_await_transcripts)
+
+    async def test_pair_omits_realtime_wiring_when_unset(self):
+        _, pair = self._build_pair()
+        # Backreferences are still created (harmless), but no shared lock
+        # is allocated when the realtime config is absent.
+        self.assertIsNone(pair.user()._pair_lock)
+        self.assertIsNone(pair.assistant()._pair_lock)
+        self.assertTrue(pair.user()._context_writes_await_turns)
+        self.assertTrue(pair.assistant()._context_writes_await_turns)
+
+    async def test_realtime_strategy_mutations_with_defaults(self):
+        _, pair = self._build_pair(realtime_service_mode=RealtimeServiceModeConfig())
+        # The mutated strategies live on the UserTurnController owned by
+        # the user aggregator.
+        strategies = pair.user()._user_turn_controller._user_turn_strategies
+        # TranscriptionUserTurnStartStrategy is dropped.
+        for s in strategies.start:
+            self.assertNotIsInstance(s, TranscriptionUserTurnStartStrategy)
+        # VAD start strategy is preserved.
+        self.assertTrue(any(isinstance(s, VADUserTurnStartStrategy) for s in strategies.start))
+        # Stop strategies that expose wait_for_transcript have it flipped.
+        for s in strategies.stop:
+            if hasattr(s, "wait_for_transcript"):
+                self.assertFalse(s.wait_for_transcript)
+
+    async def test_realtime_strategy_mutations_skipped_when_turns_await_transcripts(self):
+        _, pair = self._build_pair(
+            realtime_service_mode=RealtimeServiceModeConfig(turns_await_transcripts=True),
+        )
+        strategies = pair.user()._user_turn_controller._user_turn_strategies
+        # When turns still wait for transcripts, the transcript start
+        # strategy stays in the chain.
+        self.assertTrue(
+            any(isinstance(s, TranscriptionUserTurnStartStrategy) for s in strategies.start)
+        )
+
+    async def test_trailing_write_user_then_assistant_then_user(self):
+        _, pair = self._build_pair(realtime_service_mode=RealtimeServiceModeConfig())
+        user, assistant = pair
+
+        user_msg_added: list[UserTurnStoppedMessage] = []
+        assistant_msg_added: list[AssistantTurnStoppedMessage] = []
+
+        @user.event_handler("on_user_message_added")
+        async def _on_um(_a, msg):
+            user_msg_added.append(msg)
+
+        @assistant.event_handler("on_assistant_message_added")
+        async def _on_am(_a, msg):
+            assistant_msg_added.append(msg)
+
+        context = user.context
+
+        # Sequence: user transcript, assistant response starts (flushes
+        # user), assistant response ends (parks pending), new user
+        # transcript (flushes assistant), then EndFrame flushes the new
+        # user message.
+        frames_to_send = [
+            TranscriptionFrame(text="Hello!", user_id="", timestamp="now"),
+            SleepFrame(),
+            LLMFullResponseStartFrame(),
+            LLMTextFrame("Hi "),
+            LLMTextFrame("there!"),
+            LLMFullResponseEndFrame(),
+            SleepFrame(),
+            TranscriptionFrame(text="How are you?", user_id="", timestamp="now"),
+            SleepFrame(),
+        ]
+        await run_test(
+            Pipeline([user, assistant]),
+            frames_to_send=frames_to_send,
+        )
+
+        # Context should contain: user("Hello!"), assistant("Hi there!"),
+        # user("How are you?").
+        messages = context.get_messages()
+        roles_contents = [(m["role"], m["content"]) for m in messages]
+        self.assertEqual(
+            roles_contents,
+            [
+                ("user", "Hello!"),
+                ("assistant", "Hi there!"),
+                ("user", "How are you?"),
+            ],
+        )
+        self.assertEqual([m.content for m in user_msg_added], ["Hello!", "How are you?"])
+        self.assertEqual([m.content for m in assistant_msg_added], ["Hi there!"])
+        for msg in assistant_msg_added:
+            self.assertFalse(msg.interrupted)
+
+    async def test_interruption_writes_assistant_immediately(self):
+        _, pair = self._build_pair(realtime_service_mode=RealtimeServiceModeConfig())
+        user, assistant = pair
+
+        assistant_messages: list[AssistantTurnStoppedMessage] = []
+
+        @assistant.event_handler("on_assistant_message_added")
+        async def _on_am(_a, msg):
+            assistant_messages.append(msg)
+
+        context = user.context
+
+        frames_to_send = [
+            TranscriptionFrame(text="Hi!", user_id="", timestamp="now"),
+            LLMFullResponseStartFrame(),
+            LLMTextFrame("Hello "),
+            SleepFrame(),
+            InterruptionFrame(),
+        ]
+        await run_test(
+            Pipeline([user, assistant]),
+            frames_to_send=frames_to_send,
+        )
+
+        roles_contents = [(m["role"], m["content"]) for m in context.get_messages()]
+        # User message written when assistant started; assistant message
+        # written immediately on interruption with interrupted=True.
+        self.assertEqual(roles_contents, [("user", "Hi!"), ("assistant", "Hello")])
+        self.assertEqual(len(assistant_messages), 1)
+        self.assertTrue(assistant_messages[0].interrupted)
+
+    async def test_user_turn_stopped_in_realtime_mode_has_none_content(self):
+        # When VAD turn frames fire in realtime mode, the user-turn-stop
+        # message carries content=None — the message isn't finalized yet.
+        _, pair = self._build_pair(
+            realtime_service_mode=RealtimeServiceModeConfig(),
+            user_params=LLMUserAggregatorParams(
+                user_turn_strategies=UserTurnStrategies(
+                    stop=[
+                        SpeechTimeoutUserTurnStopStrategy(
+                            user_speech_timeout=TRANSCRIPTION_TIMEOUT,
+                        )
+                    ],
+                ),
+                user_turn_stop_timeout=USER_TURN_STOP_TIMEOUT,
+            ),
+        )
+        user, assistant = pair
+
+        stop_messages: list[UserTurnStoppedMessage] = []
+
+        @user.event_handler("on_user_turn_stopped")
+        async def _on_stop(_a, _s, msg):
+            stop_messages.append(msg)
+
+        frames_to_send = [
+            VADUserStartedSpeakingFrame(),
+            TranscriptionFrame(text="hey", user_id="", timestamp="now"),
+            VADUserStoppedSpeakingFrame(),
+            SleepFrame(sleep=TRANSCRIPTION_TIMEOUT + 0.05),
+        ]
+        await run_test(
+            Pipeline([user, assistant]),
+            frames_to_send=frames_to_send,
+        )
+        self.assertEqual(len(stop_messages), 1)
+        self.assertIsNone(stop_messages[0].content)
+
+    async def test_realtime_metadata_recommendation_log_when_unconfigured(self):
+        # Cascade pair receiving a RealtimeServiceMetadataFrame logs the
+        # one-time recommendation. The user half records the fact via
+        # _realtime_recommendation_logged.
+        _, pair = self._build_pair()
+        user = pair.user()
+
+        frames_to_send = [
+            RealtimeServiceMetadataFrame(
+                service_name="FakeRealtimeLLM", emits_user_turn_frames=False
+            ),
+        ]
+        await run_test(Pipeline([pair.user(), pair.assistant()]), frames_to_send=frames_to_send)
+        self.assertTrue(user._realtime_recommendation_logged)
+
+    async def test_realtime_metadata_no_log_when_configured(self):
+        # When realtime mode is opted in, the metadata frame is consumed
+        # without firing the recommendation log (we still flag the
+        # one-shot bookkeeping).
+        _, pair = self._build_pair(realtime_service_mode=RealtimeServiceModeConfig())
+        user = pair.user()
+
+        frames_to_send = [
+            RealtimeServiceMetadataFrame(
+                service_name="FakeRealtimeLLM", emits_user_turn_frames=False
+            ),
+        ]
+        await run_test(Pipeline([pair.user(), pair.assistant()]), frames_to_send=frames_to_send)
+        self.assertTrue(user._realtime_recommendation_logged)
+
+    async def test_realtime_mode_requires_paired_half(self):
+        # Direct construction of a half with realtime mode set but no
+        # paired_half raises at StartFrame validation. We call the
+        # validation directly so the error isn't swallowed by the
+        # pipeline's exception handler.
+        context = LLMContext()
+        cfg = RealtimeServiceModeConfig()
+        user = LLMUserAggregator(context, _realtime_service_mode=cfg)
+        with self.assertRaises(RuntimeError):
+            user._validate_realtime_pairing()
+        assistant = LLMAssistantAggregator(context, _realtime_service_mode=cfg)
+        with self.assertRaises(RuntimeError):
+            assistant._validate_realtime_pairing()
+
+    async def test_realtime_mode_rejects_mismatched_halves(self):
+        # If a user code path constructs halves with mismatched configs,
+        # StartFrame validation catches it.
+        context = LLMContext()
+        lock = asyncio.Lock()
+        user = LLMUserAggregator(
+            context,
+            _realtime_service_mode=RealtimeServiceModeConfig(),
+            _pair_lock=lock,
+        )
+        assistant = LLMAssistantAggregator(
+            context,
+            _realtime_service_mode=RealtimeServiceModeConfig(turns_await_transcripts=True),
+            _pair_lock=lock,
+        )
+        user._paired_half = assistant
+        assistant._paired_half = user
+        with self.assertRaises(RuntimeError):
+            user._validate_realtime_pairing()
+
+    async def test_function_call_no_context_push_in_realtime_mode(self):
+        # Realtime services consume function results directly via
+        # FunctionCallResultFrame, so the aggregator should not push
+        # LLMContextFrame upstream after a function call result.
+        _, pair = self._build_pair(realtime_service_mode=RealtimeServiceModeConfig())
+        assistant = pair.assistant()
+        frames_to_send = [
+            FunctionCallInProgressFrame(
+                function_name="get_weather",
+                tool_call_id="1",
+                arguments={"location": "Los Angeles"},
+                cancel_on_interruption=True,
+            ),
+            SleepFrame(),
+            FunctionCallResultFrame(
+                function_name="get_weather",
+                tool_call_id="1",
+                arguments={"location": "Los Angeles"},
+                result={"conditions": "Sunny"},
+            ),
+            SleepFrame(),
+        ]
+        _, up_frames = await run_test(
+            assistant,
+            frames_to_send=frames_to_send,
+        )
+        # No LLMContextFrame should have been pushed upstream in
+        # realtime mode (cascade would push one to re-run inference).
+        self.assertFalse(any(isinstance(f, LLMContextFrame) for f in up_frames))
+
+
 if __name__ == "__main__":
     unittest.main()