Size transcript-gather timer to STT-reported P99 TTFS

The aggregator's transcript-gather timer (used when
`wait_for_transcript_to_end_user_turn=False`) was hardcoded to
`DEFAULT_TTFS_P99`. Capture `STTMetadataFrame.ttfs_p99_latency` as
it flows through the user aggregator and prefer that value, just
like the stop strategies already do. Falls back to
`DEFAULT_TTFS_P99` when no STT service has reported a value.

src/pipecat/processors/aggregators/llm_response_universal.py

@@ -55,6 +55,7 @@ from pipecat.frames.frames import (
     LLMThoughtStartFrame,
     LLMThoughtTextFrame,
     StartFrame,
+    STTMetadataFrame,
     TextFrame,
     TranscriptionFrame,
     TranslationFrame,
@@ -701,10 +702,14 @@ class LLMUserAggregator(LLMContextAggregator):
         # (`_aggregator_gathers_transcripts == True`):
         # `on_user_turn_stopped` has fired with empty content, and the
         # aggregator is waiting on `_transcript_gather_task` before
-        # finalizing the user message into context.
+        # finalizing the user message into context. The gather window
+        # duration is taken from the last `STTMetadataFrame` seen
+        # (`STTMetadataFrame.ttfs_p99_latency`), falling back to
+        # `DEFAULT_TTFS_P99` if no STT service has reported one.
         self._gathering_for_strategy: BaseUserTurnStopStrategy | None = None
         self._inference_during_gather: bool = False
         self._transcript_gather_task: asyncio.Task | None = None
+        self._stt_ttfs_p99_latency: float | None = None
 
         self._user_turn_controller = UserTurnController(
             user_turn_strategies=user_turn_strategies,
@@ -861,6 +866,13 @@ class LLMUserAggregator(LLMContextAggregator):
             # Interim transcriptions and translations are consumed here
             # and not pushed downstream, same as final TranscriptionFrame.
             pass
+        elif isinstance(frame, STTMetadataFrame):
+            # Record the STT service's reported P99 TTFS so the
+            # transcript-gather timer can size itself to the real
+            # latency. Frame is also pushed downstream so other
+            # processors keep seeing it.
+            self._stt_ttfs_p99_latency = frame.ttfs_p99_latency
+            await self.push_frame(frame, direction)
         elif isinstance(frame, LLMRunFrame):
             await self._handle_llm_run(frame)
         elif isinstance(frame, LLMMessagesAppendFrame):
@@ -1151,8 +1163,13 @@ class LLMUserAggregator(LLMContextAggregator):
             await self._call_event_handler("on_user_turn_stopped", strategy, end_of_turn_message)
 
             self._gathering_for_strategy = strategy
+            gather_timeout = (
+                self._stt_ttfs_p99_latency
+                if self._stt_ttfs_p99_latency is not None
+                else DEFAULT_TTFS_P99
+            )
             self._transcript_gather_task = self.create_task(
-                self._transcript_gather_handler(DEFAULT_TTFS_P99),
+                self._transcript_gather_handler(gather_timeout),
                 f"{self}::transcript_gather",
             )
             return

tests/test_context_aggregators_universal.py

@@ -830,6 +830,64 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
         user_messages = [m for m in context.get_messages() if m.get("role") == "user"]
         self.assertEqual([m["content"] for m in user_messages], ["Hello!"])
 
+    async def test_no_wait_for_transcript_uses_stt_metadata_for_gather_timer(self):
+        """The transcript-gather timer prefers the STT-reported P99 TTFS
+        over ``DEFAULT_TTFS_P99``. With a long ``DEFAULT_TTFS_P99`` and
+        a short STT-reported value, the gather completes by the shorter
+        time — if the timer fell back to ``DEFAULT_TTFS_P99``, this test
+        would hang.
+        """
+        from unittest.mock import patch
+
+        from pipecat.frames.frames import STTMetadataFrame
+        from pipecat.processors.aggregators import llm_response_universal
+
+        with patch.object(llm_response_universal, "DEFAULT_TTFS_P99", 60.0):
+            context = LLMContext()
+            user_aggregator = LLMUserAggregator(
+                context,
+                params=LLMUserAggregatorParams(
+                    user_turn_strategies=UserTurnStrategies(
+                        stop=[
+                            SpeechTimeoutUserTurnStopStrategy(
+                                user_speech_timeout=TRANSCRIPTION_TIMEOUT
+                            )
+                        ],
+                    ),
+                    wait_for_transcript_to_end_user_turn=False,
+                ),
+            )
+
+            events: list[tuple[str, str | None]] = []
+
+            @user_aggregator.event_handler("on_user_turn_stopped")
+            async def on_stopped(aggregator, strategy, message):
+                events.append(("stopped", message.content))
+
+            @user_aggregator.event_handler("on_user_turn_message_finalized")
+            async def on_finalized(aggregator, strategy, message):
+                events.append(("finalized", message.content))
+
+            pipeline = Pipeline([user_aggregator])
+
+            frames_to_send = [
+                # STT service advertises its P99 TTFS latency.
+                STTMetadataFrame(service_name="TestSTT", ttfs_p99_latency=TRANSCRIPTION_TIMEOUT),
+                VADUserStartedSpeakingFrame(),
+                SleepFrame(),
+                VADUserStoppedSpeakingFrame(),
+                # Let the user_speech_timeout fire so the strategy
+                # fires turn-stopped.
+                SleepFrame(sleep=TRANSCRIPTION_TIMEOUT + 0.05),
+                TranscriptionFrame(text="Hello!", user_id="", timestamp="now"),
+                # Wait for the transcript-gather timer to fire (sized
+                # to the STT-reported TTFS, not DEFAULT_TTFS_P99).
+                SleepFrame(sleep=TRANSCRIPTION_TIMEOUT + 0.05),
+            ]
+            await run_test(pipeline, frames_to_send=frames_to_send)
+
+        self.assertEqual(events, [("stopped", None), ("finalized", "Hello!")])
+
     async def test_no_wait_for_transcript_no_transcripts_arrive(self):
         """When no transcripts arrive, the transcript-gather timer still
         runs — ``on_user_turn_message_finalized`` fires with empty