feat(transcription): streaming growing-window + interim + commit-on-pause (FR-TRX-2)

src-tauri/src/transcription/mod.rs

@@ -317,63 +317,215 @@ pub mod onnx_models;
 #[cfg(feature = "npu")]
 pub use npu::OnnxTranscriber;
 
-/// Streaming window worker (Phase 1, T1.5/T1.6): accumulates raw 16kHz-mono
-/// chunks from the `audio` service into fixed-size, **non-overlapping** windows
-/// and runs one `transcribe_stream` pass per window as it fills, forwarding
-/// each produced segment to `on_segment` (e.g. a Tauri event emit).
+const STREAM_SAMPLE_RATE: usize = 16_000;
+
+/// Cadence/length knobs for the live streaming worker (`Streamer`). Tuned for a
+/// fluid transcript that shows words ~1s after they're spoken and breaks lines
+/// at natural pauses rather than on a fixed clock.
+#[derive(Debug, Clone, Copy)]
+pub struct StreamTuning {
+    /// How often the growing window is re-decoded (and the interim line
+    /// refreshed). Lower = more responsive, but more CPU (each decode re-runs
+    /// whisper over the whole in-progress window).
+    pub step_ms: u64,
+    /// Hard cap on an uncommitted window: once the current line reaches this
+    /// without a natural pause, it's force-committed so the window (and its
+    /// per-step decode cost) can't grow without bound.
+    pub max_window_ms: u64,
+    /// Don't commit a line shorter than this on a detected pause — avoids
+    /// chopping a brief hesitation into its own one-word line.
+    pub min_commit_ms: u64,
+}
+
+impl StreamTuning {
+    /// `low_overhead` doubles the decode step (halving CPU) at the cost of a
+    /// slightly less immediate transcript — matches the Settings "low overhead"
+    /// preset (CPU + smallest model, for battery/background use).
+    pub fn new(low_overhead: bool) -> Self {
+        Self {
+            step_ms: if low_overhead { 2000 } else { 1000 },
+            max_window_ms: 12_000,
+            min_commit_ms: if low_overhead { 2000 } else { 1500 },
+        }
+    }
+}
+
+impl Default for StreamTuning {
+    fn default() -> Self {
+        Self::new(false)
+    }
+}
+
+/// Incremental live-transcript state machine, independent of any transcription
+/// engine (a `decode: &[f32] -> String` closure is injected) so its
+/// commit/interim logic is unit-testable without whisper.
 ///
-/// `transcribe_stream` takes a `SegmentSink` per the `Transcriber` trait (so a
-/// future async/threaded engine can push mid-inference), but whisper.cpp's
-/// `full()` call is synchronous — by the time a window's `transcribe_stream`
-/// call returns, every segment it produced is already sitting in a fresh
-/// per-window channel, so this drains it inline rather than needing a second
-/// long-lived thread just to bridge segments out.
+/// The model: keep one growing "uncommitted" window of audio. Every `step_ms`,
+/// re-decode the whole window and emit it as an **interim** segment with a
+/// stable id — so the current line grows in place (fluid, low-latency) instead
+/// of popping in whole every few seconds. When the decoded text stops changing
+/// for a step (the speaker paused, so the extra audio was silence) the line is
+/// **committed** (`interim = false`, same id) and a fresh window/line begins —
+/// so lines break at natural sentence pauses, not on a fixed 4s clock. A
+/// `max_window_ms` backstop force-commits a pause-free monologue.
 ///
-/// True incremental/partial-word streaming (and window overlap for continuity)
-/// are out of scope for Phase 1: whisper.cpp transcribes each window from
-/// scratch, so an overlapping window would re-emit the overlapped words a
-/// second time with no stitching logic to merge them — a worse rough edge for
-/// a live transcript than the occasional word clipped at a window boundary.
-/// "Near real time" (FR-TRX-2) is met by short (~4s) windows; `interim` stays
-/// `false` for every segment produced here.
+/// whisper.cpp isn't a true streaming recognizer (it re-decodes from scratch),
+/// so this trades CPU — the growing window is re-decoded every step — for a
+/// natural-looking transcript. `audio_ctx_for_window` keeps each decode's cost
+/// proportional to the window length rather than a full 30s encode, and
+/// committing on pauses keeps the window short for conversational speech.
+pub struct Streamer {
+    step_len: usize,
+    max_len: usize,
+    min_commit_len: usize,
+    window: Vec<f32>,
+    committed_offset_ms: u64,
+    since_last_decode: usize,
+    next_id: u64,
+    last_text: String,
+}
+
+impl Streamer {
+    pub fn new(tuning: StreamTuning) -> Self {
+        let per_ms = STREAM_SAMPLE_RATE / 1000;
+        Self {
+            step_len: tuning.step_ms as usize * per_ms,
+            max_len: tuning.max_window_ms as usize * per_ms,
+            min_commit_len: tuning.min_commit_ms as usize * per_ms,
+            window: Vec::new(),
+            committed_offset_ms: 0,
+            since_last_decode: 0,
+            next_id: 0,
+            last_text: String::new(),
+        }
+    }
+
+    /// Append newly captured audio; decode + emit once a step's worth has
+    /// accumulated. `decode(samples, offset_ms)` returns the transcript of the
+    /// window so far (empty for silence).
+    pub fn feed<D, F>(&mut self, chunk: &[f32], decode: &D, on_segment: &mut F)
+    where
+        D: Fn(&[f32], u64) -> String,
+        F: FnMut(TranscriptSegment),
+    {
+        self.window.extend_from_slice(chunk);
+        self.since_last_decode += chunk.len();
+        if self.since_last_decode >= self.step_len {
+            self.since_last_decode = 0;
+            self.tick(false, decode, on_segment);
+        }
+    }
+
+    /// Commit whatever's in flight — called once when capture stops so the last
+    /// in-progress line is finalized rather than left interim.
+    pub fn flush<D, F>(&mut self, decode: &D, on_segment: &mut F)
+    where
+        D: Fn(&[f32], u64) -> String,
+        F: FnMut(TranscriptSegment),
+    {
+        self.tick(true, decode, on_segment);
+    }
+
+    fn tick<D, F>(&mut self, force: bool, decode: &D, on_segment: &mut F)
+    where
+        D: Fn(&[f32], u64) -> String,
+        F: FnMut(TranscriptSegment),
+    {
+        if self.window.is_empty() {
+            return;
+        }
+        let window_ms = (self.window.len() as u64 * 1000) / STREAM_SAMPLE_RATE as u64;
+        let text = decode(&self.window, self.committed_offset_ms);
+        let over_max = self.window.len() >= self.max_len;
+
+        // Nothing recognized yet (leading/standalone silence): don't show an
+        // empty line, but still drop the buffer once it's grown too big so we
+        // aren't re-decoding a long silence every step.
+        if text.is_empty() {
+            if over_max || force {
+                self.reset(window_ms, false);
+            }
+            return;
+        }
+
+        let stable = text == self.last_text;
+        let commit = force || over_max || (stable && self.window.len() >= self.min_commit_len);
+        on_segment(TranscriptSegment {
+            id: self.next_id,
+            start_ms: self.committed_offset_ms,
+            end_ms: self.committed_offset_ms + window_ms,
+            // Provisional speaker; the post-stop diarization pass reassigns.
+            speaker: "S1".to_string(),
+            text: text.clone(),
+            confidence: None,
+            interim: !commit,
+        });
+        if commit {
+            self.reset(window_ms, true);
+        } else {
+            self.last_text = text;
+        }
+    }
+
+    /// Start a fresh window/line after a commit (`new_line`) or after dropping
+    /// leading silence (`!new_line`, which reuses the id since nothing was
+    /// emitted for it).
+    fn reset(&mut self, window_ms: u64, new_line: bool) {
+        self.committed_offset_ms += window_ms;
+        self.window.clear();
+        self.last_text.clear();
+        self.since_last_decode = 0;
+        if new_line {
+            self.next_id += 1;
+        }
+    }
+}
+
+/// Live streaming worker (FR-TRX-2): drives a `Streamer` from the `audio`
+/// service's 16kHz-mono frames, decoding each growing window with `transcriber`
+/// and forwarding interim-then-final segments to `on_segment` (a Tauri emit).
+///
+/// `transcribe_stream` is synchronous for whisper.cpp — by the time it returns,
+/// every segment it produced is already in the per-call channel — so this
+/// drains it inline and joins the text into the window's transcript for the
+/// `Streamer` to diff/commit.
 #[cfg(feature = "cpu-transcription")]
-pub fn run_streaming_worker<T, F>(transcriber: &T, frame_rx: Receiver<Vec<f32>>, mut on_segment: F)
-where
+pub fn run_streaming_worker<T, F>(
+    transcriber: &T,
+    frame_rx: Receiver<Vec<f32>>,
+    tuning: StreamTuning,
+    mut on_segment: F,
+) where
     // `?Sized` so the dispatcher can hand us a `&dyn Transcriber` (whisper.cpp
     // or the NPU engine, chosen at runtime) rather than a concrete type.
     T: Transcriber + ?Sized,
     F: FnMut(TranscriptSegment),
 {
-    const SAMPLE_RATE: usize = 16_000;
-    const WINDOW_SECS: f32 = 4.0;
-    let window_len = (WINDOW_SECS * SAMPLE_RATE as f32) as usize;
-
-    let mut buf: Vec<f32> = Vec::new();
-    let mut offset_ms: u64 = 0;
-
-    let run_window = |transcriber: &T, samples: Vec<f32>, offset_ms: u64, on_segment: &mut F| {
+    let decode = |samples: &[f32], offset_ms: u64| -> String {
         let (tx, rx) = std::sync::mpsc::channel();
-        let window = AudioWindow { samples, offset_ms };
+        let window = AudioWindow {
+            samples: samples.to_vec(),
+            offset_ms,
+        };
         if let Err(e) = transcriber.transcribe_stream(window, tx) {
             tracing::warn!("transcription window failed: {e}");
+            return String::new();
         }
+        let mut parts = Vec::new();
         while let Ok(segment) = rx.try_recv() {
-            on_segment(segment);
+            let t = segment.text.trim();
+            if !t.is_empty() {
+                parts.push(t.to_string());
+            }
         }
+        parts.join(" ")
     };
 
+    let mut streamer = Streamer::new(tuning);
     while let Ok(chunk) = frame_rx.recv() {
-        buf.extend_from_slice(&chunk);
-        while buf.len() >= window_len {
-            let samples: Vec<f32> = buf.drain(..window_len).collect();
-            run_window(transcriber, samples, offset_ms, &mut on_segment);
-            offset_ms += (window_len as u64 * 1000) / SAMPLE_RATE as u64;
-        }
-    }
-    // Final partial window on stop, if there's enough audio to be worth a pass.
-    if buf.len() > SAMPLE_RATE / 2 {
-        run_window(transcriber, buf, offset_ms, &mut on_segment);
+        streamer.feed(&chunk, &decode, &mut on_segment);
     }
+    streamer.flush(&decode, &mut on_segment);
 }
 
 #[cfg(test)]
@@ -381,6 +533,97 @@ where
 mod tests {
     use super::*;
 
+    #[test]
+    fn streamer_grows_interim_then_commits_on_a_pause() {
+        // Text grows for two steps, then repeats (the speaker paused, so the
+        // extra second was silence) → the line commits.
+        let tuning = StreamTuning {
+            step_ms: 1000,
+            max_window_ms: 12_000,
+            min_commit_ms: 1500,
+        };
+        let mut s = Streamer::new(tuning);
+        let script = std::cell::Cell::new(0usize);
+        let texts = ["hello", "hello world", "hello world"];
+        let decode = |_: &[f32], _off: u64| -> String {
+            let i = script.get();
+            script.set(i + 1);
+            texts.get(i).copied().unwrap_or("hello world").to_string()
+        };
+        let mut out: Vec<TranscriptSegment> = Vec::new();
+        let mut on = |seg: TranscriptSegment| out.push(seg);
+        let step = vec![0.0f32; 16_000]; // exactly one 1s step
+
+        s.feed(&step, &decode, &mut on); // "hello" — interim
+        s.feed(&step, &decode, &mut on); // "hello world" — interim
+        s.feed(&step, &decode, &mut on); // stable + past min_commit → commit
+
+        assert_eq!(out.len(), 3);
+        assert!(out[0].interim && out[0].text == "hello");
+        assert!(out[1].interim && out[1].text == "hello world");
+        assert!(!out[2].interim && out[2].text == "hello world");
+        assert_eq!(out[0].id, out[2].id, "same line id until it commits");
+
+        // A new line after the commit uses a fresh id and a later offset.
+        let d2 = |_: &[f32], _o: u64| "next sentence".to_string();
+        s.feed(&step, &d2, &mut on);
+        assert!(out[3].interim && out[3].text == "next sentence");
+        assert_ne!(out[3].id, out[2].id);
+        assert!(out[3].start_ms >= 3000, "starts after the 3 committed steps");
+    }
+
+    #[test]
+    fn streamer_shows_no_line_for_pure_silence() {
+        let mut s = Streamer::new(StreamTuning::new(false));
+        let decode = |_: &[f32], _o: u64| String::new();
+        let mut out: Vec<TranscriptSegment> = Vec::new();
+        let mut on = |seg: TranscriptSegment| out.push(seg);
+        let step = vec![0.0f32; 16_000];
+        for _ in 0..5 {
+            s.feed(&step, &decode, &mut on);
+        }
+        assert!(out.is_empty(), "silence must not emit an empty line");
+    }
+
+    #[test]
+    fn streamer_flush_commits_the_in_progress_line() {
+        let mut s = Streamer::new(StreamTuning::new(false));
+        let decode = |_: &[f32], _o: u64| "partial".to_string();
+        let mut out: Vec<TranscriptSegment> = Vec::new();
+        let mut on = |seg: TranscriptSegment| out.push(seg);
+        s.feed(&vec![0.0f32; 16_000], &decode, &mut on);
+        assert!(out.last().unwrap().interim, "still growing before flush");
+        s.flush(&decode, &mut on);
+        let last = out.last().unwrap();
+        assert!(!last.interim && last.text == "partial", "flush finalizes it");
+    }
+
+    #[test]
+    fn streamer_force_commits_a_pause_free_monologue_at_max() {
+        // Text never repeats (continuous speech), so only the max-window
+        // backstop can commit it — otherwise the window (and decode cost) grows
+        // without bound.
+        let tuning = StreamTuning {
+            step_ms: 1000,
+            max_window_ms: 3000,
+            min_commit_ms: 1500,
+        };
+        let mut s = Streamer::new(tuning);
+        let n = std::cell::Cell::new(0usize);
+        let decode = |_: &[f32], _o: u64| {
+            let i = n.get();
+            n.set(i + 1);
+            format!("word{i}")
+        };
+        let mut out: Vec<TranscriptSegment> = Vec::new();
+        let mut on = |seg: TranscriptSegment| out.push(seg);
+        let step = vec![0.0f32; 16_000];
+        s.feed(&step, &decode, &mut on); // 1s
+        s.feed(&step, &decode, &mut on); // 2s
+        s.feed(&step, &decode, &mut on); // 3s == max → force commit despite changing text
+        assert!(!out.last().unwrap().interim);
+    }
+
     #[test]
     fn audio_ctx_scales_proportionally_to_window_length() {
         // The actual streaming WINDOW_SECS (4.0) -> ~200 (201 after `.ceil()`