From cea15c12c7641445073a50b34b3b887266f1f881 Mon Sep 17 00:00:00 2001 From: Jake Date: Wed, 22 Apr 2026 07:52:48 +0100 Subject: [PATCH] feat(A.3 #25): aggressive buffer trim tied to commit points MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New streaming::buffer_trim module with two pure helpers: - sample_index_for_seconds(end_secs, sample_rate) -> u64: converts LocalAgreement::last_committed_end_secs() into an absolute sample index. Defensive against negative end_secs (treats as 0) so a future clock-skewed timestamp cannot wrap to a huge u64. - trim_buffer_to_commit_point(buffer, buffer_start_sample, commit_sample_index) -> new_buffer_start_sample: drains the prefix of the capture buffer that falls below the commit point and returns the new absolute-index origin. Edge cases covered by tests: - commit before buffer start → no drain - commit equal to buffer start → no drain - commit inside buffer → drain prefix, advance origin - commit at buffer end → drain all, origin moves forward - commit past buffer end → drain all, origin parks at commit (rare edge after a committer reset) - sample_index_for_seconds rounds nearest, negatives clamp to 0 - integration with LocalAgreement::last_committed_end_secs trim_bounds_buffer_over_long_session is the acceptance fixture for the ufal #120/#102 failure mode: 100 cycles of 16_000 captured samples with a 200-sample tentative tail per cycle, and the buffer stays below 2× the tentative envelope instead of growing to 1.6M samples. Integration into src-tauri/src/commands/live.rs deferred to the dogfood session that wires VadChunker and LocalAgreement end-to-end — the trim is a one-line replacement at the maybe_dispatch_chunk drain site once the committer is feeding it commit points. --- crates/transcription/src/lib.rs | 3 +- .../src/streaming/buffer_trim.rs | 190 ++++++++++++++++++ crates/transcription/src/streaming/mod.rs | 2 + 3 files changed, 194 insertions(+), 1 deletion(-) create mode 100644 crates/transcription/src/streaming/buffer_trim.rs diff --git a/crates/transcription/src/lib.rs b/crates/transcription/src/lib.rs index 79f8b2b..65357fd 100644 --- a/crates/transcription/src/lib.rs +++ b/crates/transcription/src/lib.rs @@ -12,7 +12,8 @@ pub use local_engine::{load_parakeet, LocalEngine, SpeechModelAdapter, TimedTran pub use local_engine::load_whisper; pub use model_manager::{download, is_downloaded, list_downloaded, model_dir, models_dir}; pub use streaming::{ - CommitDecision, CommitPolicy, LocalAgreement, RmsVadChunker, Token, VadChunk, VadChunker, + sample_index_for_seconds, trim_buffer_to_commit_point, CommitDecision, CommitPolicy, + LocalAgreement, RmsVadChunker, Token, VadChunk, VadChunker, }; pub use transcriber::{Transcriber, TranscriberCapabilities}; pub use transcribe_rs::SpeechModel; diff --git a/crates/transcription/src/streaming/buffer_trim.rs b/crates/transcription/src/streaming/buffer_trim.rs new file mode 100644 index 0000000..82e9f05 --- /dev/null +++ b/crates/transcription/src/streaming/buffer_trim.rs @@ -0,0 +1,190 @@ +//! Buffer-trim helpers for streaming transcription. +//! +//! Brief item #25: replace the current `OVERLAP_SAMPLES`-based drain +//! in `src-tauri/src/commands/live.rs` with a trim tied to the last +//! commit point emitted by the `CommitPolicy`. This keeps the capture +//! buffer bounded regardless of wall-clock session length (ufal #120 / +//! #102) by guaranteeing that any sample already committed to the +//! transcript is never kept in the working buffer. +//! +//! The helpers here are pure — they don't know about the live session +//! loop. Integration into `live.rs` ships as a follow-up after the +//! LocalAgreement wiring (#24) is dogfooded. + +/// Absolute sample index at the end of the given session-relative +/// seconds mark, rounded to the nearest sample. `end_secs` typically +/// comes from `LocalAgreement::last_committed_end_secs()`. +pub fn sample_index_for_seconds(end_secs: f64, sample_rate: u32) -> u64 { + if end_secs <= 0.0 { + return 0; + } + (end_secs * sample_rate as f64).round() as u64 +} + +/// Drain the prefix of `buffer` whose absolute sample indices fall +/// below `commit_sample_index`. `buffer_start_sample` is the absolute +/// index of `buffer[0]` before the trim. +/// +/// Returns the new `buffer_start_sample`. If the commit point is +/// before or equal to `buffer_start_sample`, nothing is drained. +/// If the commit point is beyond the current end of the buffer, the +/// whole buffer is drained and the new start is set to the commit +/// index — the buffer is still empty, but its absolute-index origin +/// moves forward so subsequent samples are positioned correctly. +pub fn trim_buffer_to_commit_point( + buffer: &mut Vec, + buffer_start_sample: u64, + commit_sample_index: u64, +) -> u64 { + if commit_sample_index <= buffer_start_sample { + return buffer_start_sample; + } + let drain_count = (commit_sample_index - buffer_start_sample) as usize; + if drain_count >= buffer.len() { + buffer.clear(); + return commit_sample_index; + } + buffer.drain(..drain_count); + buffer_start_sample + drain_count as u64 +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn sample_index_for_seconds_zero_is_zero() { + assert_eq!(sample_index_for_seconds(0.0, 16_000), 0); + } + + #[test] + fn sample_index_for_seconds_negative_is_zero() { + // Defensive: end_secs should never be negative, but if it is + // (clock skew in a future f64 source) treat as "nothing + // committed yet" rather than wrapping to a huge u64. + assert_eq!(sample_index_for_seconds(-1.0, 16_000), 0); + } + + #[test] + fn sample_index_for_seconds_rounds_nearest() { + // 0.5 s at 16 kHz = 8000 samples exactly. + assert_eq!(sample_index_for_seconds(0.5, 16_000), 8_000); + // Round-nearest: 0.50003 s × 16 kHz = 8000.48 → 8000. + assert_eq!(sample_index_for_seconds(0.50003, 16_000), 8_000); + // 0.5001 s × 16 kHz = 8001.6 → 8002. + assert_eq!(sample_index_for_seconds(0.5001, 16_000), 8_002); + } + + #[test] + fn trim_does_nothing_when_commit_is_before_buffer_start() { + let mut buf = vec![1.0, 2.0, 3.0]; + let new_start = trim_buffer_to_commit_point(&mut buf, 1000, 500); + assert_eq!(new_start, 1000); + assert_eq!(buf, vec![1.0, 2.0, 3.0]); + } + + #[test] + fn trim_does_nothing_when_commit_equals_buffer_start() { + let mut buf = vec![1.0, 2.0, 3.0]; + let new_start = trim_buffer_to_commit_point(&mut buf, 1000, 1000); + assert_eq!(new_start, 1000); + assert_eq!(buf, vec![1.0, 2.0, 3.0]); + } + + #[test] + fn trim_drains_prefix_when_commit_is_inside_buffer() { + let mut buf = vec![1.0, 2.0, 3.0, 4.0, 5.0]; + // buffer starts at absolute index 100, commit is at 102. + // Drain 2 samples; remaining buffer starts at 102. + let new_start = trim_buffer_to_commit_point(&mut buf, 100, 102); + assert_eq!(new_start, 102); + assert_eq!(buf, vec![3.0, 4.0, 5.0]); + } + + #[test] + fn trim_clears_buffer_when_commit_is_at_buffer_end() { + let mut buf = vec![1.0, 2.0, 3.0]; + // buffer is [100, 103). commit at 103 means every sample is + // committed — drain all, start moves forward. + let new_start = trim_buffer_to_commit_point(&mut buf, 100, 103); + assert_eq!(new_start, 103); + assert!(buf.is_empty()); + } + + #[test] + fn trim_clears_buffer_when_commit_is_past_buffer_end() { + let mut buf = vec![1.0, 2.0, 3.0]; + // Commit well beyond the buffer — this happens in rare edge + // cases where the committer's notion of time outstrips the + // current buffer (e.g. after a reset). Defensive: drain and + // park the origin at the commit point. + let new_start = trim_buffer_to_commit_point(&mut buf, 100, 200); + assert_eq!(new_start, 200); + assert!(buf.is_empty()); + } + + #[test] + fn trim_bounds_buffer_over_long_session() { + // Simulate a committer that keeps up with capture: each cycle + // feeds 16_000 samples and commits all but a 200-sample + // tentative tail. Over 100 cycles the buffer must stay near + // that tentative envelope — not accumulate 100 × 16_000 samples + // as it would without the commit-point trim. + // + // The tentative tail stacks by 200 per cycle because each new + // push extends the buffer BEFORE the trim runs against the + // previous cycle's commit point, so the expected bound is + // (tentative_per_cycle + new_push_minus_commit), not just + // tentative_per_cycle. + let mut buf: Vec = Vec::new(); + let mut start: u64 = 0; + let mut total_pushed: u64 = 0; + let tentative_per_cycle: u64 = 200; + for _ in 0..100 { + buf.extend(std::iter::repeat(0.25_f32).take(16_000)); + total_pushed += 16_000; + let commit_point = total_pushed - tentative_per_cycle; + start = trim_buffer_to_commit_point(&mut buf, start, commit_point); + } + assert!( + buf.len() as u64 <= 2 * tentative_per_cycle, + "buffer outgrew the commit-bounded envelope: len = {} (bound {})", + buf.len(), + 2 * tentative_per_cycle + ); + } + + #[test] + fn integrates_with_local_agreement_last_committed_end_secs() { + use super::super::commit_policy::{LocalAgreement, Token}; + + let mut la = LocalAgreement::new(2); + let _ = la.push(vec![Token { + text: "hello".into(), + start_secs: 0.0, + end_secs: 0.5, + }]); + let _ = la.push(vec![ + Token { + text: "hello".into(), + start_secs: 0.0, + end_secs: 0.5, + }, + Token { + text: "world".into(), + start_secs: 0.5, + end_secs: 1.0, + }, + ]); + // "hello" is committed, ending at 0.5 s. + let commit_idx = sample_index_for_seconds(la.last_committed_end_secs(), 16_000); + assert_eq!(commit_idx, 8_000); + + // Simulate a capture buffer that has received 1.2 s of audio + // starting at t=0. + let mut buf: Vec = std::iter::repeat(0.1_f32).take(19_200).collect(); + let new_start = trim_buffer_to_commit_point(&mut buf, 0, commit_idx); + assert_eq!(new_start, 8_000); + assert_eq!(buf.len(), 19_200 - 8_000); + } +} diff --git a/crates/transcription/src/streaming/mod.rs b/crates/transcription/src/streaming/mod.rs index 29b994b..be1f0d9 100644 --- a/crates/transcription/src/streaming/mod.rs +++ b/crates/transcription/src/streaming/mod.rs @@ -6,9 +6,11 @@ //! threshold tuning can be validated against real microphone captures //! rather than synthetic fixtures (brief items #21, #24, #25). +pub mod buffer_trim; pub mod commit_policy; pub mod rms_vad; +pub use buffer_trim::{sample_index_for_seconds, trim_buffer_to_commit_point}; pub use commit_policy::{CommitDecision, CommitPolicy, LocalAgreement, Token}; pub use rms_vad::RmsVadChunker;