Warn when VAD stop_secs misconfiguration may degrade turn detection

Add warnings in SpeechTimeoutUserTurnStopStrategy and
TurnAnalyzerUserTurnStopStrategy when stop_secs differs from the
recommended default (0.2s) or when stop_secs >= STT p99 latency,
which collapses the STT wait timeout to 0s. Document the stop_secs=0.2
assumption in stt_latency.py.

tests/test_user_turn_stop_strategy.py

@@ -6,6 +6,7 @@
 
 import asyncio
 import unittest
+from unittest.mock import patch
 
 from pipecat.frames.frames import (
     InterimTranscriptionFrame,
@@ -538,6 +539,84 @@ class TestSpeechTimeoutUserTurnStopStrategy(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(stop_count, 2)
 
 
+class TestSpeechTimeoutStopSecsWarnings(unittest.IsolatedAsyncioTestCase):
+    """Tests for stop_secs misconfiguration warnings."""
+
+    async def asyncSetUp(self) -> None:
+        self.task_manager = TaskManager()
+        self.task_manager.setup(TaskManagerParams(loop=asyncio.get_running_loop()))
+
+    async def _create_strategy(self, stt_timeout=0.35):
+        strategy = SpeechTimeoutUserTurnStopStrategy(user_speech_timeout=AGGREGATION_TIMEOUT)
+        await strategy.setup(self.task_manager)
+        await strategy.process_frame(
+            STTMetadataFrame(service_name="test", ttfs_p99_latency=stt_timeout)
+        )
+        return strategy
+
+    @patch("pipecat.turns.user_stop.speech_timeout_user_turn_stop_strategy.logger")
+    async def test_warns_on_non_default_stop_secs(self, mock_logger):
+        # Use high stt_timeout so only Warning A fires (stop_secs < stt_timeout)
+        strategy = await self._create_strategy(stt_timeout=1.0)
+
+        await strategy.process_frame(VADUserStartedSpeakingFrame())
+        await strategy.process_frame(VADUserStoppedSpeakingFrame(stop_secs=0.5))
+
+        mock_logger.warning.assert_called_once()
+        self.assertIn("differs from the recommended default", mock_logger.warning.call_args[0][0])
+
+    @patch("pipecat.turns.user_stop.speech_timeout_user_turn_stop_strategy.logger")
+    async def test_warns_on_stop_secs_gte_stt_timeout(self, mock_logger):
+        strategy = await self._create_strategy(stt_timeout=0.35)
+
+        await strategy.process_frame(VADUserStartedSpeakingFrame())
+        await strategy.process_frame(VADUserStoppedSpeakingFrame(stop_secs=0.5))
+
+        # Both warnings fire: non-default stop_secs AND stop_secs >= stt_timeout
+        self.assertEqual(mock_logger.warning.call_count, 2)
+        self.assertIn("collapsed to 0s", mock_logger.warning.call_args_list[1][0][0])
+
+    @patch("pipecat.turns.user_stop.speech_timeout_user_turn_stop_strategy.logger")
+    async def test_warns_only_once(self, mock_logger):
+        # Use high stt_timeout so only Warning A fires
+        strategy = await self._create_strategy(stt_timeout=1.0)
+
+        # First VAD stop — triggers warning
+        await strategy.process_frame(VADUserStartedSpeakingFrame())
+        await strategy.process_frame(VADUserStoppedSpeakingFrame(stop_secs=0.5))
+        self.assertEqual(mock_logger.warning.call_count, 1)
+
+        # Second VAD stop — no duplicate warning
+        await strategy.process_frame(VADUserStartedSpeakingFrame())
+        await strategy.process_frame(VADUserStoppedSpeakingFrame(stop_secs=0.5))
+        self.assertEqual(mock_logger.warning.call_count, 1)
+
+    @patch("pipecat.turns.user_stop.speech_timeout_user_turn_stop_strategy.logger")
+    async def test_warning_resets_on_new_stt_metadata(self, mock_logger):
+        # Use high stt_timeout so only Warning A fires
+        strategy = await self._create_strategy(stt_timeout=1.0)
+
+        await strategy.process_frame(VADUserStartedSpeakingFrame())
+        await strategy.process_frame(VADUserStoppedSpeakingFrame(stop_secs=0.5))
+        self.assertEqual(mock_logger.warning.call_count, 1)
+
+        # New STTMetadataFrame resets the warned flag
+        await strategy.process_frame(STTMetadataFrame(service_name="test", ttfs_p99_latency=1.0))
+
+        await strategy.process_frame(VADUserStartedSpeakingFrame())
+        await strategy.process_frame(VADUserStoppedSpeakingFrame(stop_secs=0.5))
+        self.assertEqual(mock_logger.warning.call_count, 2)
+
+    @patch("pipecat.turns.user_stop.speech_timeout_user_turn_stop_strategy.logger")
+    async def test_no_warning_on_default_stop_secs(self, mock_logger):
+        strategy = await self._create_strategy()
+
+        await strategy.process_frame(VADUserStartedSpeakingFrame())
+        await strategy.process_frame(VADUserStoppedSpeakingFrame(stop_secs=0.2))
+
+        mock_logger.warning.assert_not_called()
+
+
 class TestExternalUserTurnStopStrategy(unittest.IsolatedAsyncioTestCase):
     async def test_external_strategy(self):
         strategy = ExternalUserTurnStopStrategy()