flowcat_core/agent/
mod.rs

1// SPDX-License-Identifier: Apache-2.0
2//
3//! Higher-level **pure-logic** conversation features (no network).
4//!
5//! These are frame-logic processors that decide things from the existing frame
6//! stream without any outbound network call — so they live in core (the
7//! networked pieces, MCP-as-processor + the carrier-coordinated transfer, are
8//! out: MCP → `flowcat-services`, transfer → the embedder).
9//!
10//! Module map:
11//! - [`voicemail`] — voicemail detection over VAD/transcript frames.
12//! - [`ivr`] — IVR-navigator state machine.
13//! - [`wakeword`] — wake-word gating.
14//! - [`text_filter`] — text aggregators / output text filters.
15//! - [`fcall_filter`] — function-call filters.
16
17pub mod fcall_filter;
18pub mod ivr;
19pub mod text_filter;
20pub mod voicemail;
21pub mod wakeword;
22
23/// A tiny single-processor test harness shared by the agent unit tests. Drives a
24/// processor through one capture hop without spinning a full `PipelineTask`, so
25/// each feature's `process_frame` can be asserted deterministically (no network,
26/// no provider keys).
27#[cfg(test)]
28pub(crate) mod test_harness {
29    use std::sync::atomic::AtomicI64;
30    use std::sync::Arc;
31    use std::time::Duration;
32
33    use crate::processor::frame::{Direction, Frame};
34    use crate::processor::runtime::{channel, NORMAL_CHAN_CAP};
35    use crate::processor::{Envelope, FrameProcessor, Link};
36
37    /// Build a [`Link`] whose downstream/upstream neighbour is a single capture
38    /// channel, run `proc` over `frames` in `direction`, and return every frame
39    /// it pushed onto the capture channel (both directions funnel to the same
40    /// sink so tests can assert upstream emissions too).
41    pub(crate) async fn drive(
42        mut proc: Box<dyn FrameProcessor>,
43        frames: Vec<Frame>,
44        direction: Direction,
45    ) -> Vec<Frame> {
46        // One capture channel acts as both the downstream and upstream neighbour.
47        let (cap_tx, mut cap_rx) = channel(Arc::from("capture"), NORMAL_CHAN_CAP);
48        let link = Link {
49            next: Some(cap_tx.clone()),
50            prev: Some(cap_tx),
51            name: Arc::from(proc.name().to_string()),
52            clock: crate::processor::Clock::new(),
53            observer: None,
54            enable_metrics: false,
55            enable_usage_metrics: false,
56            ttfb_start: Arc::new(AtomicI64::new(0)),
57            processing_start: Arc::new(AtomicI64::new(0)),
58        };
59
60        for frame in frames {
61            let env = Envelope::new(frame, direction);
62            proc.process_frame(env, &link)
63                .await
64                .expect("process_frame errored");
65        }
66
67        // Drain whatever the processor emitted (system frames on the unbounded
68        // half, data/control on the bounded half), with a short settle window.
69        let mut out = Vec::new();
70        loop {
71            tokio::select! {
72                biased;
73                Some(e) = cap_rx.system.recv() => out.push(e.frame),
74                Some(e) = cap_rx.normal.recv() => out.push(e.frame),
75                _ = tokio::time::sleep(Duration::from_millis(20)) => break,
76            }
77        }
78        out
79    }
80}
flowcat_core/agent/mod.rs

flowcat_core/agent/
mod.rs
