gemini_adk_fluent_rs/telephony/
recorder.rs

1//! Stereo call recording for compliance and quality review.
2//!
3//! [`CallRecorder`] writes a 16-bit PCM WAV with the caller on the left
4//! channel and the agent on the right, aligned to the call's own clock:
5//!
6//! - Caller audio arrives in real time and is placed at the moment it
7//!   arrived. A gap in delivery becomes silence.
8//! - Agent audio arrives faster than it plays, since the model streams ahead
9//!   of the phone line. Each chunk is placed where playback reaches it: after
10//!   the audio already queued, or now if the queue is empty.
11//! - On barge-in, [`flush`](CallRecorder::flush) cuts the agent audio that
12//!   was queued but not yet played, so the recording holds what the caller
13//!   heard, not what the model generated. It returns how much was cut.
14//!
15//! Samples are written to disk as soon as they can no longer change (up to
16//! the current moment), so memory stays flat for a call of any length. Keypad
17//! masking applies upstream: while [`KeypadGuard`](super::bridge::KeypadGuard)
18//! silences the caller, the recorder receives the silence too.
19
20use std::collections::VecDeque;
21use std::fs::File;
22use std::io::{BufWriter, Seek, SeekFrom, Write};
23use std::path::Path;
24use std::time::{Duration, Instant};
25
26use parking_lot::Mutex;
27
28/// A stereo WAV recording of one call. See the module docs.
29pub struct CallRecorder {
30    inner: Mutex<Inner>,
31}
32
33struct Inner {
34    writer: Option<BufWriter<File>>,
35    rate: u32,
36    start: Instant,
37    /// Frames already written to disk.
38    written: u64,
39    /// Caller samples from position `written` on.
40    caller: VecDeque<i16>,
41    /// Agent samples from position `written` on (silence where none played).
42    agent: VecDeque<i16>,
43}
44
45impl Inner {
46    fn now_frames(&self, at: Duration) -> u64 {
47        (at.as_secs_f64() * f64::from(self.rate)) as u64
48    }
49
50    fn caller_end(&self) -> u64 {
51        self.written + self.caller.len() as u64
52    }
53
54    fn agent_end(&self) -> u64 {
55        self.written + self.agent.len() as u64
56    }
57
58    /// Write every frame before `upto` (caller and agent both final there).
59    fn drain_to(&mut self, upto: u64) -> std::io::Result<()> {
60        let Some(writer) = self.writer.as_mut() else {
61            return Ok(());
62        };
63        while self.written < upto {
64            let left = self.caller.pop_front().unwrap_or(0);
65            let right = self.agent.pop_front().unwrap_or(0);
66            writer.write_all(&left.to_le_bytes())?;
67            writer.write_all(&right.to_le_bytes())?;
68            self.written += 1;
69        }
70        Ok(())
71    }
72}
73
74impl CallRecorder {
75    /// Start recording to `path` (created or truncated) at `sample_rate`,
76    /// the call's rate (8 kHz on the phone network).
77    pub fn create(path: impl AsRef<Path>, sample_rate: u32) -> std::io::Result<Self> {
78        let mut writer = BufWriter::new(File::create(path)?);
79        writer.write_all(&wav_header(sample_rate, 0))?;
80        Ok(Self {
81            inner: Mutex::new(Inner {
82                writer: Some(writer),
83                rate: sample_rate,
84                start: Instant::now(),
85                written: 0,
86                caller: VecDeque::new(),
87                agent: VecDeque::new(),
88            }),
89        })
90    }
91
92    /// Record caller audio that just arrived.
93    pub fn caller(&self, samples: &[i16]) {
94        let at = self.inner.lock().start.elapsed();
95        self.caller_at(at, samples);
96    }
97
98    /// Record agent audio just sent toward the caller.
99    pub fn agent(&self, samples: &[i16]) {
100        let at = self.inner.lock().start.elapsed();
101        self.agent_at(at, samples);
102    }
103
104    /// The caller barged in: drop agent audio not yet played. Returns how
105    /// much was dropped.
106    pub fn flush(&self) -> Duration {
107        let at = self.inner.lock().start.elapsed();
108        self.flush_at(at)
109    }
110
111    /// Finish the file: write what remains and fix up the header. Returns
112    /// the recording's length.
113    pub fn finish(&self) -> std::io::Result<Duration> {
114        let mut inner = self.inner.lock();
115        let at = inner.start.elapsed();
116        let now = inner.now_frames(at);
117        // Agent audio still queued past now was never played.
118        let end = inner.caller_end().max(inner.agent_end().min(now));
119        inner.drain_to(end)?;
120        let frames = inner.written;
121        let rate = inner.rate;
122        if let Some(mut writer) = inner.writer.take() {
123            writer.flush()?;
124            let mut file = writer
125                .into_inner()
126                .map_err(std::io::IntoInnerError::into_error)?;
127            file.seek(SeekFrom::Start(0))?;
128            file.write_all(&wav_header(rate, frames))?;
129            file.sync_all()?;
130        }
131        Ok(Duration::from_secs_f64(frames as f64 / f64::from(rate)))
132    }
133
134    pub(crate) fn caller_at(&self, at: Duration, samples: &[i16]) {
135        let mut inner = self.inner.lock();
136        let arrived = inner.now_frames(at);
137        // The chunk ends now; anything between the last chunk and its start
138        // is a gap in delivery.
139        let begins = arrived.saturating_sub(samples.len() as u64);
140        let pad = begins.saturating_sub(inner.caller_end());
141        inner.caller.extend(std::iter::repeat_n(0, pad as usize));
142        inner.caller.extend(samples);
143        let safe = inner.caller_end().min(arrived);
144        if let Err(e) = inner.drain_to(safe) {
145            tracing::warn!("call recording stopped: {e}");
146            inner.writer = None;
147        }
148    }
149
150    pub(crate) fn agent_at(&self, at: Duration, samples: &[i16]) {
151        let mut inner = self.inner.lock();
152        let now = inner.now_frames(at);
153        let pad = now.saturating_sub(inner.agent_end());
154        inner.agent.extend(std::iter::repeat_n(0, pad as usize));
155        inner.agent.extend(samples);
156    }
157
158    pub(crate) fn flush_at(&self, at: Duration) -> Duration {
159        let mut inner = self.inner.lock();
160        let now = inner.now_frames(at).max(inner.written);
161        let keep = (now - inner.written) as usize;
162        let dropped = inner.agent.len().saturating_sub(keep);
163        inner.agent.truncate(keep);
164        Duration::from_secs_f64(dropped as f64 / f64::from(inner.rate))
165    }
166}
167
168fn wav_header(rate: u32, frames: u64) -> [u8; 44] {
169    let data = u32::try_from(frames * 4).unwrap_or(u32::MAX - 36);
170    let mut h = [0u8; 44];
171    h[0..4].copy_from_slice(b"RIFF");
172    h[4..8].copy_from_slice(&(36 + data).to_le_bytes());
173    h[8..12].copy_from_slice(b"WAVE");
174    h[12..16].copy_from_slice(b"fmt ");
175    h[16..20].copy_from_slice(&16u32.to_le_bytes());
176    h[20..22].copy_from_slice(&1u16.to_le_bytes()); // PCM
177    h[22..24].copy_from_slice(&2u16.to_le_bytes()); // stereo
178    h[24..28].copy_from_slice(&rate.to_le_bytes());
179    h[28..32].copy_from_slice(&(rate * 4).to_le_bytes()); // byte rate
180    h[32..34].copy_from_slice(&4u16.to_le_bytes()); // block align
181    h[34..36].copy_from_slice(&16u16.to_le_bytes()); // bits
182    h[36..40].copy_from_slice(b"data");
183    h[40..44].copy_from_slice(&data.to_le_bytes());
184    h
185}
186
187#[cfg(test)]
188mod tests {
189    use super::*;
190
191    fn ms(n: u64) -> Duration {
192        Duration::from_millis(n)
193    }
194
195    fn read(path: &Path) -> (u32, Vec<(i16, i16)>) {
196        let bytes = std::fs::read(path).unwrap();
197        assert_eq!(&bytes[0..4], b"RIFF");
198        assert_eq!(u16::from_le_bytes([bytes[22], bytes[23]]), 2);
199        let rate = u32::from_le_bytes(bytes[24..28].try_into().unwrap());
200        let len = u32::from_le_bytes(bytes[40..44].try_into().unwrap()) as usize;
201        assert_eq!(bytes.len(), 44 + len);
202        let frames = bytes[44..]
203            .chunks_exact(4)
204            .map(|f| {
205                (
206                    i16::from_le_bytes([f[0], f[1]]),
207                    i16::from_le_bytes([f[2], f[3]]),
208                )
209            })
210            .collect();
211        (rate, frames)
212    }
213
214    #[test]
215    fn caller_and_agent_are_aligned_on_the_call_clock() {
216        let path = std::env::temp_dir().join(format!("rec-{}.wav", std::process::id()));
217        let rec = CallRecorder::create(&path, 8_000).unwrap();
218        // 20 ms of caller audio arrives at 20 ms, then nothing until 60 ms.
219        rec.caller_at(ms(20), &[1; 160]);
220        // The agent streams 100 ms of audio at once at 30 ms: it plays
221        // from 30 ms to 130 ms.
222        rec.agent_at(ms(30), &[2; 800]);
223        rec.caller_at(ms(60), &[3; 160]);
224        // The caller barges in at 80 ms: 50 ms of agent audio never played.
225        let cut = rec.flush_at(ms(80));
226        assert_eq!(cut, ms(50));
227        rec.caller_at(ms(100), &[4; 160]);
228        rec.finish().unwrap();
229
230        let (rate, frames) = read(&path);
231        assert_eq!(rate, 8_000);
232        assert_eq!(frames.len(), 800); // 100 ms
233        let at = |t_ms: usize| frames[t_ms * 8];
234        assert_eq!(at(10), (1, 0));
235        assert_eq!(at(30), (0, 2)); // delivery gap on the left, agent on the right
236        assert_eq!(at(50), (3, 2));
237        assert_eq!(at(85), (4, 0)); // agent cut at 80 ms
238        let _ = std::fs::remove_file(path);
239    }
240}
gemini_adk_fluent_rs/telephony/recorder.rs

gemini_adk_fluent_rs/telephony/
recorder.rs
