TTS usage metrics are lost for TTSSpeakFrame text in TOKEN-streaming mode: InterruptionFrame discards _streamed_text before it can flush

pipecat version: 1.8.1

Problem:

When a TTSService subclass runs in TextAggregationMode.TOKEN (the default for DeepgramFluxTTSBase and any other token-streaming TTS service), start_tts_usage_metrics is a no-op per-call (tts_service.py:436-447 — if self._is_streaming_tokens: return). Usage is instead accumulated into self._streamed_text inside _push_tts_frames (tts_service.py:1183-1185) and only flushed into a real TTSUsageMetricsData/MetricsFrame when an LLMFullResponseEndFrame or EndFrame reaches process_frame (tts_service.py:788, 818-821).

_push_tts_frames is the single accumulation point for both LLM-generated TextFrames and directly-queued TTSSpeakFrames (canned lines — greetings, transfer announcements, fallback apologies, goodbyes). Only the LLM-token path is reliably followed by an LLMFullResponseEndFrame. A TTSSpeakFrame utterance has no equivalent flush of its own — its text still lands in the same _streamed_text accumulator, but nothing guarantees an LLMFullResponseEndFrame/EndFrame will arrive after it before something else clears the accumulator.

That "something else" is _handle_interruption (tts_service.py:1031-1040), which unconditionally sets self._streamed_text = "" with no flush at all — not even a discard metric. Pipecat's own turn-taking machinery fires an InterruptionFrame at the start of essentially every caller turn as a safety net, independent of whether the bot was actually still speaking. In practice, for any pipeline where a TTSSpeakFrame is queued and the next event is a caller turn (rather than a graceful EndFrame) — e.g. a transfer flow: greeting → caller asks for reception → tool call with no LLM-generated text → canned transfer-announcement TTSSpeakFrame → call ends via cancellation/bridge teardown rather than a graceful EndFrame — the TTSSpeakFrame's characters are silently dropped from usage metrics. They were spoken (the caller heard them), but no TTSUsageMetricsData is ever emitted for them.

Minimal repro:

# A pipeline with a TOKEN-streaming TTS service (e.g. DeepgramFluxTTSBase subclass, its
# documented default text_aggregation_mode).
await task.queue_frames([
    TTSSpeakFrame("Thanks for calling, I'm an AI assistant. How can I help?"),
])
# ... before any LLMFullResponseEndFrame/EndFrame reaches the TTS processor, the pipeline's
# turn-taking (or a test harness standing in for it) pushes an InterruptionFrame — e.g. the
# caller's next turn beginning, even if the greeting had already finished playing.
await task.queue_frames([
    InterruptionFrame(),
])
# Expected: a MetricsFrame carrying TTSUsageMetricsData(value=<len of the greeting>) should
# have been pushed downstream before or during the interruption.
# Actual: no such frame is ever pushed. self._streamed_text held the greeting's text and was
# cleared to "" by _handle_interruption with no flush.

Expected: every character actually sent to a TTS provider for synthesis — whether it arrived via LLM-token streaming or via a directly-queued TTSSpeakFrame — eventually produces a TTSUsageMetricsData/MetricsFrame, regardless of what frame follows it.

Actual: TTSSpeakFrame text in TOKEN-streaming mode is only counted if an LLMFullResponseEndFrame/EndFrame happens to reach process_frame before the next InterruptionFrame does — which is close to never for a pure-transfer/tool-call conversational flow. _handle_interruption discards accumulated TOKEN-mode text unconditionally, with no flush and no discard-metric of any kind.

Suggested fix (two independent options, not mutually exclusive):

1. Framework-level: give _handle_interruption a flush-before-clear, mirroring the existing LLMFullResponseEndFrame/EndFrame branch — i.e. if self._streamed_text is non-empty when an interruption arrives, call start_tts_usage_metrics(self._streamed_text) before resetting it to "", the same way tts_service.py:818-821 already does. This alone would recover the common "greeting/canned-line followed by the caller's next turn" case without needing any provider-specific change.
2. Provider-level, where the provider reports its own authoritative billed count: for services like DeepgramFluxTTSBase, where SpeechMetadata.billable_character_count already arrives on the wire per-turn (tts_base.py:389-397) and is currently only logged, push a MetricsFrame(data=[TTSUsageMetricsData(processor=..., model=..., value=billable_character_count)]) directly from that handler. This is more accurate than the fallback in option 1 (Deepgram's own count reflects what actually gets billed after any markup stripping, not just len() of the text pipecat sent) and sidesteps the TOKEN-mode flush/interruption timing question entirely, since it fires once per turn regardless of what triggered the turn or what happens afterward. SessionMetadata's session-total field (also currently only logged) could serve as an optional end-of-session reconciliation stamp to catch any single dropped SpeechMetadata message.

Related but non-duplicate: #4138 (provider attribution + ServiceSwitcher cost splitting — doesn't touch this accumulator/flush mechanism).