Add wait_for_transcript flag on user-turn stop strategies

SpeechTimeoutUserTurnStopStrategy and TurnAnalyzerUserTurnStopStrategy
both gate end-of-turn on a transcript arriving. That's the right default
for cascade STT/LLM/TTS pipelines, but it puts transcripts on the latency
critical path in pipelines where local turn detection is the intended
driver of end-of-turn — typically realtime LLM services consuming audio
directly. Closed PR #4480 explored this same fix in isolation.

Add wait_for_transcript: bool = True to both strategies. False makes the
strategy signal end-of-turn as soon as VAD / the turn analyzer reports
end-of-speech, independent of transcripts. The default preserves existing
behavior. LLMContextAggregatorPair will flip this in realtime mode in a
follow-up commit.

src/pipecat/turns/user_stop/speech_timeout_user_turn_stop_strategy.py

@@ -45,16 +45,35 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy):
     transcript — so the stt wait is marked done immediately.
     """
 
-    def __init__(self, *, user_speech_timeout: float = 0.6, **kwargs):
+    def __init__(
+        self,
+        *,
+        user_speech_timeout: float = 0.6,
+        wait_for_transcript: bool = True,
+        **kwargs,
+    ):
         """Initialize the speech timeout-based user turn stop strategy.
 
         Args:
             user_speech_timeout: Time to wait for the user to potentially
                 say more after they pause speaking. Defaults to 0.6 seconds.
+            wait_for_transcript: Whether to require at least one transcript
+                before triggering end-of-turn. When True (default), turn-end
+                fires only after the user-speech timer expires *and* at least
+                one transcript has been received. When False, the strategy
+                signals turn-end as soon as VAD reports end of speech and the
+                user-speech timer has elapsed — independent of transcripts.
+                Set this to False when local turn detection is the intended
+                driver of the conversation (e.g. with a realtime LLM service
+                consuming audio directly), so transcripts are off the latency
+                critical path. ``LLMContextAggregatorPair`` flips this for
+                you when ``realtime_service_mode`` is configured with
+                ``turns_await_transcripts=False``.
             **kwargs: Additional keyword arguments.
         """
         super().__init__(**kwargs)
         self._user_speech_timeout = user_speech_timeout
+        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
@@ -69,6 +88,15 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy):
         self._user_speech_wait_done: bool = False
         self._stt_wait_done: bool = False
 
+    @property
+    def wait_for_transcript(self) -> bool:
+        """Whether transcripts gate end-of-turn signalling."""
+        return self._wait_for_transcript
+
+    @wait_for_transcript.setter
+    def wait_for_transcript(self, value: bool) -> None:
+        self._wait_for_transcript = value
+
     async def reset(self):
         """Reset the strategy to its initial state."""
         await super().reset()
@@ -252,10 +280,14 @@ 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.
+        the user must not be currently speaking, and — when
+        ``wait_for_transcript`` is True — at least one transcript must
+        have been received.
         """
-        if self._vad_user_speaking or not self._text:
+        if self._vad_user_speaking:
+            return
+
+        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

@@ -44,15 +44,35 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy):
     as a fallback.
     """
 
-    def __init__(self, *, turn_analyzer: BaseTurnAnalyzer, **kwargs):
+    def __init__(
+        self,
+        *,
+        turn_analyzer: BaseTurnAnalyzer,
+        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.
+            wait_for_transcript: Whether to require a transcript before
+                triggering end-of-turn. When True (default), turn-end fires
+                only after the turn analyzer reports COMPLETE *and* either a
+                finalized transcript arrives or the STT safety-net timeout
+                elapses with text in hand. When False, the strategy signals
+                turn-end as soon as the turn analyzer reports COMPLETE —
+                independent of transcripts. Set this to False when local
+                turn detection is the intended driver of the conversation
+                (e.g. with a realtime LLM service consuming audio directly),
+                so transcripts are off the latency critical path.
+                ``LLMContextAggregatorPair`` flips this for you when
+                ``realtime_service_mode`` is configured with
+                ``turns_await_transcripts=False``.
             **kwargs: Additional keyword arguments.
         """
         super().__init__(**kwargs)
         self._turn_analyzer = turn_analyzer
+        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
 
@@ -66,6 +86,15 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy):
         self._timeout_task: asyncio.Task | None = None
         self._timeout_expired: bool = False
 
+    @property
+    def wait_for_transcript(self) -> bool:
+        """Whether transcripts gate end-of-turn signalling."""
+        return self._wait_for_transcript
+
+    @wait_for_transcript.setter
+    def wait_for_transcript(self, value: bool) -> None:
+        self._wait_for_transcript = value
+
     async def reset(self):
         """Reset the strategy to its initial state."""
         await super().reset()
@@ -256,11 +285,25 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy):
         """Trigger user turn stopped if conditions are met.
 
         Conditions:
-        - We have transcription text
         - Turn analyzer indicates turn is complete
-        - Either the timeout has elapsed OR we have a finalized transcript
+        - When ``wait_for_transcript`` is True (default): we have
+          transcription text *and* either the safety-net timeout has
+          elapsed or a finalized transcript arrived.
+        - When ``wait_for_transcript`` is False: fire as soon as the turn
+          analyzer reports COMPLETE — independent of transcripts.
         """
-        if not self._text or not self._turn_complete:
+        if not self._turn_complete:
+            return
+
+        if not self._wait_for_transcript:
+            # Turn-end is driven by the analyzer; transcripts are bookkeeping.
+            if self._timeout_task:
+                await self.task_manager.cancel_task(self._timeout_task)
+                self._timeout_task = None
+            await self.trigger_user_turn_stopped()
+            return
+
+        if not self._text:
             return
 
         # For finalized transcripts, trigger immediately