flowcat_core/processor/
frame.rs

1// SPDX-License-Identifier: Apache-2.0
2//
3//! The pipeline `Frame` taxonomy (PROCESSOR-DESIGN §1).
4//!
5//! A closed [`Frame`] enum mirroring the pipecat frame tree
6//! (`pipecat/src/pipecat/frames/frames.py`), plus a single [`Frame::Custom`]
7//! variant carrying [`Arc<dyn CustomFrame>`] as the OSS extension hatch. Core
8//! processors `match` on named variants (alloc-free, branch-predicted,
9//! exhaustiveness-checked); OSS extensions ride in `Custom` and are downcast
10//! only by the processors that care.
11//!
12//! **NOTE:** this is distinct from [`crate::types`] (today's data-shape module,
13//! renamed from `frame.rs` in step M0). `Frame` here is the *pipeline* frame
14//! that flows between [`FrameProcessor`](crate::processor::FrameProcessor)s.
15
16use std::any::Any;
17use std::sync::atomic::{AtomicU64, Ordering};
18use std::sync::Arc;
19
20use serde::{Deserialize, Serialize};
21
22use crate::processor::metrics::MetricsData;
23
24/// Direction of frame flow. Mirrors pipecat `FrameDirection`
25/// (`frame_processor.py:56`). `Downstream` = source→sink; `Upstream` = sink→source
26/// (errors, end-of-task requests, RTVI acks).
27#[derive(Debug, Clone, Copy, PartialEq, Eq)]
28pub enum Direction {
29    Downstream,
30    Upstream,
31}
32
33impl Direction {
34    /// The opposite direction (used when broadcasting both ways).
35    pub fn flip(self) -> Direction {
36        match self {
37            Direction::Downstream => Direction::Upstream,
38            Direction::Upstream => Direction::Downstream,
39        }
40    }
41}
42
43/// Frame scheduling class — mirrors pipecat's three base classes
44/// (`frames.py:95/106/118`). Drives queue priority and interruptibility:
45/// `System` jumps the queue and survives interruption; `Data` is dropped on
46/// interruption; `Control` is ordered like Data but also survives interruption
47/// when `uninterruptible()` is set (e.g. `End`, `Stop`).
48#[derive(Debug, Clone, Copy, PartialEq, Eq)]
49pub enum FrameClass {
50    System,
51    Data,
52    Control,
53}
54
55/// Process-unique monotonic frame id source (mirrors pipecat's per-object id).
56static FRAME_ID: AtomicU64 = AtomicU64::new(1);
57
58/// Allocate the next process-unique monotonic frame id.
59pub fn next_frame_id() -> u64 {
60    FRAME_ID.fetch_add(1, Ordering::Relaxed)
61}
62
63/// Per-frame metadata, carried **out of band** in the channel [`Envelope`] so the
64/// hot enum variant (`OutputAudio(Arc<AudioFrame>)`) stays a thin pointer move
65/// (PROCESSOR-DESIGN §1.1). Mirrors the pipecat `Frame` base fields
66/// (`frames.py:73-79`). Cheap to clone (the extra map is `Arc`ed).
67///
68/// [`Envelope`]: crate::processor::Envelope
69#[derive(Debug, Clone)]
70pub struct FrameMeta {
71    /// Process-unique monotonic id (see [`next_frame_id`]).
72    pub id: u64,
73    /// Human name for tracing/observers, e.g. "OutputAudio". Built lazily.
74    pub name: &'static str,
75    /// Presentation timestamp in **nanoseconds** on the pipeline clock, if set.
76    pub pts: Option<i64>,
77    /// Paired id when a frame was broadcast both directions (`frames.py:76`).
78    pub broadcast_sibling_id: Option<u64>,
79    /// Arbitrary sideband metadata (`Arc` so cloning a frame is cheap).
80    pub extra: Option<Arc<serde_json::Map<String, serde_json::Value>>>,
81    /// Transport source/destination track names (`frames.py:78-79`).
82    pub transport_source: Option<Arc<str>>,
83    pub transport_destination: Option<Arc<str>>,
84}
85
86impl FrameMeta {
87    /// A fresh meta for `frame` with a new monotonic id and the frame's static
88    /// `name`. The common path for producing a frame inside a processor.
89    pub fn new(frame: &Frame) -> Self {
90        Self {
91            id: next_frame_id(),
92            name: frame.name(),
93            pts: None,
94            broadcast_sibling_id: None,
95            extra: None,
96            transport_source: None,
97            transport_destination: None,
98        }
99    }
100}
101
102/// OSS extension point: a frame type defined outside flowcat-core, carried in
103/// [`Frame::Custom`]. Processors that understand it `downcast_ref` via
104/// [`CustomFrame::as_any`]; everyone else forwards it unchanged.
105pub trait CustomFrame: Any + Send + Sync + std::fmt::Debug {
106    /// Scheduling class (System/Data/Control) for this custom frame.
107    fn frame_class(&self) -> FrameClass;
108    /// True if this frame must survive interruption (pipecat `UninterruptibleFrame`).
109    fn uninterruptible(&self) -> bool {
110        false
111    }
112    /// A stable human name for tracing/observers.
113    fn name(&self) -> &'static str {
114        "Custom"
115    }
116    /// Downcast hook so a processor can recover the concrete type.
117    fn as_any(&self) -> &dyn Any;
118}
119
120/// Mono 16-bit LE PCM with an explicit sample rate and channel count.
121/// Mirrors today's [`crate::types::AudioChunk`] (PROCESSOR-DESIGN §1.1) but adds
122/// `num_channels`; Arc-wrapped in the [`Frame`] enum so the hot path never copies
123/// PCM. Convertible from/to `AudioChunk` (mono) via [`From`].
124#[derive(Debug, Clone, PartialEq, Eq)]
125pub struct AudioFrame {
126    pub pcm: Vec<i16>,
127    pub sample_rate: u32,
128    pub num_channels: u16,
129}
130
131impl AudioFrame {
132    /// Mono PCM at `sample_rate`.
133    pub fn mono(pcm: Vec<i16>, sample_rate: u32) -> Self {
134        Self {
135            pcm,
136            sample_rate,
137            num_channels: 1,
138        }
139    }
140
141    /// Number of samples (per channel is not tracked separately — `pcm.len()`).
142    pub fn len(&self) -> usize {
143        self.pcm.len()
144    }
145
146    /// Whether this frame carries no samples.
147    pub fn is_empty(&self) -> bool {
148        self.pcm.is_empty()
149    }
150}
151
152impl From<crate::types::AudioChunk> for AudioFrame {
153    fn from(c: crate::types::AudioChunk) -> Self {
154        AudioFrame::mono(c.pcm, c.sample_rate)
155    }
156}
157
158impl From<&crate::types::AudioChunk> for AudioFrame {
159    fn from(c: &crate::types::AudioChunk) -> Self {
160        AudioFrame::mono(c.pcm.clone(), c.sample_rate)
161    }
162}
163
164/// Lossy conversion back to the mono [`crate::types::AudioChunk`] data shape
165/// (drops `num_channels`; the live path is always mono).
166impl From<&AudioFrame> for crate::types::AudioChunk {
167    fn from(f: &AudioFrame) -> Self {
168        crate::types::AudioChunk::new(f.pcm.clone(), f.sample_rate)
169    }
170}
171
172/// Pipeline init params. Mirrors pipecat `StartFrame` (`frames.py:847`).
173#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
174pub struct StartParams {
175    pub audio_in_sample_rate: u32,
176    pub audio_out_sample_rate: u32,
177    pub enable_metrics: bool,
178    pub enable_usage_metrics: bool,
179    pub enable_tracing: bool,
180    /// Report only the first TTFB metric per turn (`frame_processor.py` TTFB gate).
181    pub report_only_initial_ttfb: bool,
182}
183
184impl Default for StartParams {
185    fn default() -> Self {
186        Self {
187            audio_in_sample_rate: 16_000,
188            audio_out_sample_rate: 24_000,
189            enable_metrics: false,
190            enable_usage_metrics: false,
191            enable_tracing: false,
192            report_only_initial_ttfb: true,
193        }
194    }
195}
196
197/// A telephony keypad entry (DTMF). Mirrors pipecat `KeypadEntry`.
198#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
199pub enum KeypadEntry {
200    Zero,
201    One,
202    Two,
203    Three,
204    Four,
205    Five,
206    Six,
207    Seven,
208    Eight,
209    Nine,
210    Star,
211    Pound,
212}
213
214/// A spoken-language tag (BCP-47-ish). Mirrors pipecat `Language` (string-backed
215/// here so the long enum doesn't bloat core; promote to a real enum if
216/// exhaustive matching is needed).
217#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
218pub struct Language(pub String);
219
220/// Which service a [`Frame::UpdateSettings`] / control frame targets. Mirrors
221/// pipecat's settings-frame routing (`frames.py:1878`).
222#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
223pub enum ServiceKind {
224    Stt,
225    Tts,
226    Llm,
227    Filter,
228    Mixer,
229    All,
230}
231
232/// VAD parameter bundle broadcast on [`Frame::SpeechControlParams`]. Mirrors
233/// pipecat `VADParams`.
234#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
235pub struct VadParams {
236    pub confidence: f32,
237    pub start_secs: f32,
238    pub stop_secs: f32,
239    pub min_volume: f32,
240}
241
242/// Turn-analyzer parameter bundle. Mirrors pipecat turn params.
243#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
244pub struct TurnParams {
245    pub stop_secs: f32,
246    pub pre_speech_ms: f32,
247    pub max_duration_secs: Option<f32>,
248}
249
250/// A function/tool call request. Mirrors pipecat `FunctionCallFromLLM`.
251#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
252pub struct FunctionCall {
253    pub function_name: String,
254    pub tool_call_id: String,
255    pub arguments: serde_json::Value,
256}
257
258/// A function/tool call result, fed back to the LLM. Mirrors pipecat
259/// `FunctionCallResultFrame` (`frames.py:719`). Uninterruptible.
260#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
261pub struct FunctionCallResult {
262    pub function_name: String,
263    pub tool_call_id: String,
264    pub result: serde_json::Value,
265}
266
267/// The universal LLM context to run. Mirrors pipecat `LLMContext`
268/// (`frames.py:502`). Opaque message list — providers interpret it.
269#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
270pub struct LlmContext {
271    pub messages: Vec<serde_json::Value>,
272    pub tools: Vec<serde_json::Value>,
273}
274
275/// The frame that flows through every processor. One closed enum mirroring the
276/// pipecat frame tree (`frames.py`), plus [`Frame::Custom`] for OSS extensions.
277///
278/// The hot audio variants box their payload in `Arc<AudioFrame>` so cloning a
279/// frame for the broadcast/observer paths never copies PCM.
280#[derive(Debug, Clone)]
281pub enum Frame {
282    // ---- System frames (priority; survive interruption) — frames.py:846+ ----
283    /// Pipeline init: carries sample rates + metric/trace toggles (`StartFrame`).
284    Start(StartParams),
285    /// Immediate stop; flush nothing (`CancelFrame`).
286    Cancel {
287        reason: Option<String>,
288    },
289    /// Error notification pushed upstream (`ErrorFrame`). `fatal` ⇒ task cancels.
290    Error {
291        message: String,
292        fatal: bool,
293        processor: Option<Arc<str>>,
294    },
295    /// Barge-in (`InterruptionFrame`). Broadcast both directions.
296    Interruption,
297    /// Raw caller audio from a transport input (`InputAudioRawFrame`).
298    InputAudio(Arc<AudioFrame>),
299    /// User-associated audio (`UserAudioRawFrame`) — carries `user_id`.
300    UserAudio {
301        audio: Arc<AudioFrame>,
302        user_id: Arc<str>,
303    },
304    /// Raw text input from a transport (`InputTextRawFrame`) — text-chat path.
305    InputText(String),
306    /// Inbound DTMF keypress (`InputDTMFFrame`).
307    InputDtmf(KeypadEntry),
308    /// VAD/turn lifecycle (`frames.py:971-1104`). One variant per pipecat frame.
309    UserStartedSpeaking,
310    UserStoppedSpeaking,
311    UserSpeaking,
312    BotStartedSpeaking,
313    BotStoppedSpeaking,
314    BotSpeaking,
315    /// Definitive VAD edges with the deciding secs (`VAD*SpeakingFrame`).
316    VadUserStartedSpeaking {
317        start_secs: f32,
318    },
319    VadUserStoppedSpeaking {
320        stop_secs: f32,
321    },
322    /// Mute/unmute the STT service (`STTMuteFrame`).
323    SttMute(bool),
324    /// Performance metrics (`MetricsFrame`) — TTFB/processing/usage/turn.
325    Metrics(Vec<MetricsData>),
326    /// Transport-level message in/out urgent (`Input/OutputTransportMessage*`).
327    TransportMessage {
328        payload: serde_json::Value,
329        urgent: bool,
330    },
331    /// SFU/transport lifecycle (`ClientConnected`).
332    ClientConnected,
333    /// SFU/transport lifecycle (`BotConnected`).
334    BotConnected,
335    /// Function-call signalling (`FunctionCallsStarted`).
336    FunctionCallsStarted(Vec<FunctionCall>),
337    /// Function-call signalling (`FunctionCallInProgress`).
338    FunctionCallInProgress {
339        call: FunctionCall,
340        cancel_on_interruption: bool,
341    },
342    /// Function-call signalling (`FunctionCallCancel`).
343    FunctionCallCancel {
344        function_name: String,
345        tool_call_id: String,
346    },
347
348    // ---- Data frames (ordered; dropped on interruption) — frames.py:190+ ----
349    /// Output audio to a transport (`OutputAudioRawFrame`).
350    OutputAudio(Arc<AudioFrame>),
351    /// TTS-generated audio, tagged with its context id (`TTSAudioRawFrame`).
352    TtsAudio {
353        audio: Arc<AudioFrame>,
354        context_id: Option<Arc<str>>,
355    },
356    /// Generic text (`TextFrame`) — flows LLM→aggregator→TTS.
357    Text(String),
358    /// LLM-generated text chunk (`LLMTextFrame`).
359    LlmText(String),
360    /// Final transcription (`TranscriptionFrame`).
361    Transcription {
362        text: String,
363        user_id: Arc<str>,
364        language: Option<Language>,
365        final_: bool,
366    },
367    /// Interim/partial transcription (`InterimTranscriptionFrame`).
368    InterimTranscription {
369        text: String,
370        user_id: Arc<str>,
371        language: Option<Language>,
372    },
373    /// Text the TTS should speak (`TTSSpeakFrame`).
374    TtsSpeak {
375        text: String,
376        append_to_context: Option<bool>,
377    },
378    /// Word/segment text emitted by TTS with its context (`TTSTextFrame`).
379    TtsText {
380        text: String,
381        context_id: Option<Arc<str>>,
382    },
383    /// Function-call result, fed back to the LLM (`FunctionCallResultFrame`).
384    /// Uninterruptible — once produced, context must always be updated.
385    FunctionCallResult(FunctionCallResult),
386    /// Trigger an LLM run over the current context (`LLMRunFrame`).
387    LlmRun,
388    /// The universal LLM context to run (`LLMContextFrame`).
389    LlmContext(Arc<LlmContext>),
390    /// Outbound DTMF (`OutputDTMFFrame`).
391    OutputDtmf(Vec<KeypadEntry>),
392
393    // ---- Control frames (ordered; End/Stop survive interruption) — :1580+ ----
394    /// Graceful shutdown after flush (`EndFrame`). Uninterruptible.
395    End {
396        reason: Option<String>,
397    },
398    /// Stop but keep processors connected (`StopFrame`). Uninterruptible.
399    Stop,
400    /// LLM response framing (`LLMFullResponseStart`).
401    LlmResponseStart,
402    /// LLM response framing (`LLMFullResponseEnd`).
403    LlmResponseEnd,
404    /// TTS response framing (`TTSStartedFrame`).
405    TtsStarted {
406        context_id: Option<Arc<str>>,
407    },
408    /// TTS response framing (`TTSStoppedFrame`).
409    TtsStopped {
410        context_id: Option<Arc<str>>,
411    },
412    /// Update a service's settings live (`ServiceUpdateSettingsFrame`).
413    /// Uninterruptible. `target` = STT/TTS/LLM/Filter/Mixer/All.
414    UpdateSettings {
415        target: ServiceKind,
416        settings: serde_json::Value,
417    },
418    /// Speech-control params broadcast (`SpeechControlParamsFrame`) + VAD param
419    /// updates (`VADParamsUpdateFrame`).
420    SpeechControlParams {
421        vad: Option<VadParams>,
422        turn: Option<TurnParams>,
423    },
424    /// Liveness probe (`HeartbeatFrame`).
425    Heartbeat {
426        timestamp_ns: i64,
427    },
428    /// Output transport ready (`OutputTransportReadyFrame`).
429    OutputTransportReady,
430    /// Enable/disable an input-leg audio filter at runtime (`FilterEnableFrame`,
431    /// `frames.py:1971`). Promoted out of `Custom` (PROCESSOR-DESIGN §1.2) so the
432    /// [`AudioFilterProcessor`](crate::audio::filter::AudioFilterProcessor) can be
433    /// toggled hot. Control class (ordered, interruptible).
434    FilterEnable(bool),
435    /// Enable/disable an output-leg audio mixer at runtime (`MixerEnableFrame`,
436    /// `frames.py:2000`). Promoted out of `Custom`. Toggles the
437    /// [`MixerProcessor`](crate::audio::mixer::MixerProcessor). Control class.
438    MixerEnable(bool),
439
440    // ---- OSS extension point ----
441    Custom(Arc<dyn CustomFrame>),
442}
443
444impl Frame {
445    /// Scheduling class — drives queue priority + interruptibility
446    /// (PROCESSOR-DESIGN §2.3). `Custom` delegates to
447    /// [`CustomFrame::frame_class`].
448    pub fn class(&self) -> FrameClass {
449        use Frame::*;
450        match self {
451            // System
452            Start(_)
453            | Cancel { .. }
454            | Error { .. }
455            | Interruption
456            | InputAudio(_)
457            | UserAudio { .. }
458            | InputText(_)
459            | InputDtmf(_)
460            | UserStartedSpeaking
461            | UserStoppedSpeaking
462            | UserSpeaking
463            | BotStartedSpeaking
464            | BotStoppedSpeaking
465            | BotSpeaking
466            | VadUserStartedSpeaking { .. }
467            | VadUserStoppedSpeaking { .. }
468            | SttMute(_)
469            | Metrics(_)
470            | TransportMessage { .. }
471            | ClientConnected
472            | BotConnected
473            | FunctionCallsStarted(_)
474            | FunctionCallInProgress { .. }
475            | FunctionCallCancel { .. } => FrameClass::System,
476
477            // Control
478            End { .. }
479            | Stop
480            | LlmResponseStart
481            | LlmResponseEnd
482            | TtsStarted { .. }
483            | TtsStopped { .. }
484            | UpdateSettings { .. }
485            | SpeechControlParams { .. }
486            | Heartbeat { .. }
487            | OutputTransportReady
488            | FilterEnable(_)
489            | MixerEnable(_) => FrameClass::Control,
490
491            // Data
492            OutputAudio(_)
493            | TtsAudio { .. }
494            | Text(_)
495            | LlmText(_)
496            | Transcription { .. }
497            | InterimTranscription { .. }
498            | TtsSpeak { .. }
499            | TtsText { .. }
500            | FunctionCallResult(_)
501            | LlmRun
502            | LlmContext(_)
503            | OutputDtmf(_) => FrameClass::Data,
504
505            Custom(c) => c.frame_class(),
506        }
507    }
508
509    /// True ⇒ kept in the queue and not cancelled on interruption (pipecat
510    /// `UninterruptibleFrame`: End, Stop, FunctionCallResult, UpdateSettings).
511    /// `Custom` delegates to [`CustomFrame::uninterruptible`].
512    pub fn uninterruptible(&self) -> bool {
513        use Frame::*;
514        match self {
515            End { .. } | Stop | FunctionCallResult(_) | UpdateSettings { .. } => true,
516            Custom(c) => c.uninterruptible(),
517            _ => false,
518        }
519    }
520
521    /// A stable, human-readable name for tracing/observers/meta.
522    pub fn name(&self) -> &'static str {
523        use Frame::*;
524        match self {
525            Start(_) => "Start",
526            Cancel { .. } => "Cancel",
527            Error { .. } => "Error",
528            Interruption => "Interruption",
529            InputAudio(_) => "InputAudio",
530            UserAudio { .. } => "UserAudio",
531            InputText(_) => "InputText",
532            InputDtmf(_) => "InputDtmf",
533            UserStartedSpeaking => "UserStartedSpeaking",
534            UserStoppedSpeaking => "UserStoppedSpeaking",
535            UserSpeaking => "UserSpeaking",
536            BotStartedSpeaking => "BotStartedSpeaking",
537            BotStoppedSpeaking => "BotStoppedSpeaking",
538            BotSpeaking => "BotSpeaking",
539            VadUserStartedSpeaking { .. } => "VadUserStartedSpeaking",
540            VadUserStoppedSpeaking { .. } => "VadUserStoppedSpeaking",
541            SttMute(_) => "SttMute",
542            Metrics(_) => "Metrics",
543            TransportMessage { .. } => "TransportMessage",
544            ClientConnected => "ClientConnected",
545            BotConnected => "BotConnected",
546            FunctionCallsStarted(_) => "FunctionCallsStarted",
547            FunctionCallInProgress { .. } => "FunctionCallInProgress",
548            FunctionCallCancel { .. } => "FunctionCallCancel",
549            OutputAudio(_) => "OutputAudio",
550            TtsAudio { .. } => "TtsAudio",
551            Text(_) => "Text",
552            LlmText(_) => "LlmText",
553            Transcription { .. } => "Transcription",
554            InterimTranscription { .. } => "InterimTranscription",
555            TtsSpeak { .. } => "TtsSpeak",
556            TtsText { .. } => "TtsText",
557            FunctionCallResult(_) => "FunctionCallResult",
558            LlmRun => "LlmRun",
559            LlmContext(_) => "LlmContext",
560            OutputDtmf(_) => "OutputDtmf",
561            End { .. } => "End",
562            Stop => "Stop",
563            LlmResponseStart => "LlmResponseStart",
564            LlmResponseEnd => "LlmResponseEnd",
565            TtsStarted { .. } => "TtsStarted",
566            TtsStopped { .. } => "TtsStopped",
567            UpdateSettings { .. } => "UpdateSettings",
568            SpeechControlParams { .. } => "SpeechControlParams",
569            Heartbeat { .. } => "Heartbeat",
570            OutputTransportReady => "OutputTransportReady",
571            FilterEnable(_) => "FilterEnable",
572            MixerEnable(_) => "MixerEnable",
573            Custom(c) => c.name(),
574        }
575    }
576
577    /// Whether this frame is a terminal lifecycle frame (`End`/`Stop`/`Cancel`)
578    /// that ends a [`PipelineTask`](crate::pipeline::PipelineTask) when it reaches
579    /// the sink.
580    pub fn is_terminal(&self) -> bool {
581        matches!(self, Frame::End { .. } | Frame::Stop | Frame::Cancel { .. })
582    }
583
584    /// The discriminant kind, for event-hook / idle-frame matching without
585    /// carrying the payload. See [`FrameKind`].
586    pub fn kind(&self) -> FrameKind {
587        use Frame::*;
588        match self {
589            Start(_) => FrameKind::Start,
590            Cancel { .. } => FrameKind::Cancel,
591            Error { .. } => FrameKind::Error,
592            Interruption => FrameKind::Interruption,
593            InputAudio(_) => FrameKind::InputAudio,
594            UserAudio { .. } => FrameKind::UserAudio,
595            InputText(_) => FrameKind::InputText,
596            InputDtmf(_) => FrameKind::InputDtmf,
597            UserStartedSpeaking => FrameKind::UserStartedSpeaking,
598            UserStoppedSpeaking => FrameKind::UserStoppedSpeaking,
599            UserSpeaking => FrameKind::UserSpeaking,
600            BotStartedSpeaking => FrameKind::BotStartedSpeaking,
601            BotStoppedSpeaking => FrameKind::BotStoppedSpeaking,
602            BotSpeaking => FrameKind::BotSpeaking,
603            VadUserStartedSpeaking { .. } => FrameKind::VadUserStartedSpeaking,
604            VadUserStoppedSpeaking { .. } => FrameKind::VadUserStoppedSpeaking,
605            SttMute(_) => FrameKind::SttMute,
606            Metrics(_) => FrameKind::Metrics,
607            TransportMessage { .. } => FrameKind::TransportMessage,
608            ClientConnected => FrameKind::ClientConnected,
609            BotConnected => FrameKind::BotConnected,
610            FunctionCallsStarted(_) => FrameKind::FunctionCallsStarted,
611            FunctionCallInProgress { .. } => FrameKind::FunctionCallInProgress,
612            FunctionCallCancel { .. } => FrameKind::FunctionCallCancel,
613            OutputAudio(_) => FrameKind::OutputAudio,
614            TtsAudio { .. } => FrameKind::TtsAudio,
615            Text(_) => FrameKind::Text,
616            LlmText(_) => FrameKind::LlmText,
617            Transcription { .. } => FrameKind::Transcription,
618            InterimTranscription { .. } => FrameKind::InterimTranscription,
619            TtsSpeak { .. } => FrameKind::TtsSpeak,
620            TtsText { .. } => FrameKind::TtsText,
621            FunctionCallResult(_) => FrameKind::FunctionCallResult,
622            LlmRun => FrameKind::LlmRun,
623            LlmContext(_) => FrameKind::LlmContext,
624            OutputDtmf(_) => FrameKind::OutputDtmf,
625            End { .. } => FrameKind::End,
626            Stop => FrameKind::Stop,
627            LlmResponseStart => FrameKind::LlmResponseStart,
628            LlmResponseEnd => FrameKind::LlmResponseEnd,
629            TtsStarted { .. } => FrameKind::TtsStarted,
630            TtsStopped { .. } => FrameKind::TtsStopped,
631            UpdateSettings { .. } => FrameKind::UpdateSettings,
632            SpeechControlParams { .. } => FrameKind::SpeechControlParams,
633            Heartbeat { .. } => FrameKind::Heartbeat,
634            OutputTransportReady => FrameKind::OutputTransportReady,
635            FilterEnable(_) => FrameKind::FilterEnable,
636            MixerEnable(_) => FrameKind::MixerEnable,
637            Custom(_) => FrameKind::Custom,
638        }
639    }
640}
641
642/// A payload-free discriminant for [`Frame`], used by event hooks
643/// (`on_frame_reached_*`), idle-frame detection, and observers that filter by
644/// type. Mirrors using `isinstance` on a pipecat frame class.
645#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
646pub enum FrameKind {
647    Start,
648    Cancel,
649    Error,
650    Interruption,
651    InputAudio,
652    UserAudio,
653    InputText,
654    InputDtmf,
655    UserStartedSpeaking,
656    UserStoppedSpeaking,
657    UserSpeaking,
658    BotStartedSpeaking,
659    BotStoppedSpeaking,
660    BotSpeaking,
661    VadUserStartedSpeaking,
662    VadUserStoppedSpeaking,
663    SttMute,
664    Metrics,
665    TransportMessage,
666    ClientConnected,
667    BotConnected,
668    FunctionCallsStarted,
669    FunctionCallInProgress,
670    FunctionCallCancel,
671    OutputAudio,
672    TtsAudio,
673    Text,
674    LlmText,
675    Transcription,
676    InterimTranscription,
677    TtsSpeak,
678    TtsText,
679    FunctionCallResult,
680    LlmRun,
681    LlmContext,
682    OutputDtmf,
683    End,
684    Stop,
685    LlmResponseStart,
686    LlmResponseEnd,
687    TtsStarted,
688    TtsStopped,
689    UpdateSettings,
690    SpeechControlParams,
691    Heartbeat,
692    OutputTransportReady,
693    FilterEnable,
694    MixerEnable,
695    Custom,
696}
697
698#[cfg(test)]
699mod tests {
700    use super::*;
701
702    // A custom frame for the extension-hatch tests.
703    #[derive(Debug)]
704    struct MyCustom {
705        class: FrameClass,
706        keep: bool,
707        payload: u32,
708    }
709    impl CustomFrame for MyCustom {
710        fn frame_class(&self) -> FrameClass {
711            self.class
712        }
713        fn uninterruptible(&self) -> bool {
714            self.keep
715        }
716        fn name(&self) -> &'static str {
717            "MyCustom"
718        }
719        fn as_any(&self) -> &dyn Any {
720            self
721        }
722    }
723
724    /// Table test: every named variant's `class()` matches the pipecat base class.
725    #[test]
726    fn class_table_covers_every_variant() {
727        let a = Arc::new(AudioFrame::mono(vec![0], 16_000));
728        let uid: Arc<str> = Arc::from("u");
729        let cases: Vec<(Frame, FrameClass)> = vec![
730            // System
731            (Frame::Start(StartParams::default()), FrameClass::System),
732            (Frame::Cancel { reason: None }, FrameClass::System),
733            (
734                Frame::Error {
735                    message: "x".into(),
736                    fatal: false,
737                    processor: None,
738                },
739                FrameClass::System,
740            ),
741            (Frame::Interruption, FrameClass::System),
742            (Frame::InputAudio(a.clone()), FrameClass::System),
743            (
744                Frame::UserAudio {
745                    audio: a.clone(),
746                    user_id: uid.clone(),
747                },
748                FrameClass::System,
749            ),
750            (Frame::InputText("hi".into()), FrameClass::System),
751            (Frame::InputDtmf(KeypadEntry::One), FrameClass::System),
752            (Frame::UserStartedSpeaking, FrameClass::System),
753            (Frame::UserStoppedSpeaking, FrameClass::System),
754            (Frame::UserSpeaking, FrameClass::System),
755            (Frame::BotStartedSpeaking, FrameClass::System),
756            (Frame::BotStoppedSpeaking, FrameClass::System),
757            (Frame::BotSpeaking, FrameClass::System),
758            (
759                Frame::VadUserStartedSpeaking { start_secs: 0.2 },
760                FrameClass::System,
761            ),
762            (
763                Frame::VadUserStoppedSpeaking { stop_secs: 0.5 },
764                FrameClass::System,
765            ),
766            (Frame::SttMute(true), FrameClass::System),
767            (Frame::Metrics(vec![]), FrameClass::System),
768            (
769                Frame::TransportMessage {
770                    payload: serde_json::json!({}),
771                    urgent: true,
772                },
773                FrameClass::System,
774            ),
775            (Frame::ClientConnected, FrameClass::System),
776            (Frame::BotConnected, FrameClass::System),
777            (Frame::FunctionCallsStarted(vec![]), FrameClass::System),
778            (
779                Frame::FunctionCallInProgress {
780                    call: FunctionCall {
781                        function_name: "f".into(),
782                        tool_call_id: "1".into(),
783                        arguments: serde_json::json!({}),
784                    },
785                    cancel_on_interruption: false,
786                },
787                FrameClass::System,
788            ),
789            (
790                Frame::FunctionCallCancel {
791                    function_name: "f".into(),
792                    tool_call_id: "1".into(),
793                },
794                FrameClass::System,
795            ),
796            // Data
797            (Frame::OutputAudio(a.clone()), FrameClass::Data),
798            (
799                Frame::TtsAudio {
800                    audio: a.clone(),
801                    context_id: None,
802                },
803                FrameClass::Data,
804            ),
805            (Frame::Text("t".into()), FrameClass::Data),
806            (Frame::LlmText("t".into()), FrameClass::Data),
807            (
808                Frame::Transcription {
809                    text: "t".into(),
810                    user_id: uid.clone(),
811                    language: None,
812                    final_: true,
813                },
814                FrameClass::Data,
815            ),
816            (
817                Frame::InterimTranscription {
818                    text: "t".into(),
819                    user_id: uid.clone(),
820                    language: None,
821                },
822                FrameClass::Data,
823            ),
824            (
825                Frame::TtsSpeak {
826                    text: "t".into(),
827                    append_to_context: None,
828                },
829                FrameClass::Data,
830            ),
831            (
832                Frame::TtsText {
833                    text: "t".into(),
834                    context_id: None,
835                },
836                FrameClass::Data,
837            ),
838            (
839                Frame::FunctionCallResult(FunctionCallResult {
840                    function_name: "f".into(),
841                    tool_call_id: "1".into(),
842                    result: serde_json::json!({}),
843                }),
844                FrameClass::Data,
845            ),
846            (Frame::LlmRun, FrameClass::Data),
847            (
848                Frame::LlmContext(Arc::new(LlmContext::default())),
849                FrameClass::Data,
850            ),
851            (
852                Frame::OutputDtmf(vec![KeypadEntry::Pound]),
853                FrameClass::Data,
854            ),
855            // Control
856            (Frame::End { reason: None }, FrameClass::Control),
857            (Frame::Stop, FrameClass::Control),
858            (Frame::LlmResponseStart, FrameClass::Control),
859            (Frame::LlmResponseEnd, FrameClass::Control),
860            (Frame::TtsStarted { context_id: None }, FrameClass::Control),
861            (Frame::TtsStopped { context_id: None }, FrameClass::Control),
862            (
863                Frame::UpdateSettings {
864                    target: ServiceKind::Llm,
865                    settings: serde_json::json!({}),
866                },
867                FrameClass::Control,
868            ),
869            (
870                Frame::SpeechControlParams {
871                    vad: None,
872                    turn: None,
873                },
874                FrameClass::Control,
875            ),
876            (Frame::Heartbeat { timestamp_ns: 0 }, FrameClass::Control),
877            (Frame::OutputTransportReady, FrameClass::Control),
878            (Frame::FilterEnable(true), FrameClass::Control),
879            (Frame::MixerEnable(false), FrameClass::Control),
880        ];
881        for (frame, expected) in cases {
882            assert_eq!(
883                frame.class(),
884                expected,
885                "class() mismatch for {}",
886                frame.name()
887            );
888        }
889    }
890
891    #[test]
892    fn uninterruptible_table() {
893        // The four uninterruptible named frames.
894        assert!(Frame::End { reason: None }.uninterruptible());
895        assert!(Frame::Stop.uninterruptible());
896        assert!(Frame::FunctionCallResult(FunctionCallResult {
897            function_name: "f".into(),
898            tool_call_id: "1".into(),
899            result: serde_json::json!({}),
900        })
901        .uninterruptible());
902        assert!(Frame::UpdateSettings {
903            target: ServiceKind::All,
904            settings: serde_json::json!({}),
905        }
906        .uninterruptible());
907        // A sample of interruptible frames.
908        assert!(!Frame::Interruption.uninterruptible());
909        assert!(!Frame::Text("t".into()).uninterruptible());
910        assert!(!Frame::OutputAudio(Arc::new(AudioFrame::mono(vec![0], 8000))).uninterruptible());
911        assert!(!Frame::Start(StartParams::default()).uninterruptible());
912    }
913
914    #[test]
915    fn custom_frame_delegates_class_and_uninterruptible() {
916        let c = Frame::Custom(Arc::new(MyCustom {
917            class: FrameClass::Control,
918            keep: true,
919            payload: 7,
920        }));
921        assert_eq!(c.class(), FrameClass::Control);
922        assert!(c.uninterruptible());
923        assert_eq!(c.name(), "MyCustom");
924        assert_eq!(c.kind(), FrameKind::Custom);
925
926        let d = Frame::Custom(Arc::new(MyCustom {
927            class: FrameClass::Data,
928            keep: false,
929            payload: 9,
930        }));
931        assert_eq!(d.class(), FrameClass::Data);
932        assert!(!d.uninterruptible());
933
934        // Downcast recovers the concrete type.
935        if let Frame::Custom(inner) = &d {
936            let mc = inner.as_any().downcast_ref::<MyCustom>().expect("downcast");
937            assert_eq!(mc.payload, 9);
938        } else {
939            panic!("expected Custom");
940        }
941    }
942
943    #[test]
944    fn frame_ids_are_monotonic_and_unique() {
945        let a = next_frame_id();
946        let b = next_frame_id();
947        let c = next_frame_id();
948        assert!(a < b && b < c);
949    }
950
951    #[test]
952    fn audio_chunk_roundtrips_to_audio_frame() {
953        let chunk = crate::types::AudioChunk::new(vec![1, 2, 3], 16_000);
954        let frame: AudioFrame = (&chunk).into();
955        assert_eq!(frame.num_channels, 1);
956        assert_eq!(frame.pcm, vec![1, 2, 3]);
957        let back: crate::types::AudioChunk = (&frame).into();
958        assert_eq!(back, chunk);
959    }
960}
flowcat_core/processor/frame.rs

flowcat_core/processor/
frame.rs
