Refactor delayed-transcript machinery; standardize vocabulary

Splits ``_maybe_emit_user_turn_stopped`` into three focused methods —
``_flush_user_message_to_context`` (push aggregation, return content +
timestamp), ``_finalize_user_turn`` (default-mode flow, emits both
events), and ``_finalize_delayed_user_message`` (delayed-mode flow,
emits only ``on_user_turn_message_finalized``). Fixes a side-issue
where ``on_user_turn_stopped`` could fire from non-end-of-turn paths
in delayed-transcript mode; that event now has a single origin (the
end-of-turn handler).

Standardizes vocabulary across docstrings and comments:

- "Default mode" / "Delayed-transcript mode" (with
  ``_expect_delayed_transcripts == False/True``)
- "End of turn" (not "audible stop" or "audible end of turn")
- "User message finalization" (the moment user-text is flushed to
  context + ``on_user_turn_message_finalized`` fires)
- "Pending finalization" (the in-between state in delayed mode)
- Transcripts (plural — the aggregator combines multiple per turn)

The timer that triggers user message finalization is no longer
described as a "backstop" — it's the sole trigger for finalization
in delayed-transcript mode, not a fallback. Renamed accordingly:
``_pending_finalization_task``, ``_pending_finalization_handler``,
``_run_pending_finalization``, ``_discard_pending_finalization``.

Adds a separate message class for the two events:
``UserTurnStoppedMessage.content`` is now ``str | None`` (``None``
at end-of-turn in delayed-transcript mode), and a new
``UserMessageFinalizedMessage`` carries the always-populated
``content`` for the finalization event.

changelog/4480.added.md

@@ -1 +1 @@
-- Added the configuration surface to drive a realtime service like Gemini Live from local turn detection without paying user-transcript latency. Cascaded pipelines wait for a transcript before ending the user's turn because the downstream LLM needs the user's words recorded in context — but that wait is pure latency in pipelines using local turn detection to drive a realtime service, which consumes user audio directly. Set `wait_for_transcript_to_end_user_turn=False` on `LLMUserAggregatorParams` to turn this on: the aggregator drops `TranscriptionUserTurnStartStrategy` from the start strategies (so late-arriving realtime transcripts don't trigger new turns), sets `wait_for_transcript=False` on any stop strategies that support it (so the turn ends without waiting for a transcript), fires `on_user_turn_stopped` on the audible end of the turn (with empty content, since the transcript hasn't arrived), and defers the context flush until the transcript arrives. A new `on_user_turn_message_finalized` event fires when the user's message has been written to context — in the default mode it coincides with `on_user_turn_stopped`; in the delayed-transcript mode it fires later. Consumers that want the populated transcript should subscribe to `on_user_turn_message_finalized`. See `examples/realtime/realtime-gemini-live-local-vad.py` for the full pattern.
+- Added the configuration surface to drive a realtime service like Gemini Live from local turn detection without paying user-transcript latency. Cascaded pipelines wait for transcripts before ending the user's turn because the downstream LLM needs the user's words recorded in context — but that wait is pure latency in pipelines using local turn detection to drive a realtime service, which consumes user audio directly. Set `wait_for_transcript_to_end_user_turn=False` on `LLMUserAggregatorParams` to turn this on: the aggregator drops `TranscriptionUserTurnStartStrategy` from the start strategies (so late-arriving realtime transcripts don't trigger new turns), sets `wait_for_transcript=False` on any stop strategies that support it (so the turn ends without waiting for transcripts), fires `on_user_turn_stopped` at the end of turn (with empty content, since no transcripts have arrived yet), and defers the context flush until after a short wait that gives the realtime service time to emit its transcripts. A new `on_user_turn_message_finalized` event fires when the user's message has been written to context — in the default mode it coincides with `on_user_turn_stopped`; in the delayed-transcript mode it fires later. Consumers that want the populated transcript should subscribe to `on_user_turn_message_finalized`. See `examples/realtime/realtime-gemini-live-local-vad.py` for the full pattern.

examples/realtime/realtime-gemini-live-local-vad.py

@@ -20,6 +20,7 @@ from pipecat.processors.aggregators.llm_response_universal import (
     AssistantTurnStoppedMessage,
     LLMContextAggregatorPair,
     LLMUserAggregatorParams,
+    UserMessageFinalizedMessage,
     UserTurnStoppedMessage,
 )
 from pipecat.runner.types import RunnerArguments
@@ -72,19 +73,19 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     )
     # `wait_for_transcript_to_end_user_turn=False` configures the user
     # aggregator for realtime services like Gemini Live that emit user
-    # transcripts after the audible end of the turn. With this flag the
+    # transcripts after the end of turn. With this flag the aggregator:
-    # aggregator:
     #
     # - drops `TranscriptionUserTurnStartStrategy` from the default start
     #   strategies (so late-arriving realtime transcripts don't trigger
     #   new turns),
     # - sets `wait_for_transcript=False` on the default stop strategy
-    #   (so the turn ends without waiting for a transcript),
+    #   (so the turn ends without waiting for transcripts),
-    # - fires `on_user_turn_stopped` on the audible end of the turn with
+    # - fires `on_user_turn_stopped` at the end of turn with empty
-    #   empty `message.content` (the transcript hasn't arrived yet), and
+    #   `message.content` (no transcripts have arrived yet), and
-    # - defers the context flush until the (late) transcript arrives, then
+    # - defers the context flush until after a short wait that gives the
-    #   emits `on_user_turn_message_finalized` with the populated message
+    #   realtime service time to emit its transcripts, then emits
-    #   so the user's words land in the LLM context for audit/history.
+    #   `on_user_turn_message_finalized` with the populated message so
+    #   the user's words land in the LLM context for audit/history.
     user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
         context,
         user_params=LLMUserAggregatorParams(
@@ -124,18 +125,20 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         await task.cancel()
 
     # With `wait_for_transcript_to_end_user_turn=False`, `on_user_turn_stopped`
-    # fires on the audible end of the turn (before the transcript arrives), so
+    # fires at the end of turn (before any transcripts arrive), so its
-    # its `message.content` is empty. Logged here to make the timing of the
+    # `message.content` is empty. Logged here to make the timing of the
-    # audible-stop signal visible alongside the later transcript event.
+    # end-of-turn signal visible alongside the later finalization event.
     @user_aggregator.event_handler("on_user_turn_stopped")
     async def on_user_turn_stopped(aggregator, strategy, message: UserTurnStoppedMessage):
-        logger.info(f"User turn ended (audible, strategy: {type(strategy).__name__})")
+        logger.info(f"User turn ended (strategy: {type(strategy).__name__})")
 
-    # `on_user_turn_message_finalized` fires when the user transcript has
+    # `on_user_turn_message_finalized` fires when the user message has
     # been written to context — later than `on_user_turn_stopped` in this
-    # mode, since transcripts arrive after the audible turn end.
+    # mode, since transcripts arrive after the end of turn.
     @user_aggregator.event_handler("on_user_turn_message_finalized")
-    async def on_user_turn_message_finalized(aggregator, strategy, message: UserTurnStoppedMessage):
+    async def on_user_turn_message_finalized(
+        aggregator, strategy, message: UserMessageFinalizedMessage
+    ):
         timestamp = f"[{message.timestamp}] " if message.timestamp else ""
         line = f"{timestamp}user: {message.content}"
         logger.info(f"Transcript: {line}")

src/pipecat/processors/aggregators/llm_response_universal.py

@@ -130,15 +130,16 @@ class LLMUserAggregatorParams:
         vad_analyzer: Voice Activity Detection analyzer instance.
         wait_for_transcript_to_end_user_turn: Defaults to True. Set to
             False when using local turn detection to drive a realtime
-            service (e.g. Gemini Live), where waiting for the transcript
+            service (e.g. Gemini Live), where waiting for transcripts
             before ending the turn is pure latency. When False:
 
             - the user turn ends as soon as the user stops speaking,
               without waiting for transcripts: ``on_user_turn_stopped``
               fires immediately with empty content
-            - the user-text flush to context and the
+            - user message finalization (the user-text flush to context
-              ``on_user_turn_message_finalized`` event are deferred
+              and the ``on_user_turn_message_finalized`` event) is
-              until a backstop timer expires
+              deferred and runs after a short wait, giving the realtime
+              service time to emit its transcripts
 
             As part of configuring this behavior, the aggregator drops
             ``TranscriptionUserTurnStartStrategy`` from start strategies
@@ -278,13 +279,40 @@ class LLMAssistantAggregatorParams:
 
 @dataclass
 class UserTurnStoppedMessage:
-    """A user turn stopped message containing a user transcript update.
+    """A message accompanying ``on_user_turn_stopped`` (end of user turn).
 
-    A message in a conversation transcript containing the user content. This is
+    In default mode (``wait_for_transcript_to_end_user_turn=True``), the
-    the aggregated transcript that is then used in the context.
+    user message is finalized at the end of the turn, so ``content``
+    carries the aggregated transcript. In delayed-transcript mode
+    (``wait_for_transcript_to_end_user_turn=False``), transcripts arrive
+    later, so ``content`` is ``None`` here — subscribe to
+    ``on_user_turn_message_finalized`` for the populated message.
 
     Parameters:
-        content: The message content/text.
+        content: The aggregated user transcript, or ``None`` if no
+            transcripts had arrived when end of turn fired (delayed-
+            transcript mode).
+        timestamp: When the user turn started.
+        user_id: Optional identifier for the user.
+
+    """
+
+    content: str | None
+    timestamp: str
+    user_id: str | None = None
+
+
+@dataclass
+class UserMessageFinalizedMessage:
+    """A message accompanying ``on_user_turn_message_finalized``.
+
+    Fired when the user message has been finalized into the context.
+    In default mode this coincides with ``on_user_turn_stopped``; in
+    delayed-transcript mode it fires later, once the realtime service
+    has emitted its transcripts. ``content`` is always populated.
+
+    Parameters:
+        content: The aggregated user transcript.
         timestamp: When the user turn started.
         user_id: Optional identifier for the user.
 
@@ -545,19 +573,23 @@ class LLMUserAggregator(LLMContextAggregator):
 
     Event handlers available:
 
-    - on_user_turn_started: Called when the user turn starts
+    - on_user_turn_started: Called when the user turn starts.
-    - on_user_turn_stopped: Called when the user turn ends. With
+    - on_user_turn_stopped: Called at the end of turn, with a
-      ``wait_for_transcript_to_end_user_turn=True`` (default), this fires
+      ``UserTurnStoppedMessage``. In default mode
-      with the populated ``UserTurnStoppedMessage.content``. With
+      (``wait_for_transcript_to_end_user_turn=True``)
-      ``wait_for_transcript_to_end_user_turn=False`` (realtime + local
+      ``message.content`` carries the aggregated transcript. In
-      turn detection), this fires on the audible end of the turn — before
+      delayed-transcript mode
-      the transcript has arrived — so ``content`` is empty; subscribe to
+      (``wait_for_transcript_to_end_user_turn=False``) the user's
-      ``on_user_turn_message_finalized`` to get the transcript.
+      transcripts haven't arrived yet, so ``message.content`` is
-    - on_user_turn_message_finalized: Called once the user message has
+      ``None``; subscribe to ``on_user_turn_message_finalized`` to get
-      been finalized into the context. Fires in both modes — in the
+      the assembled message.
-      default mode it coincides with ``on_user_turn_stopped``; in the
+    - on_user_turn_message_finalized: Called at user message
-      delayed-transcript path it fires later, when the transcript
+      finalization — when the user message has been flushed to the
-      arrives. The ``UserTurnStoppedMessage.content`` is always populated.
+      context — with a ``UserMessageFinalizedMessage``. In default
+      mode it coincides with ``on_user_turn_stopped``; in
+      delayed-transcript mode it fires a short time after, once the
+      realtime service has had time to emit its transcripts.
+      ``message.content`` is always populated.
     - on_user_turn_stop_timeout: Called when no user turn stop strategy triggers
     - on_user_turn_idle: Called when the user has been idle for the configured timeout
     - on_user_mute_started: Called when the user becomes muted
@@ -574,7 +606,7 @@ class LLMUserAggregator(LLMContextAggregator):
             ...
 
         @aggregator.event_handler("on_user_turn_message_finalized")
-        async def on_user_turn_message_finalized(aggregator, strategy: BaseUserTurnStopStrategy, message: UserTurnStoppedMessage):
+        async def on_user_turn_message_finalized(aggregator, strategy: BaseUserTurnStopStrategy, message: UserMessageFinalizedMessage):
             ...
 
         @aggregator.event_handler("on_user_turn_stop_timeout")
@@ -662,14 +694,14 @@ class LLMUserAggregator(LLMContextAggregator):
         # inferences fire before finalization.
         self._full_user_turn_aggregation: str | None = None
 
-        # State for the deferred-finalization path that activates when
+        # Pending-finalization state used in delayed-transcript mode
-        # `wait_for_transcript_to_end_user_turn=False`. The strategy fires
+        # (`_expect_delayed_transcripts == True`): the end-of-turn event
-        # turn-stopped on the audible end of the user's turn; we defer the
+        # `on_user_turn_stopped` has fired (with empty content) and user
-        # context flush and `on_user_turn_stopped` event until the transcript
+        # message finalization is scheduled to run after the
-        # arrives (or a backstop timer fires).
+        # pending-finalization timer expires.
         self._pending_stop_strategy: BaseUserTurnStopStrategy | None = None
         self._pending_inference_trigger: bool = False
-        self._delayed_transcript_backstop_task: asyncio.Task | None = None
+        self._pending_finalization_task: asyncio.Task | None = None
 
         self._user_turn_controller = UserTurnController(
             user_turn_strategies=user_turn_strategies,
@@ -715,12 +747,17 @@ class LLMUserAggregator(LLMContextAggregator):
 
     @property
     def _expect_delayed_transcripts(self) -> bool:
-        """Internal alias for the deferred-finalization mode.
+        """True in delayed-transcript mode, False in default mode.
 
-        Active whenever ``wait_for_transcript_to_end_user_turn`` is False,
+        In delayed-transcript mode the end of turn and user message
-        which is also when the strategy-mutation bundle is applied. The two
+        finalization happen at different times: ``on_user_turn_stopped``
-        always travel together — there's no configuration where one is
+        fires immediately at the end of turn (with empty content), and
-        active without the other.
+        user message finalization is scheduled to run after the
+        pending-finalization timer expires, giving the realtime service
+        time to emit its transcripts.
+
+        Internal alias for ``wait_for_transcript_to_end_user_turn=False``.
+        Always travels with the strategy-mutation bundle applied at init.
         """
         return not self._params.wait_for_transcript_to_end_user_turn
 
@@ -879,17 +916,20 @@ class LLMUserAggregator(LLMContextAggregator):
         await self._cleanup()
 
     async def _finalize_on_session_end(self):
-        """Flush any pending turn-stop on session end.
+        """Flush any pending user message on session end.
 
-        If a deferred turn finalization is pending (because
+        If user message finalization is pending (delayed-transcript mode,
-        ``expect_delayed_transcripts`` is True and the transcript hadn't
+        transcript hadn't arrived yet), replay the pending finalization
-        arrived yet), replay it first so the user message is captured before
+        so the user message is captured before the session shuts down.
-        the session shuts down.
+        Otherwise, run the mode-appropriate finalize path on whatever's
+        currently in the buffer.
         """
         if self._pending_stop_strategy is not None or self._pending_inference_trigger:
-            await self._replay_deferred_turn_finalization(on_session_end=True)
+            await self._run_pending_finalization(on_session_end=True)
+        elif self._expect_delayed_transcripts:
+            await self._finalize_delayed_user_message(on_session_end=True)
         else:
-            await self._maybe_emit_user_turn_stopped(on_session_end=True)
+            await self._finalize_user_turn(on_session_end=True)
 
     async def _cleanup(self):
         if self._vad_controller:
@@ -1021,16 +1061,17 @@ class LLMUserAggregator(LLMContextAggregator):
     ):
         logger.debug(f"{self}: User started speaking (strategy: {strategy})")
 
-        # Precondition guard for `expect_delayed_transcripts`: if the previous
+        # Precondition guard for delayed-transcript mode: if the previous
-        # turn's finalization is still pending, the precondition (turn N's
+        # turn's user message finalization is still pending, the
-        # transcript arrives before turn N+1 starts) has been violated.
+        # precondition (turn N's transcripts arrive before turn N+1
-        # Force-flush the previous turn before proceeding.
+        # starts) has been violated. Force-finalize the previous turn
+        # before proceeding.
         if self._pending_stop_strategy is not None or self._pending_inference_trigger:
             logger.warning(
                 f"{self}: user turn started while previous turn's transcript was "
                 f"still pending; flushing previous turn now"
             )
-            await self._replay_deferred_turn_finalization()
+            await self._run_pending_finalization()
 
         self._user_turn_start_timestamp = time_now_iso8601()
         self._full_user_turn_aggregation = None
@@ -1053,8 +1094,9 @@ class LLMUserAggregator(LLMContextAggregator):
         logger.debug(f"{self}: User turn inference triggered (strategy: {strategy})")
 
         if self._expect_delayed_transcripts:
-            # Defer the push_aggregation and event emission until the
+            # Defer the push_aggregation and event emission; they'll run
-            # transcript arrives (or the backstop fires).
+            # alongside user message finalization when the
+            # pending-finalization timer expires.
             self._pending_inference_trigger = True
             return
 
@@ -1083,60 +1125,67 @@ class LLMUserAggregator(LLMContextAggregator):
     ):
         logger.debug(f"{self}: User stopped speaking (strategy: {strategy})")
 
-        # Audible-stop side effects fire on the strategy event regardless of
+        # End-of-turn side effects always fire on the strategy event,
-        # whether the transcript-ready finalization is deferred.
+        # regardless of whether user message finalization is deferred.
         if params.enable_user_speaking_frames:
             await self.broadcast_frame(UserStoppedSpeakingFrame)
 
         await self._user_idle_controller.process_frame(UserStoppedSpeakingFrame())
 
         if self._expect_delayed_transcripts:
-            # Fire `on_user_turn_stopped` now to signal the audible end of the
+            # Delayed-transcript mode: fire `on_user_turn_stopped` now
-            # turn. `content` is empty because the transcript hasn't arrived
+            # for the end of turn — content is empty because no
-            # yet — consumers that want the transcript subscribe to
+            # transcripts have arrived yet. User message finalization
-            # `on_user_turn_message_finalized` instead, which fires later when
+            # is scheduled to run when the pending-finalization timer
-            # the transcript lands (or the backstop timer fires).
+            # expires; consumers wanting the assembled message subscribe
-            audible_stop_message = UserTurnStoppedMessage(
+            # to `on_user_turn_message_finalized`.
-                content="", timestamp=self._user_turn_start_timestamp
+            end_of_turn_message = UserTurnStoppedMessage(
+                content=None, timestamp=self._user_turn_start_timestamp
             )
-            await self._call_event_handler("on_user_turn_stopped", strategy, audible_stop_message)
+            await self._call_event_handler("on_user_turn_stopped", strategy, end_of_turn_message)
 
             self._pending_stop_strategy = strategy
-            self._delayed_transcript_backstop_task = self.create_task(
+            self._pending_finalization_task = self.create_task(
-                self._delayed_transcript_backstop_handler(DEFAULT_TTFS_P99),
+                self._pending_finalization_handler(DEFAULT_TTFS_P99),
-                f"{self}::delayed_transcript_backstop",
+                f"{self}::pending_finalization",
             )
             return
 
-        await self._maybe_emit_user_turn_stopped(strategy)
+        await self._finalize_user_turn(strategy)
 
-    async def _delayed_transcript_backstop_handler(self, timeout: float):
+    async def _pending_finalization_handler(self, timeout: float):
-        """Backstop timer for `expect_delayed_transcripts`.
+        """Pending-finalization timer for delayed-transcript mode.
 
-        Fires after ``timeout`` seconds and replays the deferred turn
+        Waits ``timeout`` seconds — giving the realtime service time to
-        finalization with whatever transcript has been captured by then
+        emit its transcripts — then runs the pending finalization with
-        (possibly nothing).
+        whatever transcripts have been captured by then (possibly
+        nothing). Cancelled by reset / next-turn precondition guard /
+        session end.
         """
         try:
             await asyncio.sleep(timeout)
         except asyncio.CancelledError:
             return
         finally:
-            self._delayed_transcript_backstop_task = None
+            self._pending_finalization_task = None
 
-        await self._replay_deferred_turn_finalization()
+        await self._run_pending_finalization()
 
-    async def _replay_deferred_turn_finalization(self, *, on_session_end: bool = False):
+    async def _run_pending_finalization(self, *, on_session_end: bool = False):
-        """Replay the deferred halves of inference-triggered and turn-stopped.
+        """Run the pending finalization for delayed-transcript mode.
 
-        Called from the backstop timer, the precondition guard in
+        In delayed-transcript mode the end of turn fires
-        `_on_user_turn_started`, and the session-end paths. Clears all
+        ``on_user_turn_stopped`` immediately and leaves user message
-        pending state before delegating to the normal finalization methods
+        finalization (plus any pending inference-triggered segment)
-        so they run the unmodified code path.
+        pending. This method runs that pending work: pushes the
+        accumulated user message to context and emits
+        ``on_user_turn_message_finalized``. Called from the
+        pending-finalization timer (the normal path), the precondition
+        guard in ``_on_user_turn_started``, and the session-end paths.
         """
-        if self._delayed_transcript_backstop_task:
+        if self._pending_finalization_task:
-            await self.cancel_task(self._delayed_transcript_backstop_task)
+            await self.cancel_task(self._pending_finalization_task)
-            self._delayed_transcript_backstop_task = None
+            self._pending_finalization_task = None
 
         pending_strategy = self._pending_stop_strategy
         had_pending_inference = self._pending_inference_trigger
@@ -1155,31 +1204,29 @@ class LLMUserAggregator(LLMContextAggregator):
             await self._call_event_handler("on_user_turn_inference_triggered", pending_strategy)
 
         if pending_strategy is not None or on_session_end:
-            # `on_user_turn_stopped` already fired on the audible stop;
+            # `on_user_turn_stopped` already fired at the end of turn;
-            # only fire `on_user_turn_message_finalized` here.
+            # this is the deferred user message finalization.
-            await self._maybe_emit_user_turn_stopped(
+            await self._finalize_delayed_user_message(
-                pending_strategy,
+                pending_strategy, on_session_end=on_session_end
-                on_session_end=on_session_end,
-                emit_stopped_event=False,
             )
 
     async def _on_reset_aggregation(
         self, controller: UserTurnController, strategy: BaseUserTurnStartStrategy
     ):
         logger.debug(f"{self}: Resetting aggregation (strategy: {strategy})")
-        await self._discard_deferred_turn_finalization()
+        await self._discard_pending_finalization()
         await self.reset()
 
-    async def _discard_deferred_turn_finalization(self):
+    async def _discard_pending_finalization(self):
-        """Drop any pending deferred turn finalization without replaying.
+        """Drop pending finalization state without running it.
 
-        Called from reset paths (interruption, explicit reset). "Reset" means
+        Called from reset paths (interruption, explicit reset). "Reset"
-        "throw it away" — we don't want to flush a partial transcript that
+        means "throw it away" — we don't flush a partial transcript that
         was about to be invalidated anyway.
         """
-        if self._delayed_transcript_backstop_task:
+        if self._pending_finalization_task:
-            await self.cancel_task(self._delayed_transcript_backstop_task)
+            await self.cancel_task(self._pending_finalization_task)
-            self._delayed_transcript_backstop_task = None
+            self._pending_finalization_task = None
         self._pending_stop_strategy = None
         self._pending_inference_trigger = False
 
@@ -1189,32 +1236,22 @@ class LLMUserAggregator(LLMContextAggregator):
     async def _on_user_turn_idle(self, controller):
         await self._call_event_handler("on_user_turn_idle")
 
-    async def _maybe_emit_user_turn_stopped(
+    async def _flush_user_message_to_context(
-        self,
+        self, on_session_end: bool = False
-        strategy: BaseUserTurnStopStrategy | None = None,
+    ) -> tuple[str, str] | None:
-        on_session_end: bool = False,
+        """Push the aggregated user message to context, return ``(content, timestamp)``.
-        *,
-        emit_stopped_event: bool = True,
-    ):
-        """Finalize the user message and emit the corresponding events.
 
-        Earlier inference triggers in the same turn have already pushed
+        Earlier inference triggers in the same turn already pushed their
-        their segments to the context and accumulated them into
+        segments to the context and accumulated them in
-        ``self._full_user_turn_aggregation``. Any aggregation that
+        ``self._full_user_turn_aggregation``; whatever arrived after the
-        arrived after the last inference trigger is flushed here so
+        last inference trigger is flushed here so end-of-turn content is
-        end-of-turn content is never lost from the public events.
+        never lost.
 
-        Always emits ``on_user_turn_message_finalized``. Also emits
+        Returns ``(content, timestamp)`` for the just-finalized user
-        ``on_user_turn_stopped`` when ``emit_stopped_event=True`` (the
+        message, or ``None`` when there's no content to flush and
-        default path); the delayed-transcript path passes False because it
+        ``on_session_end=True`` (avoids emitting empty events during
-        already fired ``on_user_turn_stopped`` on the audible stop.
+        session shutdown). Callers construct the appropriate message
-
+        dataclass for each event they emit.
-        Args:
-            strategy: The strategy that triggered the turn stop.
-            on_session_end: If True, only emit if there's unemitted content
-                (avoids duplicate events when session ends).
-            emit_stopped_event: Whether to fire ``on_user_turn_stopped``
-                alongside ``on_user_turn_message_finalized``.
         """
         segment = await self.push_aggregation()
         full_aggregation = self._full_user_turn_aggregation
@@ -1225,14 +1262,51 @@ class LLMUserAggregator(LLMContextAggregator):
         else:
             content = full_aggregation or segment
 
-        if not on_session_end or content:
+        if on_session_end and not content:
-            message = UserTurnStoppedMessage(
+            return None
-                content=content, timestamp=self._user_turn_start_timestamp
+
-            )
+        timestamp = self._user_turn_start_timestamp
-            if emit_stopped_event:
+        self._user_turn_start_timestamp = ""
-                await self._call_event_handler("on_user_turn_stopped", strategy, message)
+        return content, timestamp
-            await self._call_event_handler("on_user_turn_message_finalized", strategy, message)
+
-            self._user_turn_start_timestamp = ""
+    async def _finalize_user_turn(
+        self,
+        strategy: BaseUserTurnStopStrategy | None = None,
+        on_session_end: bool = False,
+    ):
+        """Finalize the user turn: flush the message, emit both events.
+
+        Used in default mode (``_expect_delayed_transcripts == False``),
+        where end of turn and user message finalization coincide. Emits
+        both ``on_user_turn_stopped`` and ``on_user_turn_message_finalized``.
+        """
+        result = await self._flush_user_message_to_context(on_session_end=on_session_end)
+        if result is None:
+            return
+        content, timestamp = result
+        stopped_msg = UserTurnStoppedMessage(content=content, timestamp=timestamp)
+        finalized_msg = UserMessageFinalizedMessage(content=content, timestamp=timestamp)
+        await self._call_event_handler("on_user_turn_stopped", strategy, stopped_msg)
+        await self._call_event_handler("on_user_turn_message_finalized", strategy, finalized_msg)
+
+    async def _finalize_delayed_user_message(
+        self,
+        strategy: BaseUserTurnStopStrategy | None = None,
+        on_session_end: bool = False,
+    ):
+        """Finalize the user message: flush to context, emit one event.
+
+        Used in delayed-transcript mode (``_expect_delayed_transcripts ==
+        True``), where user message finalization fires after the end of
+        turn. Emits ``on_user_turn_message_finalized`` only;
+        ``on_user_turn_stopped`` was already emitted at the end of turn.
+        """
+        result = await self._flush_user_message_to_context(on_session_end=on_session_end)
+        if result is None:
+            return
+        content, timestamp = result
+        finalized_msg = UserMessageFinalizedMessage(content=content, timestamp=timestamp)
+        await self._call_event_handler("on_user_turn_message_finalized", strategy, finalized_msg)
 
 
 class LLMAssistantAggregator(LLMContextAggregator):

src/pipecat/turns/user_stop/speech_timeout_user_turn_stop_strategy.py

@@ -48,9 +48,10 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy):
     doesn't need the user's words recorded in context to respond — typically
     when using local turn detection to drive a realtime service like Gemini
     Live. In that case the strategy fires the turn stop as soon as the
-    user_speech_timeout elapses, without waiting for a transcript. Pair with
+    user_speech_timeout elapses, without waiting for transcripts. Pair with
-    ``expect_delayed_transcripts=True`` on ``LLMUserAggregatorParams`` so the
+    ``wait_for_transcript_to_end_user_turn=False`` on
-    aggregator still captures the transcript when it arrives.
+    ``LLMUserAggregatorParams`` so the aggregator still captures transcripts
+    when they arrive.
     """
 
     def __init__(

src/pipecat/turns/user_stop/turn_analyzer_user_turn_stop_strategy.py

@@ -47,10 +47,10 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy):
     doesn't need the user's words recorded in context to respond — typically
     when using local turn detection to drive a realtime service like Gemini
     Live. In that case the strategy fires the turn stop as soon as the
-    analyzer concludes the turn is complete, without waiting for a
+    analyzer concludes the turn is complete, without waiting for transcripts.
-    transcript. Pair with ``expect_delayed_transcripts=True`` on
+    Pair with ``wait_for_transcript_to_end_user_turn=False`` on
-    ``LLMUserAggregatorParams`` so the aggregator still captures the
+    ``LLMUserAggregatorParams`` so the aggregator still captures transcripts
-    transcript when it arrives.
+    when they arrive.
     """
 
     def __init__(