From 944c77013bccb0d2e36af9aac0d9cde535598eb3 Mon Sep 17 00:00:00 2001 From: H-Chris233 Date: Fri, 1 May 2026 07:44:06 +0800 Subject: [PATCH] Surface recorder stream failures to the capsule Runtime cpal stream errors used to stop at the recorder log line, leaving the coordinator in an active recording phase until it processed empty audio. Add a small runtime-error channel from Recorder to the coordinator and abort only Starting/Listening sessions with an error capsule. Constraint: Issue #59 asks for minimal handling of recorder runtime failures. Rejected: Reworking recorder/coordinator ownership model | broader than needed for a single runtime error path. Confidence: high Scope-risk: narrow Tested: cargo check --manifest-path openless-all/app/src-tauri/Cargo.toml Tested: cargo test --manifest-path openless-all/app/src-tauri/Cargo.toml recorder_runtime_error_aborts_active_session Tested: git diff --check --- openless-all/app/src-tauri/src/coordinator.rs | 73 ++++++++++++++++++- openless-all/app/src-tauri/src/recorder.rs | 39 +++++++--- 2 files changed, 100 insertions(+), 12 deletions(-) diff --git a/openless-all/app/src-tauri/src/coordinator.rs b/openless-all/app/src-tauri/src/coordinator.rs index 53b1d613..fec5eaa1 100644 --- a/openless-all/app/src-tauri/src/coordinator.rs +++ b/openless-all/app/src-tauri/src/coordinator.rs @@ -26,7 +26,7 @@ use crate::persistence::{ }; use crate::polish::{OpenAICompatibleConfig, OpenAICompatibleLLMProvider}; -use crate::recorder::Recorder; +use crate::recorder::{Recorder, RecorderError}; use crate::types::{ CapsulePayload, CapsuleState, DictationSession, HotkeyCapability, HotkeyMode, HotkeyStatus, HotkeyStatusState, InsertStatus, PolishMode, @@ -583,8 +583,9 @@ fn start_recorder_for_starting( }); match Recorder::start(consumer, level_handler) { - Ok(rec) => { + Ok((rec, runtime_errors)) => { *inner.recorder.lock() = Some(rec); + spawn_recorder_error_monitor(inner, runtime_errors); stop_recorder_if_pending_start_stop(inner); log::info!("[coord] recorder started (asr={active_asr}, phase=Starting)"); } @@ -613,6 +614,56 @@ fn start_recorder_for_starting( Ok(()) } +fn spawn_recorder_error_monitor(inner: &Arc, rx: mpsc::Receiver) { + let inner = Arc::clone(inner); + std::thread::Builder::new() + .name("openless-recorder-error-monitor".into()) + .spawn(move || { + if let Ok(err) = rx.recv() { + log::error!("[coord] recorder runtime error: {err}"); + abort_recording_with_error(&inner, format!("录音中断: {err}")); + } + }) + .ok(); +} + +fn abort_recording_with_error(inner: &Arc, message: String) { + let elapsed = { + let mut state = inner.state.lock(); + if state.cancelled + || !matches!( + state.phase, + SessionPhase::Starting | SessionPhase::Listening + ) + { + return; + } + state.cancelled = true; + state.phase = SessionPhase::Idle; + state.started_at.elapsed().as_millis() as u64 + }; + + if let Some(rec) = inner.recorder.lock().take() { + rec.stop(); + } + if let Some(asr) = inner.asr.lock().take() { + match asr { + ActiveAsr::Volcengine(v) => v.cancel(), + ActiveAsr::Whisper(w) => w.cancel(), + } + } + + emit_capsule( + inner, + CapsuleState::Error, + 0.0, + elapsed, + Some(message), + None, + ); + schedule_capsule_idle(inner, CAPSULE_AUTO_HIDE_DELAY_MS); +} + async fn start_recorder_and_enter_listening( inner: &Arc, active_asr: &str, @@ -1157,6 +1208,24 @@ mod tests { assert!(state.pending_stop); } + #[test] + fn recorder_runtime_error_aborts_active_session() { + let coordinator = Coordinator::new(); + { + let mut state = coordinator.inner.state.lock(); + state.phase = SessionPhase::Listening; + state.cancelled = false; + } + + abort_recording_with_error(&coordinator.inner, "录音中断: stream failed".to_string()); + + let state = coordinator.inner.state.lock(); + assert_eq!(state.phase, SessionPhase::Idle); + assert!(state.cancelled); + assert!(coordinator.inner.recorder.lock().is_none()); + assert!(coordinator.inner.asr.lock().is_none()); + } + #[tokio::test] async fn repeated_pressed_edge_during_hold_session_does_not_restart() { let coordinator = Coordinator::new(); diff --git a/openless-all/app/src-tauri/src/recorder.rs b/openless-all/app/src-tauri/src/recorder.rs index e7fa96b3..4ab04aa6 100644 --- a/openless-all/app/src-tauri/src/recorder.rs +++ b/openless-all/app/src-tauri/src/recorder.rs @@ -11,7 +11,7 @@ //! - 主线程通过 `AtomicBool` 通知"该停了"，并 `join` 线程；线程内 `drop` Stream。 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; -use std::sync::mpsc::{channel, Sender}; +use std::sync::mpsc::{channel, Receiver, Sender}; use std::sync::Arc; use std::thread::{self, JoinHandle}; @@ -57,16 +57,24 @@ impl Recorder { pub fn start( consumer: Arc, level_handler: Arc, - ) -> Result { + ) -> Result<(Self, Receiver), RecorderError> { // 启动信号：子线程构造 Stream 完成后通过 startup_tx 报告结果。 let (startup_tx, startup_rx) = channel::>(); + // 运行期错误：Stream 已成功启动后，cpal 通过 err_cb 异步上报。 + let (runtime_error_tx, runtime_error_rx) = channel::(); let stop_flag = Arc::new(AtomicBool::new(false)); let stop_for_thread = Arc::clone(&stop_flag); let join_handle = thread::Builder::new() .name("openless-recorder".into()) .spawn(move || { - run_audio_thread(consumer, level_handler, stop_for_thread, startup_tx); + run_audio_thread( + consumer, + level_handler, + stop_for_thread, + startup_tx, + runtime_error_tx, + ); }) .map_err(|e| RecorderError::EngineFailed(format!("spawn audio thread: {e}")))?; @@ -77,10 +85,13 @@ impl Recorder { .map_err(|e| RecorderError::EngineFailed(format!("audio thread vanished: {e}")))?; startup_result?; - Ok(Self { - stop_flag, - join_handle: Mutex::new(Some(join_handle)), - }) + Ok(( + Self { + stop_flag, + join_handle: Mutex::new(Some(join_handle)), + }, + runtime_error_rx, + )) } /// 停止采集并等待音频线程退出。 @@ -102,8 +113,9 @@ fn run_audio_thread( level_handler: Arc, stop_flag: Arc, startup_tx: Sender>, + runtime_error_tx: Sender, ) { - let stream = match build_input_stream(consumer, level_handler) { + let stream = match build_input_stream(consumer, level_handler, runtime_error_tx) { Ok(s) => s, Err(err) => { // 启动失败：通知主线程后即退出。 @@ -133,6 +145,7 @@ fn run_audio_thread( fn build_input_stream( consumer: Arc, level_handler: Arc, + runtime_error_tx: Sender, ) -> Result { let host = cpal::default_host(); let device = host @@ -165,6 +178,7 @@ fn build_input_stream( state, input_sr, channels, + runtime_error_tx, ) } @@ -202,14 +216,19 @@ fn build_stream_for_format( state: Arc, input_sr: u32, channels: usize, + runtime_error_tx: Sender, ) -> Result { - let err_cb = |err| log::error!("[recorder] stream error: {err}"); - macro_rules! make_stream { ($t:ty, $to_f32:expr) => {{ let consumer = Arc::clone(&consumer); let level_handler = Arc::clone(&level_handler); let state = Arc::clone(&state); + let runtime_error_tx = runtime_error_tx.clone(); + let err_cb = move |err| { + log::error!("[recorder] stream error: {err}"); + let _ = + runtime_error_tx.send(RecorderError::EngineFailed(format!("stream: {err}"))); + }; device .build_input_stream::<$t, _, _>( config,