gemini_adk_fluent_rs/telephony/
twilio.rs

1//! Twilio Media Streams — a phone call as two channels of JSON text frames.
2//!
3//! Twilio's [Media Streams](https://www.twilio.com/docs/voice/media-streams)
4//! forks a live call's audio over a WebSocket: μ-law 8 kHz frames arrive as
5//! base64 inside JSON text messages, and JSON text messages you send play
6//! back to the caller. This module speaks that protocol and adapts it onto
7//! [`voice::pump`](crate::voice::pump) — so a governed Live session answers
8//! the phone with the same barge-in guarantees as a local microphone:
9//!
10//! - inbound `media` frames are μ-law-decoded to PCM16 @ 8 kHz and fed to the
11//!   pump, which resamples to the session's 16 kHz input;
12//! - session audio comes back from the pump at 8 kHz, is μ-law-encoded, and
13//!   sent as outbound `media` frames;
14//! - an interruption ([`Playback::Flush`](crate::voice::Playback)) becomes a
15//!   Twilio `clear` message, dropping every frame Twilio has buffered — the
16//!   telephone form of "stop talking the instant the caller does";
17//! - DTMF digits land in session state under [`KEY_DTMF`] / [`KEY_DTMF_HISTORY`],
18//!   where flow guards and watchers read them like any other fact.
19//!
20//! The bridge owns no socket. [`TwilioCall::attach`] returns a sender for the
21//! text frames Twilio delivers and a receiver of the text frames to forward
22//! back — wire them to any WebSocket server (see `examples/telephony` for an
23//! axum one). Raw SIP (no Twilio in the path) is out of scope here; that is
24//! the roadmap's `rsipstack` integration.
25
26use std::collections::HashMap;
27
28use base64::Engine as _;
29use serde::{Deserialize, Serialize};
30use tokio::sync::{mpsc, watch};
31use tokio::task::JoinHandle;
32
33use gemini_adk_rs::live::LiveHandle;
34
35use super::g711;
36use crate::voice::{Playback, VoicePump, pump};
37
38/// The sample rate of every Twilio Media Stream, both directions.
39pub const TWILIO_HZ: u32 = 8_000;
40
41// The state-key vocabulary is shared across every connector — see
42// [`super::bridge`]. Re-exported here so existing imports keep working.
43pub use super::bridge::{KEY_CALL_SID, KEY_DTMF, KEY_DTMF_HISTORY, KEY_STREAM_SID};
44
45// ── Inbound protocol (Twilio → us) ───────────────────────────────────────────
46
47/// Metadata delivered by Twilio's `start` frame when the stream opens.
48#[derive(Debug, Clone, Deserialize, Default, PartialEq)]
49#[serde(rename_all = "camelCase", default)]
50pub struct StartMeta {
51    /// The stream SID — required in every frame sent back to Twilio.
52    pub stream_sid: String,
53    /// The SID of the underlying voice call.
54    pub call_sid: String,
55    /// The Twilio account SID.
56    pub account_sid: String,
57    /// Which tracks are being forked (normally `["inbound"]`).
58    pub tracks: Vec<String>,
59    /// Audio format of the stream (normally `audio/x-mulaw` @ 8000 Hz mono).
60    pub media_format: MediaFormat,
61    /// `<Parameter>` values set on the `<Stream>` TwiML noun.
62    pub custom_parameters: HashMap<String, String>,
63}
64
65/// The audio encoding Twilio declares in the `start` frame.
66#[derive(Debug, Clone, Deserialize, Default, PartialEq)]
67#[serde(rename_all = "camelCase", default)]
68pub struct MediaFormat {
69    /// MIME-style encoding name, e.g. `audio/x-mulaw`.
70    pub encoding: String,
71    /// Sample rate in Hz (8000 for telephone audio).
72    pub sample_rate: u32,
73    /// Channel count (1 for a single forked track).
74    pub channels: u32,
75}
76
77#[derive(Debug, Clone, Deserialize)]
78#[serde(rename_all = "camelCase")]
79struct MediaPayload {
80    #[serde(default)]
81    track: String,
82    payload: String,
83}
84
85#[derive(Debug, Clone, Deserialize)]
86struct DtmfPayload {
87    digit: String,
88}
89
90#[derive(Debug, Clone, Deserialize)]
91struct MarkPayload {
92    name: String,
93}
94
95#[derive(Debug, Clone, Deserialize)]
96#[serde(tag = "event", rename_all = "lowercase")]
97enum RawInbound {
98    Connected {},
99    Start { start: StartMeta },
100    Media { media: MediaPayload },
101    Dtmf { dtmf: DtmfPayload },
102    Mark { mark: MarkPayload },
103    Stop {},
104}
105
106/// One decoded frame from Twilio, ready for application handling.
107#[derive(Debug, Clone, PartialEq)]
108pub enum Inbound {
109    /// The WebSocket handshake completed (`connected`). No stream SID yet.
110    Connected,
111    /// The stream opened; carries SIDs, tracks, format, and TwiML parameters.
112    Started(StartMeta),
113    /// One chunk of caller audio, μ-law-decoded to mono PCM16 @ 8 kHz.
114    Audio(Vec<i16>),
115    /// The caller pressed a DTMF key.
116    Dtmf(char),
117    /// Twilio confirms playback reached a `mark` we sent.
118    Mark(String),
119    /// The stream ended (call hung up or `<Stream>` stopped).
120    Stopped,
121    /// An event this version does not model (forward-compatible skip).
122    Ignored,
123}
124
125/// Parse one Twilio Media Streams text frame.
126///
127/// Unknown event types parse to [`Inbound::Ignored`] so a new Twilio event
128/// never breaks an existing bridge; malformed JSON is an error.
129pub fn parse_inbound(text: &str) -> Result<Inbound, TwilioError> {
130    let raw: RawInbound = match serde_json::from_str(text) {
131        Ok(raw) => raw,
132        Err(_) => {
133            // Distinguish "not JSON" from "JSON with an unknown event tag".
134            let value: serde_json::Value =
135                serde_json::from_str(text).map_err(TwilioError::Malformed)?;
136            return if value.get("event").is_some() {
137                Ok(Inbound::Ignored)
138            } else {
139                Err(TwilioError::NotAFrame)
140            };
141        }
142    };
143    Ok(match raw {
144        RawInbound::Connected {} => Inbound::Connected,
145        RawInbound::Start { start } => Inbound::Started(start),
146        RawInbound::Media { media } => {
147            // Only the caller's track feeds the session; if outbound audio is
148            // also forked (`tracks: ["both"]`), skip our own voice.
149            if !media.track.is_empty() && media.track != "inbound" {
150                return Ok(Inbound::Ignored);
151            }
152            let mulaw = base64::engine::general_purpose::STANDARD
153                .decode(media.payload.as_bytes())
154                .map_err(TwilioError::BadPayload)?;
155            Inbound::Audio(g711::decode_ulaw(&mulaw))
156        }
157        RawInbound::Dtmf { dtmf } => match dtmf.digit.chars().next() {
158            Some(digit) => Inbound::Dtmf(digit),
159            None => Inbound::Ignored,
160        },
161        RawInbound::Mark { mark } => Inbound::Mark(mark.name),
162        RawInbound::Stop {} => Inbound::Stopped,
163    })
164}
165
166/// Errors from parsing Twilio frames.
167#[derive(Debug)]
168pub enum TwilioError {
169    /// The text was not valid JSON.
170    Malformed(serde_json::Error),
171    /// The JSON carried no `event` field — not a Media Streams frame.
172    NotAFrame,
173    /// A `media` frame's payload was not valid base64.
174    BadPayload(base64::DecodeError),
175}
176
177impl std::fmt::Display for TwilioError {
178    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
179        match self {
180            Self::Malformed(e) => write!(f, "malformed Twilio frame: {e}"),
181            Self::NotAFrame => write!(f, "JSON without an event field"),
182            Self::BadPayload(e) => write!(f, "invalid base64 media payload: {e}"),
183        }
184    }
185}
186
187impl std::error::Error for TwilioError {}
188
189// ── Outbound protocol (us → Twilio) ──────────────────────────────────────────
190
191#[derive(Serialize)]
192#[serde(rename_all = "camelCase")]
193struct OutMedia<'a> {
194    event: &'static str,
195    stream_sid: &'a str,
196    media: OutMediaPayload,
197}
198
199#[derive(Serialize)]
200struct OutMediaPayload {
201    payload: String,
202}
203
204#[derive(Serialize)]
205#[serde(rename_all = "camelCase")]
206struct OutClear<'a> {
207    event: &'static str,
208    stream_sid: &'a str,
209}
210
211#[derive(Serialize)]
212#[serde(rename_all = "camelCase")]
213struct OutMark<'a> {
214    event: &'static str,
215    stream_sid: &'a str,
216    mark: OutMarkPayload<'a>,
217}
218
219#[derive(Serialize)]
220struct OutMarkPayload<'a> {
221    name: &'a str,
222}
223
224/// Build an outbound `media` frame from mono PCM16 samples @ 8 kHz.
225pub fn media_frame(stream_sid: &str, samples: &[i16]) -> String {
226    let payload = base64::engine::general_purpose::STANDARD.encode(g711::encode_ulaw(samples));
227    serde_json::to_string(&OutMedia {
228        event: "media",
229        stream_sid,
230        media: OutMediaPayload { payload },
231    })
232    .expect("media frame serializes")
233}
234
235/// Build a `clear` frame — Twilio drops all buffered outbound audio.
236/// This is barge-in on the telephone: send it on [`Playback::Flush`].
237pub fn clear_frame(stream_sid: &str) -> String {
238    serde_json::to_string(&OutClear {
239        event: "clear",
240        stream_sid,
241    })
242    .expect("clear frame serializes")
243}
244
245/// Build a `mark` frame — Twilio echoes it back once playback reaches it.
246pub fn mark_frame(stream_sid: &str, name: &str) -> String {
247    serde_json::to_string(&OutMark {
248        event: "mark",
249        stream_sid,
250        mark: OutMarkPayload { name },
251    })
252    .expect("mark frame serializes")
253}
254
255// ── Call bridge ──────────────────────────────────────────────────────────────
256
257/// A live phone call attached to a session — the telephone counterpart of
258/// `Talk::talk`.
259///
260/// [`attach`](TwilioCall::attach) wires the session's [`pump`] to a pair of
261/// text-frame channels speaking the Media Streams protocol. The caller owns
262/// the WebSocket: forward every text message Twilio delivers into
263/// [`from_twilio`](TwilioCall::from_twilio), and forward every message from
264/// [`to_twilio`](TwilioCall::to_twilio) back over the socket.
265///
266/// ```ignore
267/// // `ignore`: `ws` is the application's Twilio WebSocket (see examples/telephony).
268/// let session = Live::builder()
269///     .instruction("You are the front desk. Answer the call.")
270///     .greeting("Greet the caller.")
271///     .connect_from_env().await?;
272/// let mut call = TwilioCall::attach(&session);
273/// loop {
274///     tokio::select! {
275///         Some(msg) = ws.recv() => call.from_twilio.send(msg.into_text()?).await?,
276///         Some(out) = call.to_twilio.recv() => ws.send(Message::Text(out)).await?,
277///         else => break,
278///     }
279/// }
280/// ```
281pub struct TwilioCall {
282    /// Feed Twilio's inbound WebSocket text frames here.
283    pub from_twilio: mpsc::Sender<String>,
284    /// Text frames to send back to Twilio over the WebSocket.
285    pub to_twilio: mpsc::Receiver<String>,
286    pump: VoicePump,
287    inbound_task: JoinHandle<()>,
288    outbound_task: JoinHandle<()>,
289}
290
291impl TwilioCall {
292    /// Attach a Twilio Media Stream to a connected session.
293    ///
294    /// Audio starts flowing once Twilio's `start` frame arrives (outbound
295    /// frames need its stream SID; audio generated before it is dropped —
296    /// there is no call to play it into yet). Stream metadata and DTMF
297    /// digits are written into session state under the `telephony:` keys.
298    pub fn attach(handle: &LiveHandle) -> TwilioCall {
299        let (from_tx, mut from_rx) = mpsc::channel::<String>(64);
300        let (to_tx, to_rx) = mpsc::channel::<String>(64);
301        let (mic_tx, mic_rx) = mpsc::channel::<Vec<i16>>(64);
302        let (speaker_tx, mut speaker_rx) = mpsc::channel::<Playback>(64);
303        let (sid_tx, sid_rx) = watch::channel::<Option<String>>(None);
304
305        let voice_pump = pump(handle, mic_rx, TWILIO_HZ, speaker_tx, TWILIO_HZ);
306
307        let state = handle.state().clone();
308        let inbound_task = tokio::spawn(async move {
309            while let Some(text) = from_rx.recv().await {
310                match parse_inbound(&text) {
311                    Ok(Inbound::Audio(samples)) => {
312                        if mic_tx.send(samples).await.is_err() {
313                            break;
314                        }
315                    }
316                    Ok(Inbound::Started(meta)) => {
317                        let _ = state.set(KEY_CALL_SID, meta.call_sid.clone());
318                        let _ = state.set(KEY_STREAM_SID, meta.stream_sid.clone());
319                        let _ = sid_tx.send(Some(meta.stream_sid));
320                    }
321                    Ok(Inbound::Dtmf(digit)) => super::bridge::record_dtmf(&state, digit),
322                    Ok(Inbound::Stopped) => break,
323                    Ok(Inbound::Connected | Inbound::Mark(_) | Inbound::Ignored) => {}
324                    Err(err) => tracing::warn!("dropping unparseable Twilio frame: {err}"),
325                }
326            }
327        });
328
329        let outbound_task = tokio::spawn(async move {
330            while let Some(playback) = speaker_rx.recv().await {
331                // Frames are unsendable until `start` delivers the SID.
332                let Some(sid) = sid_rx.borrow().clone() else {
333                    continue;
334                };
335                let frame = match &playback {
336                    Playback::Chunk(samples) => media_frame(&sid, samples),
337                    Playback::Flush => clear_frame(&sid),
338                };
339                if to_tx.send(frame).await.is_err() {
340                    break;
341                }
342            }
343        });
344
345        TwilioCall {
346            from_twilio: from_tx,
347            to_twilio: to_rx,
348            pump: voice_pump,
349            inbound_task,
350            outbound_task,
351        }
352    }
353
354    /// Wait until the call ends (stream stopped, session closed, or the
355    /// WebSocket side dropped the channels).
356    pub async fn join(self) {
357        let _ = self.inbound_task.await;
358        let _ = self.outbound_task.await;
359        self.pump.join().await;
360    }
361
362    /// Tear the bridge down immediately.
363    pub fn abort(&self) {
364        self.inbound_task.abort();
365        self.outbound_task.abort();
366        self.pump.abort();
367    }
368}
369
370#[cfg(test)]
371mod tests {
372    use super::*;
373
374    #[test]
375    fn parses_the_start_frame() {
376        let text = r#"{
377            "event": "start", "sequenceNumber": "1", "streamSid": "MZxyz",
378            "start": {
379                "accountSid": "ACabc", "streamSid": "MZxyz", "callSid": "CAdef",
380                "tracks": ["inbound"],
381                "mediaFormat": {"encoding": "audio/x-mulaw", "sampleRate": 8000, "channels": 1},
382                "customParameters": {"agent": "front-desk"}
383            }
384        }"#;
385        match parse_inbound(text).unwrap() {
386            Inbound::Started(meta) => {
387                assert_eq!(meta.stream_sid, "MZxyz");
388                assert_eq!(meta.call_sid, "CAdef");
389                assert_eq!(meta.media_format.sample_rate, 8000);
390                assert_eq!(meta.custom_parameters["agent"], "front-desk");
391            }
392            other => panic!("expected Started, got {other:?}"),
393        }
394    }
395
396    #[test]
397    fn media_frames_decode_to_pcm() {
398        // 0xFF is μ-law silence; four bytes → four zero samples.
399        let payload = base64::engine::general_purpose::STANDARD.encode([0xFFu8; 4]);
400        let text = format!(
401            r#"{{"event":"media","streamSid":"MZ1","media":{{"track":"inbound","chunk":"1","timestamp":"5","payload":"{payload}"}}}}"#
402        );
403        assert_eq!(parse_inbound(&text).unwrap(), Inbound::Audio(vec![0i16; 4]));
404    }
405
406    #[test]
407    fn outbound_track_media_is_skipped() {
408        let payload = base64::engine::general_purpose::STANDARD.encode([0xFFu8; 4]);
409        let text = format!(
410            r#"{{"event":"media","streamSid":"MZ1","media":{{"track":"outbound","payload":"{payload}"}}}}"#
411        );
412        assert_eq!(parse_inbound(&text).unwrap(), Inbound::Ignored);
413    }
414
415    #[test]
416    fn dtmf_marks_stop_and_unknown_events() {
417        assert_eq!(
418            parse_inbound(
419                r#"{"event":"dtmf","streamSid":"MZ1","dtmf":{"track":"inbound_track","digit":"7"}}"#
420            )
421            .unwrap(),
422            Inbound::Dtmf('7')
423        );
424        assert_eq!(
425            parse_inbound(r#"{"event":"mark","streamSid":"MZ1","mark":{"name":"m1"}}"#).unwrap(),
426            Inbound::Mark("m1".into())
427        );
428        assert_eq!(
429            parse_inbound(r#"{"event":"stop","streamSid":"MZ1","stop":{}}"#).unwrap(),
430            Inbound::Stopped
431        );
432        assert_eq!(
433            parse_inbound(r#"{"event":"connected","protocol":"Call","version":"1.0.0"}"#).unwrap(),
434            Inbound::Connected
435        );
436        // Forward compatibility: an event we don't model is skipped, not fatal.
437        assert_eq!(
438            parse_inbound(r#"{"event":"totally-new-thing"}"#).unwrap(),
439            Inbound::Ignored
440        );
441        assert!(parse_inbound("not json").is_err());
442        assert!(parse_inbound(r#"{"no_event": true}"#).is_err());
443    }
444
445    #[test]
446    fn media_frame_round_trips_through_the_codec() {
447        let samples = vec![0i16, 1000, -1000, 8000];
448        let frame = media_frame("MZ1", &samples);
449        let value: serde_json::Value = serde_json::from_str(&frame).unwrap();
450        assert_eq!(value["event"], "media");
451        assert_eq!(value["streamSid"], "MZ1");
452        let mulaw = base64::engine::general_purpose::STANDARD
453            .decode(value["media"]["payload"].as_str().unwrap())
454            .unwrap();
455        let decoded = g711::decode_ulaw(&mulaw);
456        assert_eq!(decoded.len(), samples.len());
457        // Companded round-trip: close, not exact.
458        for (orig, rt) in samples.iter().zip(&decoded) {
459            assert!(((orig - rt) as i32).abs() <= (orig.unsigned_abs() as i32 / 16).max(16));
460        }
461    }
462
463    #[test]
464    fn clear_and_mark_frames_have_the_wire_shape() {
465        let clear: serde_json::Value = serde_json::from_str(&clear_frame("MZ9")).unwrap();
466        assert_eq!(
467            clear,
468            serde_json::json!({"event": "clear", "streamSid": "MZ9"})
469        );
470        let mark: serde_json::Value = serde_json::from_str(&mark_frame("MZ9", "done")).unwrap();
471        assert_eq!(
472            mark,
473            serde_json::json!({"event": "mark", "streamSid": "MZ9", "mark": {"name": "done"}})
474        );
475    }
476}
gemini_adk_fluent_rs/telephony/twilio.rs

gemini_adk_fluent_rs/telephony/
twilio.rs
