rvsttd: increase fade-out to 25ms

This reverts commit 84474731e7.

rvsttd/src/main.rs

@@ -8,6 +8,7 @@ use std::io::{BufRead, BufReader, Write};
 use std::net::{TcpListener, TcpStream};
 use std::path::PathBuf;
 use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::mpsc;
 use std::sync::{Arc, Mutex};
 use std::thread;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
@@ -69,12 +70,112 @@ fn line_text(line: &transcript_line_t) -> String {
         .into_owned()
 }
 
+fn unix_now() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_secs()
+}
+
+// ─── debug session recorder ──────────────────────────────────────────────
+
+struct DebugRecorder {
+    dir: PathBuf,
+    audio: Vec<f32>,
+    log_lines: Vec<String>,
+}
+
+impl DebugRecorder {
+    fn new(base_dir: &PathBuf) -> Self {
+        let dir = base_dir.join(unix_now().to_string());
+        std::fs::create_dir_all(&dir).ok();
+        Self {
+            dir,
+            audio: Vec::new(),
+            log_lines: Vec::new(),
+        }
+    }
+
+    fn log_event(&mut self, event: &str) {
+        self.log_lines.push(format!("[{}] {}", ts(), event));
+    }
+
+    fn log_segment(&mut self, line: &transcript_line_t, prefix: &str) {
+        let text = line_text(line);
+        let event = format!(
+            "SEGMENT id={} prefix={} is_complete={} start={:.3}s duration={:.3}s text=\"{}\"",
+            line.id, prefix, line.is_complete, line.start_time, line.duration, text
+        );
+        self.log_event(&event);
+    }
+
+    fn add_audio(&mut self, samples: &[f32]) {
+        self.audio.extend_from_slice(samples);
+    }
+
+    fn save(self) {
+        // Write log
+        let log_path = self.dir.join("session.log");
+        match std::fs::File::create(&log_path) {
+            Ok(mut f) => {
+                for line in &self.log_lines {
+                    writeln!(f, "{}", line).ok();
+                }
+            }
+            Err(e) => log(&format!("debug: failed to write log: {}", e)),
+        }
+
+        // Write WAV
+        let wav_path = self.dir.join("audio.wav");
+        match write_wav(&wav_path, &self.audio, SAMPLE_RATE as u32) {
+            Ok(()) => {
+                let secs = self.audio.len() as f64 / SAMPLE_RATE as f64;
+                log(&format!("debug: saved {} ({:.1}s, {} samples)", self.dir.display(), secs, self.audio.len()));
+            }
+            Err(e) => log(&format!("debug: failed to write wav: {}", e)),
+        }
+    }
+}
+
+fn write_wav(path: &PathBuf, samples: &[f32], sample_rate: u32) -> Result<()> {
+    let num_samples = samples.len() as u32;
+    let data_size = num_samples * 4; // f32 = 4 bytes
+    let file_size = 44 + data_size;
+
+    let mut f = std::fs::File::create(path)?;
+    // RIFF header
+    f.write_all(b"RIFF")?;
+    f.write_all(&(file_size - 8).to_le_bytes())?;
+    f.write_all(b"WAVE")?;
+    // fmt chunk
+    f.write_all(b"fmt ")?;
+    f.write_all(&16u32.to_le_bytes())?;       // chunk size
+    f.write_all(&3u16.to_le_bytes())?;         // IEEE float
+    f.write_all(&1u16.to_le_bytes())?;         // mono
+    f.write_all(&sample_rate.to_le_bytes())?;
+    f.write_all(&(sample_rate * 4).to_le_bytes())?; // byte rate
+    f.write_all(&4u16.to_le_bytes())?;         // block align
+    f.write_all(&32u16.to_le_bytes())?;        // bits per sample
+    // data chunk
+    f.write_all(b"data")?;
+    f.write_all(&data_size.to_le_bytes())?;
+
+    // Convert f32 samples to little-endian bytes
+    let mut bytes = Vec::with_capacity(data_size as usize);
+    for &s in samples {
+        bytes.extend_from_slice(&s.to_le_bytes());
+    }
+    f.write_all(&bytes)?;
+    Ok(())
+}
+
 // ─── shared state ─────────────────────────────────────────────────────────
 
 struct Shared {
     writer: Mutex<TcpStream>,
     session_id: u64,
     transcriber_handle: i32,
+    debug_dir: Option<PathBuf>,
 }
 
 impl Shared {
@@ -100,23 +201,26 @@ struct Session {
     stop_signal: Arc<AtomicBool>,
     aborted: Arc<AtomicBool>,
     transcriber: thread::JoinHandle<()>,
-    cpal_stream: cpal::Stream,
+    cpal_stream: Option<cpal::Stream>,
     stream_handle: i32,
 }
 
 impl Session {
-    fn stop(self) {
+    fn stop(mut self) {
+        // Signal transcriber to exit main loop, then wait for it to drain
+        // trailing audio + final flush. cpal stream stays alive during drain.
         self.stop_signal.store(true, Ordering::SeqCst);
-        drop(self.cpal_stream);
         self.transcriber.join().ok();
+        // Now safe to kill ALSA — transcriber is done
+        self.cpal_stream.take();
         unsafe { moonshine_free_stream(self.shared.transcriber_handle, self.stream_handle) };
     }
 
-    fn abort(self) {
+    fn abort(mut self) {
-        self.stop_signal.store(true, Ordering::SeqCst);
         self.aborted.store(true, Ordering::SeqCst);
-        drop(self.cpal_stream);
+        self.stop_signal.store(true, Ordering::SeqCst);
         self.transcriber.join().ok();
+        self.cpal_stream.take();
         unsafe { moonshine_free_stream(self.shared.transcriber_handle, self.stream_handle) };
     }
 }
@@ -134,12 +238,12 @@ fn start_session(shared: Arc<Shared>) -> Option<Session> {
         return None;
     }
 
-    let audio_buf: Arc<Mutex<Vec<f32>>> = Arc::new(Mutex::new(Vec::new()));
+    let (tx, rx) = mpsc::channel::<Vec<f32>>();
     let stop_signal = Arc::new(AtomicBool::new(false));
     let aborted = Arc::new(AtomicBool::new(false));
 
-    let cpal_stream = match start_cpal(audio_buf.clone(), stop_signal.clone()) {
+    let cpal_stream = match start_cpal(tx) {
-        Ok(s) => s,
+        Ok(s) => Some(s),
         Err(e) => {
             log(&format!("cpal failed: {}", e));
             unsafe { moonshine_free_stream(shared.transcriber_handle, stream_handle) };
@@ -152,7 +256,7 @@ fn start_session(shared: Arc<Shared>) -> Option<Session> {
     let aborted_clone = aborted.clone();
 
     let transcriber = thread::spawn(move || {
-        transcriber_loop(shared_clone, audio_buf, stop_signal_clone, aborted_clone, stream_handle);
+        transcriber_loop(shared_clone, rx, stop_signal_clone, aborted_clone, stream_handle);
     });
 
     Some(Session {
@@ -167,7 +271,7 @@ fn start_session(shared: Arc<Shared>) -> Option<Session> {
 
 fn transcriber_loop(
     shared: Arc<Shared>,
-    audio_buf: Arc<Mutex<Vec<f32>>>,
+    rx: mpsc::Receiver<Vec<f32>>,
     stop_signal: Arc<AtomicBool>,
     aborted: Arc<AtomicBool>,
     stream_handle: i32,
@@ -175,17 +279,91 @@ fn transcriber_loop(
     let handle = shared.transcriber_handle;
     let mut sent_ids: HashSet<u64> = HashSet::new();
 
-    while !stop_signal.load(Ordering::SeqCst) {
+    // Debug recorder (if enabled)
-        let chunk = {
+    let mut recorder = shared.debug_dir.as_ref().map(|_| DebugRecorder::new(
-            let mut buf = audio_buf.lock().unwrap();
+        &shared.debug_dir.as_ref().unwrap().join(shared.session_id.to_string()),
-            if buf.is_empty() {
+    ));
-                drop(buf);
-                thread::sleep(Duration::from_millis(5));
-                continue;
-            }
-            std::mem::take(&mut *buf)
-        };
 
+    if let Some(ref mut r) = recorder {
+        r.log_event(&format!("session {} started", shared.session_id));
+    }
+
+    // Main loop: process audio until stop_signal
+    while !stop_signal.load(Ordering::SeqCst) {
+        match rx.recv_timeout(Duration::from_millis(100)) {
+            Ok(chunk) => {
+                if let Some(ref mut r) = recorder {
+                    r.add_audio(&chunk);
+                }
+
+                unsafe {
+                    moonshine_transcribe_add_audio_to_stream(
+                        handle, stream_handle,
+                        chunk.as_ptr(), chunk.len() as u64,
+                        SAMPLE_RATE, 0,
+                    );
+                }
+
+                let mut t_ptr: *mut transcript_t = std::ptr::null_mut();
+                let rc = unsafe { moonshine_transcribe_stream(handle, stream_handle, 0, &mut t_ptr) };
+                if rc != 0 || t_ptr.is_null() {
+                    continue;
+                }
+
+                if let Some(ref mut r) = recorder {
+                    log_transcript_lines(r, t_ptr);
+                }
+
+                send_new_segments(&shared, t_ptr, &mut sent_ids, "P");
+            }
+            Err(mpsc::RecvTimeoutError::Timeout) => continue,
+            Err(mpsc::RecvTimeoutError::Disconnected) => break,
+        }
+    }
+
+    // Drain trailing audio from ALSA buffer (cpal stream still alive).
+    // Fixed 100ms window — cpal delivers every ~50ms (800 samples @ 16kHz),
+    // so this captures 1-2 more callbacks worth of trailing audio.
+    // Hold one chunk back so we can apply a fade-out to the very last one.
+    let drain_deadline = std::time::Instant::now() + Duration::from_millis(100);
+    let mut held_chunk: Option<Vec<f32>> = None;
+    while std::time::Instant::now() < drain_deadline {
+        match rx.recv_timeout(drain_deadline - std::time::Instant::now()) {
+            Ok(chunk) => {
+                // Feed previously held chunk to Moonshine + recorder (no fade)
+                if let Some(prev) = held_chunk.take() {
+                    if let Some(ref mut r) = recorder {
+                        r.add_audio(&prev);
+                    }
+                    unsafe {
+                        moonshine_transcribe_add_audio_to_stream(
+                            handle, stream_handle,
+                            prev.as_ptr(), prev.len() as u64,
+                            SAMPLE_RATE, 0,
+                        );
+                    }
+                }
+                held_chunk = Some(chunk);
+            }
+            Err(mpsc::RecvTimeoutError::Timeout) => break,
+            Err(mpsc::RecvTimeoutError::Disconnected) => break,
+        }
+    }
+
+    // Apply 5ms fade-out to the last chunk, then feed to both Moonshine and recorder
+    if let Some(mut chunk) = held_chunk.take() {
+        let fade_samples = (SAMPLE_RATE as usize * 25) / 1000; // 25ms
+        if chunk.len() > fade_samples {
+            let start = chunk.len() - fade_samples;
+            for i in 0..fade_samples {
+                let t = 1.0 - (i as f32 / fade_samples as f32);
+                chunk[start + i] *= t;
+            }
+        }
+
+        if let Some(ref mut r) = recorder {
+            r.add_audio(&chunk);
+        }
         unsafe {
             moonshine_transcribe_add_audio_to_stream(
                 handle, stream_handle,
@@ -193,36 +371,17 @@ fn transcriber_loop(
                 SAMPLE_RATE, 0,
             );
         }
-
-        let mut t_ptr: *mut transcript_t = std::ptr::null_mut();
-        let rc = unsafe { moonshine_transcribe_stream(handle, stream_handle, 0, &mut t_ptr) };
-        if rc != 0 || t_ptr.is_null() {
-            continue;
-        }
-
-        send_new_segments(&shared, t_ptr, &mut sent_ids, "P");
     }
 
     // If aborted (new session took over), skip final flush entirely
     if aborted.load(Ordering::SeqCst) {
         unsafe { moonshine_stop_stream(handle, stream_handle) };
         log(&format!("Session {} aborted, skipping final flush", shared.session_id));
-        return;
+        if let Some(mut r) = recorder {
-    }
+            r.log_event("aborted (new session took over)");
-
+            r.save();
-    // Drain remaining audio
-    let remaining = {
-        let mut buf = audio_buf.lock().unwrap();
-        std::mem::take(&mut *buf)
-    };
-    if !remaining.is_empty() {
-        unsafe {
-            moonshine_transcribe_add_audio_to_stream(
-                handle, stream_handle,
-                remaining.as_ptr(), remaining.len() as u64,
-                SAMPLE_RATE, 0,
-            );
         }
+        return;
     }
 
     // Final flush
@@ -234,6 +393,10 @@ fn transcriber_loop(
         let t = unsafe { &*t_ptr };
         let mut new_segments: Vec<String> = Vec::new();
 
+        if let Some(ref mut r) = recorder {
+            log_transcript_lines(r, t_ptr);
+        }
+
         for i in 0..t.line_count as usize {
             let line = unsafe { &*t.lines.add(i) };
             if line.text.is_null() || line.is_complete == 0 {
@@ -255,10 +418,32 @@ fn transcriber_loop(
             for (i, text) in new_segments.iter().enumerate() {
                 let prefix = if i == last { "F" } else { "P" };
                 shared.send_msg(prefix, text);
+                if let Some(ref mut r) = recorder {
+                    r.log_event(&format!("TX {} \"{}\"", prefix, text));
+                }
             }
         }
     } else {
         shared.send_msg("F", "");
+        if let Some(ref mut r) = recorder {
+            r.log_event("TX F (empty)");
+        }
+    }
+
+    if let Some(mut r) = recorder {
+        r.log_event("session ended");
+        r.save();
+    }
+}
+
+fn log_transcript_lines(recorder: &mut DebugRecorder, t_ptr: *const transcript_t) {
+    let t = unsafe { &*t_ptr };
+    for i in 0..t.line_count as usize {
+        let line = unsafe { &*t.lines.add(i) };
+        if line.text.is_null() {
+            continue;
+        }
+        recorder.log_segment(line, if line.is_complete != 0 { "complete" } else { "partial" });
     }
 }
 
@@ -289,8 +474,7 @@ fn send_new_segments(
 // ─── cpal ─────────────────────────────────────────────────────────────────
 
 fn start_cpal(
-    audio_buf: Arc<Mutex<Vec<f32>>>,
+    tx: mpsc::Sender<Vec<f32>>,
-    stop_signal: Arc<AtomicBool>,
 ) -> Result<cpal::Stream> {
     let host = cpal::default_host();
     let dev = host
@@ -314,6 +498,7 @@ fn start_cpal(
         config.channels = 1;
     }
     config.sample_rate = SampleRate(16000);
+    config.buffer_size = cpal::BufferSize::Fixed(800);
 
     let err_fn = |e: cpal::StreamError| log(&format!("cpal error: {}", e));
 
@@ -321,9 +506,7 @@ fn start_cpal(
         SampleFormat::F32 => dev.build_input_stream(
             &config,
             move |data: &[f32], _: &_| {
-                if !stop_signal.load(Ordering::Relaxed) {
+                let _ = tx.send(data.to_vec());
-                    audio_buf.lock().unwrap().extend_from_slice(data);
-                }
             },
             err_fn,
             None,
@@ -331,9 +514,7 @@ fn start_cpal(
         SampleFormat::I16 => dev.build_input_stream(
             &config,
             move |data: &[i16], _: &_| {
-                if !stop_signal.load(Ordering::Relaxed) {
+                let _ = tx.send(data.iter().map(|&x| x as f32 / 32768.0).collect());
-                    audio_buf.lock().unwrap().extend(data.iter().map(|&x| x as f32 / 32768.0));
-                }
             },
             err_fn,
             None,
@@ -341,9 +522,7 @@ fn start_cpal(
         SampleFormat::U8 => dev.build_input_stream(
             &config,
             move |data: &[u8], _: &_| {
-                if !stop_signal.load(Ordering::Relaxed) {
+                let _ = tx.send(data.iter().map(|&x| (x as f32 - 128.0) / 128.0).collect());
-                    audio_buf.lock().unwrap().extend(data.iter().map(|&x| (x as f32 - 128.0) / 128.0));
-                }
             },
             err_fn,
             None,
@@ -468,17 +647,22 @@ fn main() -> Result<()> {
 
     let mut args = first.into_iter().chain(args);
     let mut model_dir = default_model;
+    let mut debug = false;
 
     while let Some(a) = args.next() {
         match a.as_str() {
             "--model-dir" | "-m" => {
                 model_dir = args.next().unwrap_or(model_dir);
             }
+            "--debug" => {
+                debug = true;
+            }
             "--help" | "-h" => {
-                println!("Usage: rvsttd [--model-dir DIR]");
+                println!("Usage: rvsttd [--model-dir DIR] [--debug]");
                 println!("       rvsttd fetch [DIR]");
                 println!("Listens on TCP {}", BIND_ADDR);
                 println!("Model: medium-streaming (Moonshine)");
+                println!("--debug: save session audio + transcript log to ~/.rvsttd/debug/");
                 println!("'rvsttd fetch' downloads the English medium-streaming model");
                 return Ok(());
             }
@@ -486,6 +670,15 @@ fn main() -> Result<()> {
         }
     }
 
+    let debug_dir = if debug {
+        let d = PathBuf::from(&home).join(".rvsttd").join("debug");
+        std::fs::create_dir_all(&d).ok();
+        log(&format!("Debug mode enabled — sessions saved to {}", d.display()));
+        Some(d)
+    } else {
+        None
+    };
+
     let model_path = std::fs::canonicalize(&model_dir)
         .unwrap_or_else(|_| std::path::PathBuf::from(&model_dir));
 
@@ -547,6 +740,7 @@ fn main() -> Result<()> {
                     writer: Mutex::new(writer_stream.try_clone()?),
                     session_id: new_session_id,
                     transcriber_handle,
+                    debug_dir: debug_dir.clone(),
                 });
 
                 log(&format!("PTT on session {}", new_session_id));