gemini_adk_fluent_rs/telephony/
bridge.rs

1//! Vendor-neutral call-bridge components.
2//!
3//! Every contact-center connector — Twilio Media Streams, a raw SIP/RTP leg,
4//! a platform's gRPC virtual-agent slot — reduces to the same duties: move
5//! audio frames both ways through [`voice::pump`](crate::voice::pump), land
6//! caller keypresses and call identity in session state where flow guards
7//! read them, and keep the caller's ear busy while slow work runs. This
8//! module holds those duties as small, connector-agnostic components, so a
9//! new connector composes them instead of re-inventing them.
10//!
11//! Nothing here owns a socket. Connectors own transport; these components
12//! own semantics.
13
14use std::sync::Arc;
15use std::time::Duration;
16
17use tokio::sync::{broadcast, mpsc};
18use tokio::task::JoinHandle;
19
20use gemini_adk_rs::State;
21use gemini_adk_rs::live::{LiveEvent, LiveHandle};
22
23use crate::voice::Playback;
24
25// ── Session-state vocabulary ─────────────────────────────────────────────────
26//
27// One key set for every connector, so a flow guard like
28// `Guard::eq("telephony:dtmf", "1")` works identically behind Twilio, SIP,
29// or any future transport.
30
31/// State key holding the most recent DTMF digit pressed by the caller.
32pub const KEY_DTMF: &str = "telephony:dtmf";
33/// State key holding every DTMF digit pressed so far, concatenated in order.
34pub const KEY_DTMF_HISTORY: &str = "telephony:dtmf_history";
35/// State key holding the transport's call identifier once known.
36pub const KEY_CALL_SID: &str = "telephony:call_sid";
37/// State key holding the transport's media-stream identifier once known.
38pub const KEY_STREAM_SID: &str = "telephony:stream_sid";
39/// State key holding the caller identity the transport presented
40/// (SIP `From`, a platform's ANI field, …).
41pub const KEY_CALLER: &str = "telephony:caller";
42
43/// Record one DTMF keypress into session state under the shared keys.
44///
45/// Sets [`KEY_DTMF`] to the digit and appends it to [`KEY_DTMF_HISTORY`] —
46/// the exact writes every connector must make, factored to one place.
47pub fn record_dtmf(state: &State, digit: char) {
48    let _ = state.set(KEY_DTMF, digit.to_string());
49    let _ = state.modify(KEY_DTMF_HISTORY, String::new(), |mut history| {
50        history.push(digit);
51        history
52    });
53}
54
55/// Deduplicates RFC 4733 end-of-event packets.
56///
57/// A telephone-event keypress ends with its final packet conventionally
58/// retransmitted three times, all sharing one RTP timestamp. Feed every
59/// end-marked event through [`accept`](Self::accept); only the first per
60/// timestamp comes back `true`.
61#[derive(Debug, Default)]
62pub struct DtmfDeduper {
63    last_end_timestamp: Option<u32>,
64}
65
66impl DtmfDeduper {
67    /// `true` exactly once per keypress: for the first end-marked packet
68    /// carrying a given RTP timestamp.
69    pub fn accept(&mut self, end: bool, rtp_timestamp: u32) -> bool {
70        if !end {
71            return false;
72        }
73        if self.last_end_timestamp == Some(rtp_timestamp) {
74            return false;
75        }
76        self.last_end_timestamp = Some(rtp_timestamp);
77        true
78    }
79}
80
81// ── Latency filler ───────────────────────────────────────────────────────────
82
83/// Configuration for [`spawn_latency_filler`].
84#[derive(Clone)]
85pub struct FillerConfig {
86    /// The filler clip: mono PCM16 at the connector's playback sample rate
87    /// (the `speaker_hz` given to [`voice::pump`](crate::voice::pump)) —
88    /// e.g. a pre-synthesized "one moment, let me check that".
89    pub clip: Arc<Vec<i16>>,
90    /// Silence to tolerate after the caller stops speaking before playing
91    /// the clip. Below ~1.5 s the filler fires on normal model latency and
92    /// talks over the answer's first syllables.
93    pub delay: Duration,
94    /// At most one filler per this interval, so a long tool call gets one
95    /// reassurance, not a loop of them.
96    pub min_interval: Duration,
97}
98
99impl FillerConfig {
100    /// A filler clip with the conventional pacing: 2 s of tolerated silence,
101    /// at most one filler per 10 s.
102    pub fn new(clip: Vec<i16>) -> Self {
103        Self {
104            clip: Arc::new(clip),
105            delay: Duration::from_secs(2),
106            min_interval: Duration::from_secs(10),
107        }
108    }
109
110    /// Override the tolerated-silence delay.
111    pub fn delay(mut self, delay: Duration) -> Self {
112        self.delay = delay;
113        self
114    }
115
116    /// Override the per-filler minimum interval.
117    pub fn min_interval(mut self, interval: Duration) -> Self {
118        self.min_interval = interval;
119        self
120    }
121}
122
123/// Keep the caller's ear busy while the model is slow: when the caller stops
124/// speaking ([`LiveEvent::VadEnd`]) and no model audio arrives within
125/// `config.delay`, inject the configured clip into the connector's playback
126/// channel. Model audio or an interruption disarms the timer; a
127/// [`Playback::Flush`] from a barge-in clears any queued filler exactly like
128/// any other queued audio.
129///
130/// Masking is not a substitute for reducing latency — it buys tolerance for
131/// the tail, and because it is driven by the same event stream as the
132/// telemetry lane, the silences it papers over remain visible in the
133/// latency metrics.
134///
135/// The task ends when the session's event stream closes; abort the handle to
136/// stop it sooner.
137pub fn spawn_latency_filler(
138    handle: &LiveHandle,
139    speaker: mpsc::Sender<Playback>,
140    config: FillerConfig,
141) -> JoinHandle<()> {
142    let events = handle.events();
143    tokio::spawn(filler_task(events, speaker, config))
144}
145
146/// The filler loop itself, taking the event stream directly — the seam tests
147/// drive without a session.
148pub(crate) async fn filler_task(
149    mut events: broadcast::Receiver<LiveEvent>,
150    speaker: mpsc::Sender<Playback>,
151    config: FillerConfig,
152) {
153    let mut armed_at: Option<tokio::time::Instant> = None;
154    let mut last_filler: Option<tokio::time::Instant> = None;
155    loop {
156        let deadline = armed_at.map(|at| at + config.delay);
157        tokio::select! {
158            event = events.recv() => match event {
159                Ok(LiveEvent::VadEnd) => armed_at = Some(tokio::time::Instant::now()),
160                // Model audio (or the user cutting in) means silence ended.
161                Ok(LiveEvent::Audio(_)) | Ok(LiveEvent::VadStart) | Ok(LiveEvent::Interrupted) => {
162                    armed_at = None;
163                }
164                Ok(_) => {}
165                Err(broadcast::error::RecvError::Lagged(_)) => continue,
166                Err(broadcast::error::RecvError::Closed) => break,
167            },
168            () = async {
169                match deadline {
170                    Some(deadline) => tokio::time::sleep_until(deadline).await,
171                    None => std::future::pending().await,
172                }
173            } => {
174                armed_at = None;
175                let recently = last_filler
176                    .is_some_and(|at| at.elapsed() < config.min_interval);
177                if !recently {
178                    last_filler = Some(tokio::time::Instant::now());
179                    let _ = speaker
180                        .send(Playback::Chunk(config.clip.as_ref().clone()))
181                        .await;
182                }
183            }
184        }
185    }
186}
187
188#[cfg(test)]
189mod tests {
190    use super::*;
191
192    #[test]
193    fn dtmf_dedup_accepts_one_end_per_timestamp() {
194        let mut dedup = DtmfDeduper::default();
195        assert!(!dedup.accept(false, 100), "non-end packets never emit");
196        assert!(dedup.accept(true, 100), "first end emits");
197        assert!(!dedup.accept(true, 100), "retransmitted end is dropped");
198        assert!(!dedup.accept(true, 100));
199        assert!(dedup.accept(true, 900), "next keypress emits again");
200    }
201
202    #[test]
203    fn record_dtmf_writes_the_shared_keys() {
204        let state = State::new();
205        record_dtmf(&state, '4');
206        record_dtmf(&state, '#');
207        assert_eq!(state.get::<String>(KEY_DTMF), Some("#".into()));
208        assert_eq!(state.get::<String>(KEY_DTMF_HISTORY), Some("4#".into()));
209    }
210
211    #[tokio::test(start_paused = true)]
212    async fn filler_fires_after_silence_and_respects_min_interval() {
213        let (event_tx, event_rx) = broadcast::channel(16);
214        let (speaker_tx, mut speaker_rx) = mpsc::channel(4);
215        let config = FillerConfig::new(vec![7i16; 80])
216            .delay(Duration::from_secs(2))
217            .min_interval(Duration::from_secs(10));
218        let task = tokio::spawn(filler_task(event_rx, speaker_tx, config));
219
220        // Caller stops speaking; nothing for 2 s → filler plays.
221        event_tx.send(LiveEvent::VadEnd).unwrap();
222        tokio::time::sleep(Duration::from_millis(2100)).await;
223        match speaker_rx.recv().await {
224            Some(Playback::Chunk(samples)) => assert_eq!(samples, vec![7i16; 80]),
225            other => panic!("expected filler chunk, got {other:?}"),
226        }
227
228        // A second silence inside min_interval stays quiet.
229        event_tx.send(LiveEvent::VadEnd).unwrap();
230        tokio::time::sleep(Duration::from_millis(2100)).await;
231        assert!(
232            speaker_rx.try_recv().is_err(),
233            "min_interval suppresses a second filler"
234        );
235
236        drop(event_tx);
237        let _ = task.await;
238    }
239
240    #[tokio::test(start_paused = true)]
241    async fn model_audio_disarms_the_filler() {
242        let (event_tx, event_rx) = broadcast::channel(16);
243        let (speaker_tx, mut speaker_rx) = mpsc::channel(4);
244        let task = tokio::spawn(filler_task(
245            event_rx,
246            speaker_tx,
247            FillerConfig::new(vec![1i16]).delay(Duration::from_secs(2)),
248        ));
249
250        event_tx.send(LiveEvent::VadEnd).unwrap();
251        tokio::time::sleep(Duration::from_millis(500)).await;
252        // The model answers within the window — no filler.
253        event_tx
254            .send(LiveEvent::Audio(bytes::Bytes::from_static(&[0, 0])))
255            .unwrap();
256        tokio::time::sleep(Duration::from_secs(5)).await;
257        assert!(speaker_rx.try_recv().is_err(), "audio disarmed the filler");
258
259        drop(event_tx);
260        let _ = task.await;
261    }
262}
gemini_adk_fluent_rs/telephony/bridge.rs

gemini_adk_fluent_rs/telephony/
bridge.rs
