Feature request: safe opt-in empty-user-turn recovery via a developer message and one LLM run

Problem Statement

We would like an opt-in, documented recovery pattern (or a small reusable helper) for a completed user turn that contains no transcription in a cascade STT → LLM → TTS pipeline.

Our chosen application policy is to append a developer message describing the missing transcription and run the LLM once, letting the model choose a short, context-appropriate clarification. VAD interruption should remain immediate; recovery should use the existing turn-finalization boundary, without adding a second fixed STT wait.

This follows the maintainer's recommendation in #4343. We understand that user-idle detection and missing-transcription recovery are different concerns, and that response wording belongs to the application. This is an enhancement request to make that pattern reliable and easy to adopt, not a request to reopen #4343 or change idle semantics globally.

The motivating sequence is:

1. User speech activity interrupts the bot.
2. Bot playback stops while the user turn is active.
3. No transcription arrives; the watchdog closes the user turn with empty content.
4. Without an application recovery handler, no LLM context is emitted and no subsequent bot speech re-arms user idle. The conversation remains silent until new input arrives.

There is a trap when implementing the timeout recommendation: #4848 reports that reading the live aggregation buffer in on_user_turn_stop_timeout can misclassify an already flushed, nonempty turn and launch a duplicate response. We are citing that report, not claiming to have independently reproduced its race here.

Proposed Solution

Provide an opt-in helper or a tested cookbook example with this contract:

• Classify the complete finalized turn, including text already flushed by earlier inference triggers. For cascade mode, on_user_turn_stopped and its message.content appear to be the appropriate boundary; please confirm. Timeout alone, or an empty aggregation_string(), is not sufficient evidence. Realtime-mode content=None must not be treated as an empty cascade transcript.
• If the turn is genuinely empty and no other response/work is pending, append an application-supplied developer instruction and request one LLM run.
• Keep the instruction and run associated with that turn. A newer turn, late transcription, or a competing response must invalidate a queued recovery. Do not leave a stale developer instruction in the next normal response.
• Respect active speech, generation (including pending response start), tool execution, playback, application mute/recovery policy, and session shutdown.
• Bound retries, and document how the application handles a recovery that itself produces no speech or fails. Merely queuing an LLM run does not guarantee output.
• Keep the wording and whether recovery is enabled under application control. A full framework state-machine redesign is not required; a tested reference implementation using existing public hooks would also be useful.

Example developer instruction selected by our team:

Speech activity was detected, but this completed user turn has no recognized text. The content and intent of any utterance are unknown. Briefly respond in context: ask the user to repeat or clarify, or briefly repeat your interrupted question when appropriate. Do not invent what the user said or treat this event as an answer, agreement, or refusal. Do not save answers or advance the workflow based on this event. Do not mention internal speech-processing mechanisms.

Applications with state-changing tools should also enforce the corresponding tool restrictions for that recovery run; prompt wording alone is not a guarantee.

Alternative Solutions

• Gating interruption on an STT result adds interruption latency and does not handle the case where a transcript never arrives.
• Re-arming idle on every user stop can race a normal slow response; it also changes the meaning of user-idle detection.
• A fixed spoken apology loses conversational context.
• The current minimal timeout example starts a new inference without checking the finalized transcript or coordinating with an already pending response.

Additional Context and Verification

Executed offline on Pipecat 1.10.0, Python 3.12.13, Linux. The script below uses the real LLMUserAggregator and watchdog, with no models, provider calls, audio recordings, or application code. Its shortened watchdog and deliberately long speech-stop timeout make both empty and nonempty turns take the watchdog path; they are reproduction settings, not production recommendations.

Observed output:

empty, no handler {'timeouts': 1, 'stopped': [''], 'idle': 0, 'context_frames': 0, 'roles': []}
empty, recovery handler {'timeouts': 1, 'stopped': [''], 'idle': 0, 'context_frames': 1, 'roles': ['developer']}
nonempty, recovery handler {'timeouts': 1, 'stopped': ['A real answer'], 'idle': 0, 'context_frames': 1, 'roles': ['user']}

This verifies the recovery trigger and a nonempty control, not the proposed cancellation guarantees or live LLM behavior. No assistant audio is simulated, so zero idle events in the recovery case is expected too.

The helper/example should additionally cover duplicate callbacks, earlier text flushes, late transcription/new speech on either side of recovery dispatch, pending LLM/tool work, mute/shutdown, repeated empty turns, and failed recovery.

Related: #4343 (application-level policy), #4848 (empty-buffer misclassification), #5005 (watchdog and turn-strategy edge cases). The empty-LLM-response recovery in #5185 addresses a different stage: here the user turn never starts an LLM response.

AI-assisted write-up; the offline script was executed in the submitting user's development environment. No claim is made that the proposed helper is implemented or that these races have all been tested.

Runnable demonstration

"""Offline cascade-pipeline reproduction; no STT, LLM, TTS, or call data."""
import asyncio
from importlib.metadata import version

from pipecat.frames.frames import (
    BotStartedSpeakingFrame, BotStoppedSpeakingFrame, LLMContextFrame,
    LLMMessagesAppendFrame, TranscriptionFrame,
    VADUserStartedSpeakingFrame, VADUserStoppedSpeakingFrame,
)
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
    LLMUserAggregator, LLMUserAggregatorParams,
)
from pipecat.tests.utils import SleepFrame, run_test
from pipecat.turns.user_start import VADUserTurnStartStrategy
from pipecat.turns.user_stop import SpeechTimeoutUserTurnStopStrategy
from pipecat.turns.user_turn_strategies import UserTurnStrategies


async def check(*, has_text=False, should_recover=False):
    context = LLMContext(messages=[])
    user = LLMUserAggregator(context, params=LLMUserAggregatorParams(
        user_turn_strategies=UserTurnStrategies(
            start=[VADUserTurnStartStrategy()],
            stop=[SpeechTimeoutUserTurnStopStrategy(user_speech_timeout=10.0)],
        ),
        user_turn_stop_timeout=0.1,
        user_idle_timeout=0.05,
    ))
    observed = {"timeouts": 0, "stopped": [], "idle": 0}

    @user.event_handler("on_user_turn_stop_timeout")
    async def on_timeout(aggregator):
        observed["timeouts"] += 1

    @user.event_handler("on_user_turn_stopped")
    async def on_stopped(aggregator, strategy, message):
        observed["stopped"].append(message.content)
        if should_recover and message.content == "":
            # Demonstrates only the chosen response mechanism. A production
            # implementation also needs turn ownership and cancellation guards.
            await aggregator.queue_frame(LLMMessagesAppendFrame(messages=[{
                "role": "developer",
                "content": "Speech activity was detected but this completed turn "
                           "has no transcript. The content is unknown. Briefly "
                           "ask for clarification appropriate to the conversation. "
                           "Do not infer an answer or claim to know what was said.",
            }], run_llm=True))

    @user.event_handler("on_user_turn_idle")
    async def on_idle(aggregator):
        observed["idle"] += 1

    frames = [
        BotStartedSpeakingFrame(), SleepFrame(sleep=0.02),
        VADUserStartedSpeakingFrame(), SleepFrame(sleep=0.02),
        BotStoppedSpeakingFrame(),
    ]
    if has_text:
        frames += [TranscriptionFrame(text="A real answer", user_id="test-user",
                                      timestamp="2026-01-01T00:00:00Z"),
                   SleepFrame(sleep=0.02)]
    frames += [VADUserStoppedSpeakingFrame(), SleepFrame(sleep=0.4)]
    down, _ = await run_test(user, frames_to_send=frames)
    observed["context_frames"] = sum(isinstance(f, LLMContextFrame) for f in down)
    observed["roles"] = [m["role"] for m in context.messages]
    return observed


async def main():
    print("Pipecat", version("pipecat-ai"))
    for name, args in [
        ("empty, no handler", {}),
        ("empty, recovery handler", {"should_recover": True}),
        ("nonempty, recovery handler", {"has_text": True, "should_recover": True}),
    ]:
        print(name, await check(**args))


asyncio.run(main())