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/// Set this state key to `true` to mask the keypad: while it is set, the
44/// caller's audio reaches neither the model nor the recording (it is
45/// replaced with silence), and keypresses go to [`KEY_KEYPAD_ENTRY`] instead
46/// of [`KEY_DTMF`]. This is the "pause and resume" pattern for taking card
47/// numbers and PINs by keypad. Set it back to `false` to resume.
48pub const KEY_KEYPAD_MASK: &str = "telephony:keypad_mask";
49/// Digits keyed while the keypad was masked, in order. The key is marked
50/// sensitive ([`State::redact_keys`]): tools read the real value, while
51/// journals, exports and telemetry see it redacted.
52pub const KEY_KEYPAD_ENTRY: &str = "telephony:keypad_entry";
53/// How many digits [`KEY_KEYPAD_ENTRY`] holds. Not sensitive, so a flow can
54/// guard on it.
55pub const KEY_KEYPAD_LEN: &str = "telephony:keypad_len";
56/// Set to `true` when the caller presses `#` while the keypad is masked.
57pub const KEY_KEYPAD_DONE: &str = "telephony:keypad_done";
58
59/// Whether the keypad is masked (see [`KEY_KEYPAD_MASK`]).
60pub fn keypad_masked(state: &State) -> bool {
61    state.get::<bool>(KEY_KEYPAD_MASK).unwrap_or(false)
62}
63
64/// Record one DTMF keypress into session state under the shared keys.
65///
66/// Sets [`KEY_DTMF`] to the digit and appends it to [`KEY_DTMF_HISTORY`] —
67/// the exact writes every connector must make, factored to one place. While
68/// the keypad is masked ([`KEY_KEYPAD_MASK`]) the digit goes to the
69/// sensitive [`KEY_KEYPAD_ENTRY`] instead, and `#` ends the entry.
70pub fn record_dtmf(state: &State, digit: char) {
71    if keypad_masked(state) {
72        state.redact_keys([KEY_KEYPAD_ENTRY]);
73        if digit == '#' {
74            let _ = state.set(KEY_KEYPAD_DONE, true);
75            return;
76        }
77        let mut len = 0usize;
78        let _ = state.modify(KEY_KEYPAD_ENTRY, String::new(), |mut entry| {
79            entry.push(digit);
80            len = entry.len();
81            entry
82        });
83        let _ = state.set(KEY_KEYPAD_LEN, len);
84        return;
85    }
86    let _ = state.set(KEY_DTMF, digit.to_string());
87    let _ = state.modify(KEY_DTMF_HISTORY, String::new(), |mut history| {
88        history.push(digit);
89        history
90    });
91}
92
93/// Keeps keypad tones and masked keypad entry away from the model.
94///
95/// An [`InputAudioProcessor`](crate::voice::InputAudioProcessor) for the
96/// caller's audio at the connector's rate:
97///
98/// - Frames carrying DTMF tones are silenced (from the first detection, for
99///   60 ms after), so the model neither hears nor transcribes keypresses.
100///   Up to one detection block (26 ms) of a tone's onset can pass before it
101///   is recognized; that is too short to be decoded.
102/// - While [`KEY_KEYPAD_MASK`] is set, every frame is silenced.
103/// - With [`record_in_band`](Self::record_in_band), digits found in the audio
104///   are recorded with [`record_dtmf`]. Use it for transports without
105///   out-of-band DTMF; where the platform reports keypresses itself (Twilio
106///   `dtmf` events, RFC 4733), leave it off so digits aren't counted twice.
107pub struct KeypadGuard {
108    state: State,
109    detector: super::dtmf::DtmfDetector,
110    record_in_band: bool,
111    hold_samples: usize,
112    held: usize,
113}
114
115impl KeypadGuard {
116    /// A guard for caller audio at `sample_rate`, writing to `state`.
117    pub fn new(state: State, sample_rate: u32) -> Self {
118        Self {
119            state,
120            detector: super::dtmf::DtmfDetector::new(sample_rate),
121            record_in_band: false,
122            hold_samples: (sample_rate as usize * 60) / 1000,
123            held: 0,
124        }
125    }
126
127    /// Also record digits detected in the audio (see the type docs).
128    pub fn record_in_band(mut self, enabled: bool) -> Self {
129        self.record_in_band = enabled;
130        self
131    }
132}
133
134impl crate::voice::InputAudioProcessor for KeypadGuard {
135    fn process_frame(&mut self, frame: &mut Vec<i16>) {
136        for digit in self.detector.feed(frame) {
137            if self.record_in_band {
138                record_dtmf(&self.state, digit);
139            }
140        }
141        let tone = self.detector.tone_present();
142        if tone {
143            self.held = self.hold_samples;
144        }
145        let silence = tone || self.held > 0 || keypad_masked(&self.state);
146        self.held = self.held.saturating_sub(frame.len());
147        if silence {
148            frame.fill(0);
149        }
150    }
151}
152
153/// Deduplicates RFC 4733 end-of-event packets.
154///
155/// A telephone-event keypress ends with its final packet conventionally
156/// retransmitted three times, all sharing one RTP timestamp. Feed every
157/// end-marked event through [`accept`](Self::accept); only the first per
158/// timestamp comes back `true`.
159#[derive(Debug, Default)]
160pub struct DtmfDeduper {
161    last_end_timestamp: Option<u32>,
162}
163
164impl DtmfDeduper {
165    /// `true` exactly once per keypress: for the first end-marked packet
166    /// carrying a given RTP timestamp.
167    pub fn accept(&mut self, end: bool, rtp_timestamp: u32) -> bool {
168        if !end {
169            return false;
170        }
171        if self.last_end_timestamp == Some(rtp_timestamp) {
172            return false;
173        }
174        self.last_end_timestamp = Some(rtp_timestamp);
175        true
176    }
177}
178
179// ── Latency filler ───────────────────────────────────────────────────────────
180
181/// Configuration for [`spawn_latency_filler`].
182#[derive(Clone)]
183pub struct FillerConfig {
184    /// The filler clip: mono PCM16 at the connector's playback sample rate
185    /// (the `speaker_hz` given to [`voice::pump`](crate::voice::pump)) —
186    /// e.g. a pre-synthesized "one moment, let me check that".
187    pub clip: Arc<Vec<i16>>,
188    /// Silence to tolerate after the caller stops speaking before playing
189    /// the clip. Below ~1.5 s the filler fires on normal model latency and
190    /// talks over the answer's first syllables.
191    pub delay: Duration,
192    /// At most one filler per this interval, so a long tool call gets one
193    /// reassurance, not a loop of them.
194    pub min_interval: Duration,
195}
196
197impl FillerConfig {
198    /// A filler clip with the conventional pacing: 2 s of tolerated silence,
199    /// at most one filler per 10 s.
200    pub fn new(clip: Vec<i16>) -> Self {
201        Self {
202            clip: Arc::new(clip),
203            delay: Duration::from_secs(2),
204            min_interval: Duration::from_secs(10),
205        }
206    }
207
208    /// Override the tolerated-silence delay.
209    pub fn delay(mut self, delay: Duration) -> Self {
210        self.delay = delay;
211        self
212    }
213
214    /// Override the per-filler minimum interval.
215    pub fn min_interval(mut self, interval: Duration) -> Self {
216        self.min_interval = interval;
217        self
218    }
219}
220
221/// Keep the caller's ear busy while the model is slow: when the caller stops
222/// speaking ([`LiveEvent::VadEnd`]) and no model audio arrives within
223/// `config.delay`, inject the configured clip into the connector's playback
224/// channel. Model audio or an interruption disarms the timer; a
225/// [`Playback::Flush`] from a barge-in clears any queued filler exactly like
226/// any other queued audio.
227///
228/// Masking is not a substitute for reducing latency — it buys tolerance for
229/// the tail, and because it is driven by the same event stream as the
230/// telemetry lane, the silences it papers over remain visible in the
231/// latency metrics.
232///
233/// The task ends when the session's event stream closes; abort the handle to
234/// stop it sooner.
235pub fn spawn_latency_filler(
236    handle: &LiveHandle,
237    speaker: mpsc::Sender<Playback>,
238    config: FillerConfig,
239) -> JoinHandle<()> {
240    let events = handle.events();
241    tokio::spawn(filler_task(events, speaker, config))
242}
243
244/// The filler loop itself, taking the event stream directly — the seam tests
245/// drive without a session.
246pub(crate) async fn filler_task(
247    mut events: broadcast::Receiver<LiveEvent>,
248    speaker: mpsc::Sender<Playback>,
249    config: FillerConfig,
250) {
251    let mut armed_at: Option<tokio::time::Instant> = None;
252    let mut last_filler: Option<tokio::time::Instant> = None;
253    loop {
254        let deadline = armed_at.map(|at| at + config.delay);
255        tokio::select! {
256            event = events.recv() => match event {
257                Ok(LiveEvent::VadEnd) => armed_at = Some(tokio::time::Instant::now()),
258                // Model audio (or the user cutting in) means silence ended.
259                Ok(LiveEvent::Audio(_)) | Ok(LiveEvent::VadStart) | Ok(LiveEvent::Interrupted) => {
260                    armed_at = None;
261                }
262                Ok(_) => {}
263                Err(broadcast::error::RecvError::Lagged(_)) => continue,
264                Err(broadcast::error::RecvError::Closed) => break,
265            },
266            () = async {
267                match deadline {
268                    Some(deadline) => tokio::time::sleep_until(deadline).await,
269                    None => std::future::pending().await,
270                }
271            } => {
272                armed_at = None;
273                let recently = last_filler
274                    .is_some_and(|at| at.elapsed() < config.min_interval);
275                if !recently {
276                    last_filler = Some(tokio::time::Instant::now());
277                    let _ = speaker
278                        .send(Playback::Chunk(config.clip.as_ref().clone()))
279                        .await;
280                }
281            }
282        }
283    }
284}
285
286#[cfg(test)]
287mod tests {
288    use super::*;
289
290    #[test]
291    fn dtmf_dedup_accepts_one_end_per_timestamp() {
292        let mut dedup = DtmfDeduper::default();
293        assert!(!dedup.accept(false, 100), "non-end packets never emit");
294        assert!(dedup.accept(true, 100), "first end emits");
295        assert!(!dedup.accept(true, 100), "retransmitted end is dropped");
296        assert!(!dedup.accept(true, 100));
297        assert!(dedup.accept(true, 900), "next keypress emits again");
298    }
299
300    #[test]
301    fn record_dtmf_writes_the_shared_keys() {
302        let state = State::new();
303        record_dtmf(&state, '4');
304        record_dtmf(&state, '#');
305        assert_eq!(state.get::<String>(KEY_DTMF), Some("#".into()));
306        assert_eq!(state.get::<String>(KEY_DTMF_HISTORY), Some("4#".into()));
307    }
308
309    #[test]
310    fn a_masked_keypad_keeps_digits_out_of_the_shared_keys() {
311        let state = State::new();
312        record_dtmf(&state, '1');
313        let _ = state.set(KEY_KEYPAD_MASK, true);
314        for digit in "4242#".chars() {
315            record_dtmf(&state, digit);
316        }
317        let _ = state.set(KEY_KEYPAD_MASK, false);
318        // The shared keys never saw the masked digits.
319        assert_eq!(state.get::<String>(KEY_DTMF_HISTORY), Some("1".into()));
320        assert_eq!(state.get::<String>(KEY_KEYPAD_ENTRY), Some("4242".into()));
321        assert_eq!(state.get::<usize>(KEY_KEYPAD_LEN), Some(4));
322        assert_eq!(state.get::<bool>(KEY_KEYPAD_DONE), Some(true));
323        // The entry is sensitive wherever state leaves the process.
324        assert!(state.redacted_keys().contains(KEY_KEYPAD_ENTRY));
325    }
326
327    #[test]
328    fn the_keypad_guard_silences_tones_and_masked_speech() {
329        use crate::voice::InputAudioProcessor as _;
330        let state = State::new();
331        let mut guard = KeypadGuard::new(state.clone(), 8_000).record_in_band(true);
332        let tones = crate::telephony::dtmf::tests::keypresses("73", 80, 60, 6_000.0);
333        let mut heard = Vec::new();
334        for chunk in tones.chunks(160) {
335            let mut frame = chunk.to_vec();
336            guard.process_frame(&mut frame);
337            heard.extend(frame);
338        }
339        assert_eq!(state.get::<String>(KEY_DTMF_HISTORY), Some("73".into()));
340        // At most one detection block of each tone's onset gets through.
341        let leaked = heard.iter().filter(|&&s| s != 0).count();
342        assert!(leaked <= 2 * 205, "{leaked} tone samples reached the model");
343
344        // Speech passes, until the keypad is masked.
345        let speech: Vec<i16> = (0..160).map(|n| ((n % 40) as i16 - 20) * 300).collect();
346        let mut frame = speech.clone();
347        guard.process_frame(&mut frame);
348        assert_eq!(frame, speech);
349        let _ = state.set(KEY_KEYPAD_MASK, true);
350        let mut frame = speech.clone();
351        guard.process_frame(&mut frame);
352        assert!(frame.iter().all(|&s| s == 0));
353    }
354
355    #[tokio::test(start_paused = true)]
356    async fn filler_fires_after_silence_and_respects_min_interval() {
357        let (event_tx, event_rx) = broadcast::channel(16);
358        let (speaker_tx, mut speaker_rx) = mpsc::channel(4);
359        let config = FillerConfig::new(vec![7i16; 80])
360            .delay(Duration::from_secs(2))
361            .min_interval(Duration::from_secs(10));
362        let task = tokio::spawn(filler_task(event_rx, speaker_tx, config));
363
364        // Caller stops speaking; nothing for 2 s → filler plays.
365        event_tx.send(LiveEvent::VadEnd).unwrap();
366        tokio::time::sleep(Duration::from_millis(2100)).await;
367        match speaker_rx.recv().await {
368            Some(Playback::Chunk(samples)) => assert_eq!(samples, vec![7i16; 80]),
369            other => panic!("expected filler chunk, got {other:?}"),
370        }
371
372        // A second silence inside min_interval stays quiet.
373        event_tx.send(LiveEvent::VadEnd).unwrap();
374        tokio::time::sleep(Duration::from_millis(2100)).await;
375        assert!(
376            speaker_rx.try_recv().is_err(),
377            "min_interval suppresses a second filler"
378        );
379
380        drop(event_tx);
381        let _ = task.await;
382    }
383
384    #[tokio::test(start_paused = true)]
385    async fn model_audio_disarms_the_filler() {
386        let (event_tx, event_rx) = broadcast::channel(16);
387        let (speaker_tx, mut speaker_rx) = mpsc::channel(4);
388        let task = tokio::spawn(filler_task(
389            event_rx,
390            speaker_tx,
391            FillerConfig::new(vec![1i16]).delay(Duration::from_secs(2)),
392        ));
393
394        event_tx.send(LiveEvent::VadEnd).unwrap();
395        tokio::time::sleep(Duration::from_millis(500)).await;
396        // The model answers within the window — no filler.
397        event_tx
398            .send(LiveEvent::Audio(bytes::Bytes::from_static(&[0, 0])))
399            .unwrap();
400        tokio::time::sleep(Duration::from_secs(5)).await;
401        assert!(speaker_rx.try_recv().is_err(), "audio disarmed the filler");
402
403        drop(event_tx);
404        let _ = task.await;
405    }
406}
gemini_adk_fluent_rs/telephony/bridge.rs

gemini_adk_fluent_rs/telephony/
bridge.rs
