diff --git a/crates/transcription/src/lib.rs b/crates/transcription/src/lib.rs index cf9ad10..79f8b2b 100644 --- a/crates/transcription/src/lib.rs +++ b/crates/transcription/src/lib.rs @@ -11,6 +11,8 @@ pub use local_engine::{load_parakeet, LocalEngine, SpeechModelAdapter, TimedTran #[cfg(feature = "whisper")] pub use local_engine::load_whisper; pub use model_manager::{download, is_downloaded, list_downloaded, model_dir, models_dir}; -pub use streaming::{RmsVadChunker, VadChunk, VadChunker}; +pub use streaming::{ + CommitDecision, CommitPolicy, LocalAgreement, RmsVadChunker, Token, VadChunk, VadChunker, +}; pub use transcriber::{Transcriber, TranscriberCapabilities}; pub use transcribe_rs::SpeechModel; diff --git a/crates/transcription/src/streaming/commit_policy.rs b/crates/transcription/src/streaming/commit_policy.rs new file mode 100644 index 0000000..9277f58 --- /dev/null +++ b/crates/transcription/src/streaming/commit_policy.rs @@ -0,0 +1,361 @@ +//! LocalAgreement-n commit policy for streaming transcription. +//! +//! Source: ufal/whisper_streaming. Tokens emitted by a streaming ASR +//! pipeline are held as tentative until `n` consecutive passes produce +//! the same prefix. Only the agreed prefix is "committed" — the rest +//! is a tentative tail the UI renders differently (dashed underline +//! per brief item #24, workstream-B contract). +//! +//! This module ships the committer plus a Token type carrying +//! timestamps so brief item #25 (aggressive buffer trim tied to commit +//! points) can compute the absolute sample index of the last +//! committed token and drain the capture buffer up to that point. +//! +//! Integration into `src-tauri/src/commands/live.rs` lands in a +//! separate commit so the tentative/committed partition can be +//! validated against real streaming captures. + +use std::collections::VecDeque; + +/// A single token (word or sub-segment) emitted by the ASR pipeline. +/// +/// Equality on `Token` is text-only — the committer matches tokens +/// across passes by their spelling, since timestamps drift slightly +/// between overlapping Whisper windows. Start and end seconds are +/// absolute (session-relative) so #25 can translate them to sample +/// indices. +#[derive(Debug, Clone)] +pub struct Token { + pub text: String, + pub start_secs: f64, + pub end_secs: f64, +} + +impl PartialEq for Token { + fn eq(&self, other: &Self) -> bool { + self.text == other.text + } +} + +impl Eq for Token {} + +/// Outcome of pushing a new pass through the committer. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct CommitDecision { + /// Tokens newly committed by this pass. Empty if no new agreement + /// was reached. Append to the frontend's committed list. + pub newly_committed: Vec, + /// Tentative tail — tokens past the agreement prefix in the most + /// recent pass. Replaces (not appends to) any previous tentative. + pub tentative: Vec, +} + +/// Commit policy selector. Keeping this as an enum leaves room for +/// future policies (AlignAtt, length-capped, etc.) without a breaking +/// API change. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CommitPolicy { + /// LocalAgreement-n: `n` consecutive passes must produce the same + /// prefix before emission. `n = 2` is the ufal default. + LocalAgreement { n: usize }, +} + +impl Default for CommitPolicy { + fn default() -> Self { + CommitPolicy::LocalAgreement { n: 2 } + } +} + +/// Stateful LocalAgreement-n committer. +/// +/// Invariants: +/// - `history` holds at most `n` most-recent passes. +/// - `committed_count` counts tokens committed so far; these are +/// always a prefix of every pass in `history`. +/// - `last_committed_end_secs` is 0 when nothing is committed, +/// otherwise the `end_secs` of the most recent committed token. +pub struct LocalAgreement { + n: usize, + history: VecDeque>, + committed_count: usize, + last_committed_end_secs: f64, +} + +impl LocalAgreement { + pub fn new(n: usize) -> Self { + assert!(n >= 1, "LocalAgreement-n requires n >= 1"); + Self { + n, + history: VecDeque::with_capacity(n), + committed_count: 0, + last_committed_end_secs: 0.0, + } + } + + pub fn from_policy(policy: CommitPolicy) -> Self { + match policy { + CommitPolicy::LocalAgreement { n } => Self::new(n), + } + } + + /// Feed the next pass of transcribed tokens. Returns newly + /// committed tokens and the current tentative tail. + pub fn push(&mut self, pass: Vec) -> CommitDecision { + self.history.push_back(pass); + while self.history.len() > self.n { + self.history.pop_front(); + } + + // Can't commit anything until we have n passes in hand. + if self.history.len() < self.n { + let tentative = self + .history + .back() + .cloned() + .unwrap_or_default(); + return CommitDecision { + newly_committed: Vec::new(), + tentative, + }; + } + + let lcp_len = longest_common_prefix_len(&self.history); + + // The agreed prefix can only grow — never shrink below what we + // already committed. ufal's invariant: once committed, stay + // committed. + let new_lcp_len = lcp_len.max(self.committed_count); + + let latest = self.history.back().expect("history is non-empty here"); + let newly_committed = if new_lcp_len > self.committed_count { + latest[self.committed_count..new_lcp_len].to_vec() + } else { + Vec::new() + }; + + if let Some(last) = newly_committed.last() { + self.last_committed_end_secs = last.end_secs; + } + self.committed_count = new_lcp_len; + + let tentative = latest[new_lcp_len..].to_vec(); + + CommitDecision { + newly_committed, + tentative, + } + } + + /// End-of-stream: commit anything still tentative in the latest + /// pass and return it. Callers do this when the session closes so + /// the final utterance reaches the transcript. + pub fn flush(&mut self) -> Vec { + let Some(latest) = self.history.back().cloned() else { + return Vec::new(); + }; + if latest.len() <= self.committed_count { + return Vec::new(); + } + let flushed = latest[self.committed_count..].to_vec(); + if let Some(last) = flushed.last() { + self.last_committed_end_secs = last.end_secs; + } + self.committed_count = latest.len(); + flushed + } + + /// Absolute (session-relative) seconds at the end of the most + /// recently committed token. `0.0` when nothing has committed yet. + /// Brief item #25 will multiply this by the capture sample rate to + /// get the buffer-drain target. + pub fn last_committed_end_secs(&self) -> f64 { + self.last_committed_end_secs + } + + /// Drop all state — used after a repetition-detector context + /// reset (#26) so the committer doesn't carry stale history + /// across the reset boundary. + pub fn reset(&mut self) { + self.history.clear(); + self.committed_count = 0; + self.last_committed_end_secs = 0.0; + } +} + +fn longest_common_prefix_len(passes: &VecDeque>) -> usize { + let Some(first) = passes.front() else { + return 0; + }; + let shortest = passes.iter().map(|p| p.len()).min().unwrap_or(0); + for i in 0..shortest { + let candidate = &first[i]; + for pass in passes.iter().skip(1) { + if pass[i] != *candidate { + return i; + } + } + } + shortest +} + +#[cfg(test)] +mod tests { + use super::*; + + fn tok(text: &str, start: f64, end: f64) -> Token { + Token { + text: text.into(), + start_secs: start, + end_secs: end, + } + } + + #[test] + fn first_pass_is_all_tentative() { + let mut la = LocalAgreement::new(2); + let decision = la.push(vec![tok("hello", 0.0, 0.5), tok("world", 0.5, 1.0)]); + assert!(decision.newly_committed.is_empty()); + assert_eq!(decision.tentative.len(), 2); + assert_eq!(la.last_committed_end_secs(), 0.0); + } + + #[test] + fn two_matching_passes_commit_common_prefix() { + let mut la = LocalAgreement::new(2); + let _ = la.push(vec![tok("the", 0.0, 0.3), tok("cat", 0.3, 0.6)]); + let decision = la.push(vec![ + tok("the", 0.0, 0.3), + tok("cat", 0.3, 0.6), + tok("sat", 0.6, 0.9), + ]); + assert_eq!(decision.newly_committed.len(), 2); + assert_eq!(decision.newly_committed[0].text, "the"); + assert_eq!(decision.newly_committed[1].text, "cat"); + assert_eq!(decision.tentative.len(), 1); + assert_eq!(decision.tentative[0].text, "sat"); + assert!((la.last_committed_end_secs() - 0.6).abs() < f64::EPSILON); + } + + #[test] + fn divergent_second_pass_commits_nothing() { + let mut la = LocalAgreement::new(2); + let _ = la.push(vec![tok("hello", 0.0, 0.5)]); + let decision = la.push(vec![tok("yellow", 0.0, 0.5)]); + assert!( + decision.newly_committed.is_empty(), + "no common prefix — must not commit" + ); + assert_eq!(decision.tentative.len(), 1); + assert_eq!(decision.tentative[0].text, "yellow"); + } + + #[test] + fn extending_agreement_commits_newly_agreed_tokens() { + let mut la = LocalAgreement::new(2); + let _ = la.push(vec![tok("a", 0.0, 0.1), tok("b", 0.1, 0.2)]); + let _ = la.push(vec![ + tok("a", 0.0, 0.1), + tok("b", 0.1, 0.2), + tok("c", 0.2, 0.3), + ]); + // Now history has [[a,b], [a,b,c]], committed = 2 (a, b). + let decision = la.push(vec![ + tok("a", 0.0, 0.1), + tok("b", 0.1, 0.2), + tok("c", 0.2, 0.3), + tok("d", 0.3, 0.4), + ]); + assert_eq!(decision.newly_committed.len(), 1, "c becomes committed"); + assert_eq!(decision.newly_committed[0].text, "c"); + assert_eq!(decision.tentative.len(), 1); + assert_eq!(decision.tentative[0].text, "d"); + } + + #[test] + fn tentative_tail_tracks_latest_pass_only() { + let mut la = LocalAgreement::new(2); + let _ = la.push(vec![tok("x", 0.0, 0.1)]); + let _ = la.push(vec![tok("x", 0.0, 0.1), tok("y_guess", 0.1, 0.2)]); + // x is committed, tail is y_guess. + let decision = la.push(vec![tok("x", 0.0, 0.1), tok("y_real", 0.1, 0.2)]); + assert!(decision.newly_committed.is_empty()); + assert_eq!(decision.tentative.len(), 1); + assert_eq!( + decision.tentative[0].text, "y_real", + "tentative must reflect the latest pass, not carry stale y_guess" + ); + } + + #[test] + fn committed_prefix_never_shrinks() { + // Even if a later pass contradicts an earlier commit, the + // committed prefix stays frozen. This is ufal's invariant. + let mut la = LocalAgreement::new(2); + let _ = la.push(vec![tok("foo", 0.0, 0.3)]); + let _ = la.push(vec![tok("foo", 0.0, 0.3), tok("bar", 0.3, 0.6)]); + // "foo" is committed. + assert_eq!(la.committed_count, 1); + + let decision = la.push(vec![tok("fop", 0.0, 0.3), tok("baz", 0.3, 0.6)]); + // LCP with previous pass [foo, bar] is 0 — but we already + // committed "foo", so committed_count stays at 1. + assert_eq!(la.committed_count, 1); + assert!(decision.newly_committed.is_empty()); + } + + #[test] + fn flush_emits_remaining_tentative() { + let mut la = LocalAgreement::new(2); + let _ = la.push(vec![tok("a", 0.0, 0.1), tok("b", 0.1, 0.2)]); + let _ = la.push(vec![ + tok("a", 0.0, 0.1), + tok("b", 0.1, 0.2), + tok("c", 0.2, 0.3), + ]); + // Committed: a, b. Tentative: c. + let flushed = la.flush(); + assert_eq!(flushed.len(), 1); + assert_eq!(flushed[0].text, "c"); + assert!((la.last_committed_end_secs() - 0.3).abs() < f64::EPSILON); + } + + #[test] + fn flush_with_no_history_is_empty() { + let mut la = LocalAgreement::new(2); + assert!(la.flush().is_empty()); + } + + #[test] + fn reset_clears_commit_state() { + let mut la = LocalAgreement::new(2); + let _ = la.push(vec![tok("a", 0.0, 0.1)]); + let _ = la.push(vec![tok("a", 0.0, 0.1), tok("b", 0.1, 0.2)]); + la.reset(); + assert_eq!(la.committed_count, 0); + assert_eq!(la.last_committed_end_secs(), 0.0); + let decision = la.push(vec![tok("z", 0.0, 0.1)]); + assert!(decision.newly_committed.is_empty()); + assert_eq!(decision.tentative[0].text, "z"); + } + + #[test] + fn n_three_requires_three_matching_passes_to_commit() { + let mut la = LocalAgreement::new(3); + let _ = la.push(vec![tok("x", 0.0, 0.1)]); + let _ = la.push(vec![tok("x", 0.0, 0.1)]); + // Only 2 passes so far; with n=3 no commit yet. + let decision = la.push(vec![tok("x", 0.0, 0.1), tok("y", 0.1, 0.2)]); + assert_eq!( + decision.newly_committed.len(), + 1, + "on the 3rd matching pass, x becomes committed" + ); + assert_eq!(decision.newly_committed[0].text, "x"); + } + + #[test] + fn from_policy_default_is_local_agreement_n2() { + let la = LocalAgreement::from_policy(CommitPolicy::default()); + assert_eq!(la.n, 2); + } +} diff --git a/crates/transcription/src/streaming/mod.rs b/crates/transcription/src/streaming/mod.rs index 2bd7030..29b994b 100644 --- a/crates/transcription/src/streaming/mod.rs +++ b/crates/transcription/src/streaming/mod.rs @@ -6,8 +6,10 @@ //! threshold tuning can be validated against real microphone captures //! rather than synthetic fixtures (brief items #21, #24, #25). +pub mod commit_policy; pub mod rms_vad; +pub use commit_policy::{CommitDecision, CommitPolicy, LocalAgreement, Token}; pub use rms_vad::RmsVadChunker; /// A span of audio the VAD considers worth transcribing. `start_sample`