realtime + local turn detection: drop the user-transcript wait

Add 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. With that single flag 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 strategy that supports
  it (so the turn ends on the audible end of the turn, 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 or a backstop
  timer fires.

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` — it's the event that
always carries the user message, regardless of mode.

Strategy mutations are logged: loudly when the user passed their own
strategies (we're overwriting parts of their config), quietly
otherwise. The strategy-level `wait_for_transcript` parameter on
`TurnAnalyzerUserTurnStopStrategy` and `SpeechTimeoutUserTurnStopStrategy`
remains exposed for advanced cases.

The example `realtime-gemini-live-local-vad.py` demonstrates the full
pattern.

changelog/4480.added.md

@@ -0,0 +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.

changelog/4480.fixed.md

@@ -1 +0,0 @@
-- `TurnAnalyzerUserTurnStopStrategy` and `SpeechTimeoutUserTurnStopStrategy` no longer always wait for a transcript before ending the user turn. The wait is still correct for cascaded pipelines, where the downstream LLM can't respond until the user's words are recorded in its context — but it's pure latency in pipelines using local turn detection to drive a realtime service like Gemini Live. The strategies now infer from the presence of an `STTMetadataFrame` whether a transcript is needed, and accept a new `require_transcript: bool | None` parameter to override the heuristic explicitly.

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

@@ -70,10 +70,26 @@ 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
+    # 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),
+    # - fires `on_user_turn_stopped` on the audible end of the turn with
+    #   empty `message.content` (the transcript hasn't arrived yet), and
+    # - defers the context flush until the (late) transcript arrives, then
+    #   emits `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(
             vad_analyzer=SileroVADAnalyzer(),
+            wait_for_transcript_to_end_user_turn=False,
         ),
     )
 
@@ -107,8 +123,19 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         logger.info(f"Client disconnected")
         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
+    # its `message.content` is empty. Logged here to make the timing of the
+    # audible-stop signal visible alongside the later transcript 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__})")
+
+    # `on_user_turn_message_finalized` fires when the user transcript has
+    # been written to context — later than `on_user_turn_stopped` in this
+    # mode, since transcripts arrive after the audible turn end.
+    @user_aggregator.event_handler("on_user_turn_message_finalized")
+    async def on_user_turn_message_finalized(aggregator, strategy, message: UserTurnStoppedMessage):
         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

@@ -80,6 +80,7 @@ from pipecat.processors.aggregators.llm_context_summarizer import (
     SummaryAppliedEvent,
 )
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.services.stt_latency import DEFAULT_TTFS_P99
 from pipecat.turns.user_idle_controller import UserIdleController
 from pipecat.turns.user_mute import BaseUserMuteStrategy
 from pipecat.turns.user_start import BaseUserTurnStartStrategy, UserTurnStartedParams
@@ -127,6 +128,30 @@ class LLMUserAggregatorParams:
             has been idle (not speaking) for this duration. Set to 0 to disable
             idle detection.
         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 like Gemini Live, where the LLM consumes user audio
+            directly and doesn't need the transcript in context to respond
+            — so waiting for the transcript before ending the turn is pure
+            latency. When False, 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 strategy that
+              supports it (so the turn ends without waiting for a
+              transcript), and
+            - defers the context flush and public ``on_user_turn_stopped``
+              event until the (late) transcript arrives or a backstop timer
+              fires (so the user's words still land in the LLM context).
+
+            Adjustments to user-provided strategies are logged. The
+            precondition for the deferred-context path: every user turn
+            produces one transcript in conversational order — turn N's
+            transcript arrives before any of turn N's assistant content, and
+            before turn N+1 starts. Violations are logged and
+            best-effort-recovered by force-flushing the pending turn before
+            starting the next one.
         filter_incomplete_user_turns: [DEPRECATED] Use
             ``user_turn_strategies=FilterIncompleteUserTurnStrategies()``
             instead. When enabled, the LLM outputs a turn-completion
@@ -157,6 +182,7 @@ class LLMUserAggregatorParams:
     user_turn_stop_timeout: float = 5.0
     user_idle_timeout: float = 0
     vad_analyzer: VADAnalyzer | None = None
+    wait_for_transcript_to_end_user_turn: bool = True
     filter_incomplete_user_turns: bool = False
     user_turn_completion_config: UserTurnCompletionConfig | None = None
 
@@ -527,7 +553,18 @@ class LLMUserAggregator(LLMContextAggregator):
     Event handlers available:
 
     - on_user_turn_started: Called when the user turn starts
-    - on_user_turn_stopped: Called when the user turn ends
+    - on_user_turn_stopped: Called when the user turn ends. With
+      ``wait_for_transcript_to_end_user_turn=True`` (default), this fires
+      with the populated ``UserTurnStoppedMessage.content``. With
+      ``wait_for_transcript_to_end_user_turn=False`` (realtime + local
+      turn detection), this fires on the audible end of the turn — before
+      the transcript has arrived — so ``content`` is empty; subscribe to
+      ``on_user_turn_message_finalized`` to get the transcript.
+    - on_user_turn_message_finalized: Called once the user message has
+      been finalized into the context. Fires in both modes — in the
+      default mode it coincides with ``on_user_turn_stopped``; in the
+      delayed-transcript path it fires later, when the transcript
+      arrives. The ``UserTurnStoppedMessage.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
@@ -543,6 +580,10 @@ class LLMUserAggregator(LLMContextAggregator):
         async def on_user_turn_stopped(aggregator, strategy: BaseUserTurnStopStrategy, message: UserTurnStoppedMessage):
             ...
 
+        @aggregator.event_handler("on_user_turn_message_finalized")
+        async def on_user_turn_message_finalized(aggregator, strategy: BaseUserTurnStopStrategy, message: UserTurnStoppedMessage):
+            ...
+
         @aggregator.event_handler("on_user_turn_stop_timeout")
         async def on_user_turn_stop_timeout(aggregator):
             ...
@@ -586,12 +627,14 @@ class LLMUserAggregator(LLMContextAggregator):
 
         self._register_event_handler("on_user_turn_started")
         self._register_event_handler("on_user_turn_stopped")
+        self._register_event_handler("on_user_turn_message_finalized")
         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_mute_started")
         self._register_event_handler("on_user_mute_stopped")
 
+        user_provided_strategies = self._params.user_turn_strategies is not None
         user_turn_strategies = self._params.user_turn_strategies or UserTurnStrategies()
 
         # Deprecated path: translate filter_incomplete_user_turns into
@@ -605,6 +648,16 @@ class LLMUserAggregator(LLMContextAggregator):
             )
             self._params.user_turn_strategies = user_turn_strategies
 
+        # When the user opts out of waiting for transcripts to end the user
+        # turn, mutate the strategies to match — drop the transcription start
+        # strategy, flip `wait_for_transcript=False` on the stop strategies
+        # that support it. Loud log if the user passed their own strategies
+        # (we're overwriting parts of their config); quiet log otherwise.
+        if not self._params.wait_for_transcript_to_end_user_turn:
+            self._apply_no_transcript_wait_bundle(
+                user_turn_strategies, user_provided_strategies=user_provided_strategies
+            )
+
         self._user_is_muted = False
         self._user_turn_start_timestamp = ""
         # Full transcript across the user turn. Each
@@ -616,6 +669,15 @@ class LLMUserAggregator(LLMContextAggregator):
         # inferences fire before finalization.
         self._full_user_turn_aggregation: str | None = None
 
+        # State for the deferred-finalization path that activates when
+        # `wait_for_transcript_to_end_user_turn=False`. The strategy fires
+        # turn-stopped on the audible end of the user's turn; we defer the
+        # context flush and `on_user_turn_stopped` event until the transcript
+        # arrives (or a backstop timer fires).
+        self._pending_stop_strategy: BaseUserTurnStopStrategy | None = None
+        self._pending_inference_trigger: bool = False
+        self._delayed_transcript_backstop_task: asyncio.Task | None = None
+
         self._user_turn_controller = UserTurnController(
             user_turn_strategies=user_turn_strategies,
             user_turn_stop_timeout=self._params.user_turn_stop_timeout,
@@ -658,6 +720,75 @@ class LLMUserAggregator(LLMContextAggregator):
             self._vad_controller.add_event_handler("on_push_frame", self._on_push_frame)
             self._vad_controller.add_event_handler("on_broadcast_frame", self._on_broadcast_frame)
 
+    @property
+    def _expect_delayed_transcripts(self) -> bool:
+        """Internal alias for the deferred-finalization mode.
+
+        Active whenever ``wait_for_transcript_to_end_user_turn`` is False,
+        which is also when the strategy-mutation bundle is applied. The two
+        always travel together — there's no configuration where one is
+        active without the other.
+        """
+        return not self._params.wait_for_transcript_to_end_user_turn
+
+    def _apply_no_transcript_wait_bundle(
+        self,
+        user_turn_strategies: UserTurnStrategies,
+        *,
+        user_provided_strategies: bool,
+    ):
+        """Adjust strategies to match ``wait_for_transcript_to_end_user_turn=False``.
+
+        Drops ``TranscriptionUserTurnStartStrategy`` from start strategies
+        (so late-arriving realtime transcripts don't trigger new turns) and
+        sets ``wait_for_transcript=False`` on any stop strategy that supports
+        it (so the turn ends without waiting for a transcript).
+
+        Logs loudly when adjusting user-provided strategies — we're
+        mutating objects the caller passed in. Logs quietly when only
+        synthesized defaults are in play.
+        """
+        # Local import to avoid a top-level cycle with `turns.user_start`.
+        from pipecat.turns.user_start import TranscriptionUserTurnStartStrategy
+
+        adjustments: list[str] = []
+
+        if user_turn_strategies.start:
+            filtered = [
+                s
+                for s in user_turn_strategies.start
+                if not isinstance(s, TranscriptionUserTurnStartStrategy)
+            ]
+            dropped = len(user_turn_strategies.start) - len(filtered)
+            if dropped:
+                user_turn_strategies.start = filtered
+                adjustments.append(
+                    f"dropped {dropped} TranscriptionUserTurnStartStrategy from start strategies"
+                )
+
+        flipped = 0
+        for s in user_turn_strategies.stop or []:
+            if hasattr(s, "_wait_for_transcript") and s._wait_for_transcript:
+                s._wait_for_transcript = False
+                flipped += 1
+        if flipped:
+            adjustments.append(
+                f"set wait_for_transcript=False on {flipped} stop "
+                f"strateg{'y' if flipped == 1 else 'ies'}"
+            )
+
+        if not adjustments:
+            return
+
+        message = (
+            f"{self}: wait_for_transcript_to_end_user_turn=False adjusted "
+            f"user turn strategies: {'; '.join(adjustments)}."
+        )
+        if user_provided_strategies:
+            logger.warning(message)
+        else:
+            logger.info(message)
+
     async def cleanup(self):
         """Clean up processor resources."""
         await super().cleanup()
@@ -747,13 +878,26 @@ 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)
+        await self._finalize_on_session_end()
         await self._cleanup()
 
     async def _cancel(self, frame: CancelFrame):
-        await self._maybe_emit_user_turn_stopped(on_session_end=True)
+        await self._finalize_on_session_end()
         await self._cleanup()
 
+    async def _finalize_on_session_end(self):
+        """Flush any pending turn-stop on session end.
+
+        If a deferred turn finalization is pending (because
+        ``expect_delayed_transcripts`` is True and the transcript hadn't
+        arrived yet), replay it first so the user message is captured before
+        the session shuts down.
+        """
+        if self._pending_stop_strategy is not None or self._pending_inference_trigger:
+            await self._replay_deferred_turn_finalization(on_session_end=True)
+        else:
+            await self._maybe_emit_user_turn_stopped(on_session_end=True)
+
     async def _cleanup(self):
         if self._vad_controller:
             await self._vad_controller.cleanup()
@@ -884,6 +1028,17 @@ class LLMUserAggregator(LLMContextAggregator):
     ):
         logger.debug(f"{self}: User started speaking (strategy: {strategy})")
 
+        # Precondition guard for `expect_delayed_transcripts`: if the previous
+        # turn's finalization is still pending, the precondition (turn N's
+        # transcript arrives before turn N+1 starts) has been violated.
+        # Force-flush 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()
+
         self._user_turn_start_timestamp = time_now_iso8601()
         self._full_user_turn_aggregation = None
 
@@ -904,6 +1059,12 @@ 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
+            # transcript arrives (or the backstop fires).
+            self._pending_inference_trigger = True
+            return
+
         # Push aggregation now: this writes the user message segment to
         # the context and emits LLMContextFrame, which kicks LLM
         # inference. Concatenate the segment into
@@ -929,19 +1090,106 @@ class LLMUserAggregator(LLMContextAggregator):
     ):
         logger.debug(f"{self}: User stopped speaking (strategy: {strategy})")
 
+        # Audible-stop side effects fire on the strategy event regardless of
+        # whether the transcript-ready 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
+            # turn. `content` is empty because the transcript hasn't arrived
+            # yet — consumers that want the transcript subscribe to
+            # `on_user_turn_message_finalized` instead, which fires later when
+            # the transcript lands (or the backstop timer fires).
+            audible_stop_message = UserTurnStoppedMessage(
+                content="", timestamp=self._user_turn_start_timestamp
+            )
+            await self._call_event_handler("on_user_turn_stopped", strategy, audible_stop_message)
+
+            self._pending_stop_strategy = strategy
+            self._delayed_transcript_backstop_task = self.create_task(
+                self._delayed_transcript_backstop_handler(DEFAULT_TTFS_P99),
+                f"{self}::delayed_transcript_backstop",
+            )
+            return
+
         await self._maybe_emit_user_turn_stopped(strategy)
 
+    async def _delayed_transcript_backstop_handler(self, timeout: float):
+        """Backstop timer for `expect_delayed_transcripts`.
+
+        Fires after ``timeout`` seconds and replays the deferred turn
+        finalization with whatever transcript has been captured by then
+        (possibly nothing).
+        """
+        try:
+            await asyncio.sleep(timeout)
+        except asyncio.CancelledError:
+            return
+        finally:
+            self._delayed_transcript_backstop_task = None
+
+        await self._replay_deferred_turn_finalization()
+
+    async def _replay_deferred_turn_finalization(self, *, on_session_end: bool = False):
+        """Replay the deferred halves of inference-triggered and turn-stopped.
+
+        Called from the backstop timer, the precondition guard in
+        `_on_user_turn_started`, and the session-end paths. Clears all
+        pending state before delegating to the normal finalization methods
+        so they run the unmodified code path.
+        """
+        if self._delayed_transcript_backstop_task:
+            await self.cancel_task(self._delayed_transcript_backstop_task)
+            self._delayed_transcript_backstop_task = None
+
+        pending_strategy = self._pending_stop_strategy
+        had_pending_inference = self._pending_inference_trigger
+        self._pending_stop_strategy = None
+        self._pending_inference_trigger = False
+
+        if had_pending_inference:
+            segment = await self.push_aggregation()
+            if segment:
+                if self._full_user_turn_aggregation:
+                    self._full_user_turn_aggregation = (
+                        f"{self._full_user_turn_aggregation} {segment}".strip()
+                    )
+                else:
+                    self._full_user_turn_aggregation = segment
+            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;
+            # only fire `on_user_turn_message_finalized` here.
+            await self._maybe_emit_user_turn_stopped(
+                pending_strategy,
+                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.reset()
 
+    async def _discard_deferred_turn_finalization(self):
+        """Drop any pending deferred turn finalization without replaying.
+
+        Called from reset paths (interruption, explicit reset). "Reset" means
+        "throw it away" — we don't want to flush a partial transcript that
+        was about to be invalidated anyway.
+        """
+        if self._delayed_transcript_backstop_task:
+            await self.cancel_task(self._delayed_transcript_backstop_task)
+            self._delayed_transcript_backstop_task = None
+        self._pending_stop_strategy = None
+        self._pending_inference_trigger = False
+
     async def _on_user_turn_stop_timeout(self, controller):
         await self._call_event_handler("on_user_turn_stop_timeout")
 
@@ -952,19 +1200,28 @@ class LLMUserAggregator(LLMContextAggregator):
         self,
         strategy: BaseUserTurnStopStrategy | None = None,
         on_session_end: bool = False,
+        *,
+        emit_stopped_event: bool = True,
     ):
-        """Maybe emit user turn stopped event.
+        """Finalize the user message and emit the corresponding events.
 
         Earlier inference triggers in the same turn have already pushed
         their segments to the context and accumulated them into
         ``self._full_user_turn_aggregation``. Any aggregation that
         arrived after the last inference trigger is flushed here so
-        end-of-turn content is never lost from the public event.
+        end-of-turn content is never lost from the public events.
+
+        Always emits ``on_user_turn_message_finalized``. Also emits
+        ``on_user_turn_stopped`` when ``emit_stopped_event=True`` (the
+        default path); the delayed-transcript path passes False because it
+        already fired ``on_user_turn_stopped`` on the audible stop.
 
         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
@@ -979,7 +1236,9 @@ class LLMUserAggregator(LLMContextAggregator):
             message = UserTurnStoppedMessage(
                 content=content, timestamp=self._user_turn_start_timestamp
             )
-            await self._call_event_handler("on_user_turn_stopped", strategy, message)
+            if emit_stopped_event:
+                await self._call_event_handler("on_user_turn_stopped", strategy, message)
+            await self._call_event_handler("on_user_turn_message_finalized", strategy, message)
             self._user_turn_start_timestamp = ""
 
 

src/pipecat/turns/user_stop/speech_timeout_user_turn_stop_strategy.py

@@ -44,23 +44,20 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy):
     is defined relative to VAD stop, and STT has already emitted a
     transcript — so the stt wait is marked done immediately.
 
-    Whether a transcript is required to end the turn is controlled by
-    ``require_transcript``:
-
-    - ``None`` (default): infer from the pipeline. If an ``STTMetadataFrame``
-      has been seen, a transcript is required; otherwise it is not. This is a
-      heuristic — the underlying question is whether the downstream LLM needs
-      the user's words recorded in context before responding, which we proxy
-      by "is there an STT service in the pipeline?". Works for typical
-      cascaded and speech-to-speech setups.
-    - ``True`` / ``False``: explicit override when the heuristic doesn't fit.
+    Set ``wait_for_transcript=False`` for pipelines where the downstream LLM
+    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
+    ``expect_delayed_transcripts=True`` on ``LLMUserAggregatorParams`` so the
+    aggregator still captures the transcript when it arrives.
     """
 
     def __init__(
         self,
         *,
         user_speech_timeout: float = 0.6,
-        require_transcript: bool | None = None,
+        wait_for_transcript: bool = True,
         **kwargs,
     ):
         """Initialize the speech timeout-based user turn stop strategy.
@@ -68,18 +65,15 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy):
         Args:
             user_speech_timeout: Time to wait for the user to potentially
                 say more after they pause speaking. Defaults to 0.6 seconds.
-            require_transcript: Whether to wait for a transcript before ending
-                the user turn. ``None`` (default) infers from the presence of
-                an ``STTMetadataFrame``. ``True``/``False`` overrides the
-                heuristic.
+            wait_for_transcript: Whether to wait for a transcript before
+                ending the user turn. Defaults to True. Set to False when
+                using local turn detection with a realtime service that
+                doesn't need user transcripts in context to respond.
             **kwargs: Additional keyword arguments.
         """
         super().__init__(**kwargs)
         self._user_speech_timeout = user_speech_timeout
-        self._require_transcript = require_transcript
-        # Set when an STTMetadataFrame is received. Used as the heuristic
-        # signal when require_transcript is None.
-        self._has_stt: bool = False
+        self._wait_for_transcript = wait_for_transcript
         self._stt_timeout: float = 0.0  # STT P99 latency from STTMetadataFrame
         self._stop_secs: float = 0.0  # VAD stop_secs from VADUserStoppedSpeakingFrame
         self._stop_secs_warned: bool = False
@@ -94,13 +88,6 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy):
         self._user_speech_wait_done: bool = False
         self._stt_wait_done: bool = False
 
-    @property
-    def _transcript_required(self) -> bool:
-        """Whether the current pipeline requires a transcript to end the turn."""
-        if self._require_transcript is not None:
-            return self._require_transcript
-        return self._has_stt
-
     async def reset(self):
         """Reset the strategy to its initial state."""
         await super().reset()
@@ -138,7 +125,6 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy):
             Always returns CONTINUE so subsequent stop strategies are evaluated.
         """
         if isinstance(frame, STTMetadataFrame):
-            self._has_stt = True
             self._stt_timeout = frame.ttfs_p99_latency
             self._stop_secs_warned = False
         elif isinstance(frame, VADUserStartedSpeakingFrame):
@@ -191,12 +177,12 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy):
         # fallback-mode run of the same timer is superseded here.
         await self._restart_user_speech_timer()
 
-        # stt_timeout is a safety net. Short-circuit it if no transcript is
-        # required (no STT to wait for), the transcript is already finalized,
-        # or if the VAD stop_secs already covered it.
+        # stt_timeout is a safety net. Short-circuit it if we're not waiting
+        # for a transcript, if the transcript is already finalized, or if the
+        # VAD stop_secs already covered it.
         self._stt_wait_done = False
         effective_stt_wait = max(0.0, self._stt_timeout - self._stop_secs)
-        if not self._transcript_required or self._transcript_finalized or effective_stt_wait <= 0:
+        if not self._wait_for_transcript or self._transcript_finalized or effective_stt_wait <= 0:
             self._stt_wait_done = True
         else:
             self._stt_timeout_task = self.task_manager.create_task(
@@ -287,11 +273,11 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy):
         Both timers must be done (stt is marked done immediately on the
         fallback path and when finalization short-circuits the safety net),
         the user must not be currently speaking, and at least one transcript
-        must have been received (skipped when a transcript isn't required).
+        must have been received (skipped when ``wait_for_transcript`` is False).
         """
         if self._vad_user_speaking:
             return
-        if self._transcript_required and not self._text:
+        if self._wait_for_transcript and not self._text:
             return
 
         if self._user_speech_wait_done and self._stt_wait_done:

src/pipecat/turns/user_stop/turn_analyzer_user_turn_stop_strategy.py

@@ -43,41 +43,36 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy):
     received. Otherwise, an STT timeout (adjusted by VAD stop_secs) is used
     as a fallback.
 
-    Whether a transcript is required to end the turn is controlled by
-    ``require_transcript``:
-
-    - ``None`` (default): infer from the pipeline. If an ``STTMetadataFrame``
-      has been seen, a transcript is required; otherwise it is not. This is a
-      heuristic — the underlying question is whether the downstream LLM needs
-      the user's words recorded in context before responding, which we proxy
-      by "is there an STT service in the pipeline?". Works for typical
-      cascaded and speech-to-speech setups.
-    - ``True`` / ``False``: explicit override when the heuristic doesn't fit.
+    Set ``wait_for_transcript=False`` for pipelines where the downstream LLM
+    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
+    transcript. Pair with ``expect_delayed_transcripts=True`` on
+    ``LLMUserAggregatorParams`` so the aggregator still captures the
+    transcript when it arrives.
     """
 
     def __init__(
         self,
         *,
         turn_analyzer: BaseTurnAnalyzer,
-        require_transcript: bool | None = None,
+        wait_for_transcript: bool = True,
         **kwargs,
     ):
         """Initialize the user turn stop strategy.
 
         Args:
             turn_analyzer: The turn detection analyzer instance to detect end of user turn.
-            require_transcript: Whether to wait for a transcript before ending
-                the user turn. ``None`` (default) infers from the presence of
-                an ``STTMetadataFrame``. ``True``/``False`` overrides the
-                heuristic.
+            wait_for_transcript: Whether to wait for a transcript before
+                ending the user turn. Defaults to True. Set to False when
+                using local turn detection with a realtime service that
+                doesn't need user transcripts in context to respond.
             **kwargs: Additional keyword arguments.
         """
         super().__init__(**kwargs)
         self._turn_analyzer = turn_analyzer
-        self._require_transcript = require_transcript
-        # Set when an STTMetadataFrame is received. Used as the heuristic
-        # signal when require_transcript is None.
-        self._has_stt: bool = False
+        self._wait_for_transcript = wait_for_transcript
         self._stt_timeout: float = 0.0  # STT P99 latency from STTMetadataFrame
         self._stop_secs: float = 0.0  # VAD stop_secs from VADUserStoppedSpeakingFrame
 
@@ -91,13 +86,6 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy):
         self._timeout_task: asyncio.Task | None = None
         self._timeout_expired: bool = False
 
-    @property
-    def _transcript_required(self) -> bool:
-        """Whether the current pipeline requires a transcript to end the turn."""
-        if self._require_transcript is not None:
-            return self._require_transcript
-        return self._has_stt
-
     async def reset(self):
         """Reset the strategy to its initial state."""
         await super().reset()
@@ -141,7 +129,6 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy):
         if isinstance(frame, StartFrame):
             await self._start(frame)
         elif isinstance(frame, STTMetadataFrame):
-            self._has_stt = True
             self._stt_timeout = frame.ttfs_p99_latency
             self._stop_secs_warned = False
         elif isinstance(frame, VADUserStartedSpeakingFrame):
@@ -202,7 +189,7 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy):
         # wait for transcriptions.
         self._turn_complete = state == EndOfTurnState.COMPLETE
 
-        if not self._transcript_required:
+        if not self._wait_for_transcript:
             # No transcript to wait for. Trigger now if the turn is already
             # complete; otherwise the analyzer's audio path will trigger once
             # it indicates completion.
@@ -296,13 +283,13 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy):
         """Trigger user turn stopped if conditions are met.
 
         Conditions:
-        - We have transcription text (skipped when a transcript isn't required)
+        - We have transcription text (skipped when ``wait_for_transcript`` is False)
         - Turn analyzer indicates turn is complete
         - Either the timeout has elapsed OR we have a finalized transcript
         """
         if not self._turn_complete:
             return
-        if self._transcript_required and not self._text:
+        if self._wait_for_transcript and not self._text:
             return
 
         # For finalized transcripts, trigger immediately