User Turn silently skipped when STT results arrives later than VAD

[anotine10]

I am debugging specific issue:

• Deepgram STT flux outputs results later than expected.
• TUrn is already stopped as VAD is not detecting any activity (and it is stopped with content='' as deepgram has not output any transcription yet)
• When Deepgram finally outputs the final transcript, The _trigger_user_turn_stop triggered with the External Strategy is silently exiting because use_turn is already false.

https://github.com/pipecat-ai/pipecat/blame/e248c4c04905fd986bec8e793a55791b0f7698bc/src/pipecat/turns/user_turn_controller.py#L282

 async def _trigger_user_turn_stop(self, strategy, params):
      # Prevent two consecutive user turn stops.
      if not self._user_turn:
          return   # ← SILENTLY EXITS
      ...
      await self._call_event_handler("on_user_turn_stopped", strategy, params)

@markbackman , is there any good reason to not remove this condition?

To force reproduction:
i have put:

deepgram_params = DeepgramFluxSTTService.InputParams(
                eager_eot_threshold=0.5,  # default None
                eot_threshold=0.8,  # default 0.7
                eot_timeout_ms=6000,  # Maximum silence duration (ms) before forcing EndOfTurn, default 5000
                min_confidence=0.5,  # default 0.3
            )

user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
            context,
            user_params=LLMUserAggregatorParams(
                user_turn_strategies=turn_strategies,
                user_mute_strategies=mute_strategies,
                vad_analyzer=vad_analyzer,
                user_idle_timeout=user_idle_timeout,
                user_turn_stop_timeout=2, << less than 6000
            ),
            assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=True),
        )

[anotine10]

Proposal as solution: Don't fire the watchdog with empty content
Behavior:

• Subclasses pipecat's UserTurnController
• Tracks whether a final transcription (TranscriptionFrame with non-empty text) has arrived during the current turn
• Interim transcriptions keep the turn alive (via parent's event reset) but don't count as "content captured"
• Overrides the watchdog loop: if timeout fires with no final transcription yet, skip up to MAX_EMPTY_WATCHDOG_FIRES=3 times (giving slow STT ~15s grace at
  user_turn_stop_timeout=5)
• After the grace budget is exhausted, the watchdog fires as usual (safety net for genuinely dead turns)
• Turn-start resets the counter and the "has final" flag

"""Patched UserTurnController that tolerates late STT transcripts.

The parent `UserTurnController` runs an inactivity watchdog that force-stops
the user turn after `user_turn_stop_timeout` seconds with `strategy=None`.
When the STT service (e.g., Deepgram Flux with `eot_timeout_ms=4000`) hasn't
delivered a final transcription yet, this races: the watchdog closes the turn
with empty content, and the late-arriving transcription is silently dropped
because `_trigger_user_turn_stop` returns early via the `if not self._user_turn`
guard.

This subclass keeps the turn open across watchdog fires that would close it
with no captured final transcription — giving slow STT services extra grace
before giving up. After `MAX_EMPTY_WATCHDOG_FIRES` consecutive empty fires,
the watchdog proceeds as a safety net for genuinely dead turns.
"""

import asyncio

from pipecat.frames.frames import (
    InterimTranscriptionFrame,
    TranscriptionFrame,
)
from pipecat.turns.user_stop import UserTurnStoppedParams
from pipecat.turns.user_turn_controller import UserTurnController
from pipecat.turns.user_turn_strategies import UserTurnStrategies
from loguru import logger


class CustomUserTurnController(UserTurnController):
    """UserTurnController that skips empty-content watchdog fires.

    See module docstring for the rationale. Only the inactivity watchdog
    loop is overridden; turn start/stop semantics are otherwise identical
    to the parent.
    """

    # Max number of consecutive watchdog timeouts allowed to pass without
    # closing the turn while waiting for a final transcription. After this,
    # the watchdog closes the turn as usual (safety net).
    MAX_EMPTY_WATCHDOG_FIRES = 3

    def __init__(
        self,
        *,
        user_turn_strategies: UserTurnStrategies,
        user_turn_stop_timeout: float = 5.0,
    ):
        super().__init__(
            user_turn_strategies=user_turn_strategies,
            user_turn_stop_timeout=user_turn_stop_timeout,
        )
        self._has_final_transcription_in_turn: bool = False
        self._empty_watchdog_fires: int = 0

    async def _handle_transcription(
        self, frame: TranscriptionFrame | InterimTranscriptionFrame
    ):
        await super()._handle_transcription(frame)
        # Only final transcriptions count as "content captured". Interim
        # frames indicate the STT is still processing — they should keep
        # the turn alive but not signal that we're ready to close it.
        if isinstance(frame, TranscriptionFrame) and frame.text and frame.text.strip():
            self._has_final_transcription_in_turn = True

    async def _trigger_user_turn_start(self, strategy, params):
        # Reset per-turn tracking when a new turn begins.
        self._has_final_transcription_in_turn = False
        self._empty_watchdog_fires = 0
        await super()._trigger_user_turn_start(strategy, params)

    async def _user_turn_stop_timeout_task_handler(self):
        """Override of the parent watchdog loop.

        The only behavioural change vs. the parent is the guarded skip
        when the turn has no captured final transcription yet. All other
        mechanics (event wait, clear, force-stop call) are preserved.
        """
        while True:
            try:
                await asyncio.wait_for(
                    self._user_turn_stop_timeout_event.wait(),
                    timeout=self._user_turn_stop_timeout,
                )
                self._user_turn_stop_timeout_event.clear()
                # Any activity resets the empty-fire counter.
                self._empty_watchdog_fires = 0
            except asyncio.TimeoutError:
                if not (self._user_turn and not self._user_speaking):
                    continue

                if (
                    not self._has_final_transcription_in_turn
                    and self._empty_watchdog_fires < self.MAX_EMPTY_WATCHDOG_FIRES
                ):
                    # STT likely still processing. Skip this fire and wait more.
                    self._empty_watchdog_fires += 1
                    logger.debug(
                        f"{self}: watchdog fire skipped "
                        f"({self._empty_watchdog_fires}/{self.MAX_EMPTY_WATCHDOG_FIRES}) "
                        f"- awaiting final transcription"
                    )
                    continue

                await self._call_event_handler("on_user_turn_stop_timeout")
                await self._trigger_user_turn_stop(
                    None,
                    UserTurnStoppedParams(enable_user_speaking_frames=True),
                )
                self._empty_watchdog_fires = 0

[markbackman]

For Deepgram Flux, the recommended configuration is to rely on Flux itself to act as turn controller:

    user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
        context,
        user_params=LLMUserAggregatorParams(
            user_turn_strategies=ExternalUserTurnStrategies(),
            vad_analyzer=SileroVADAnalyzer(),
        ),
    )

The vad_analyzer is included only to produce latency measurements.

Do you have issues when using this configuration?

[anotine10]

So, I agree that Flux strategies does not need the Vad Analyzer to work. However if i want to use the IdleController i do no need the VAD analyzer (correct me if I am wrong), and this is when VAD enters in conflict with the Deepgram FLux, no?

Also, i have the feeling that this case also occurs when i use nova-3 and a TranscriptionUserTurnStartStrategy and SpeechTimeoutUserTurnStopStrategy.

[anotine10]

There is something i do not get.

looking at the code

class ExternalUserTurnStopStrategy(BaseUserTurnStopStrategy):
   ....
    async def _task_handler(self):
        """Asynchronously monitor transcriptions and trigger user end turn when ready.

        If transcription text exists and the user is not currently speaking,
        triggers the user end turn. Handles multiple or delayed transcriptions
        gracefully.

        """
        while True:
            try:
                await asyncio.wait_for(self._event.wait(), timeout=self._timeout)
                self._event.clear()
            except asyncio.TimeoutError:
                await self._maybe_trigger_user_turn_stopped()

You do have a logic based on timeout that will "override" the flux logic, no?