diff --git a/openless-all/app/src-tauri/src/coordinator.rs b/openless-all/app/src-tauri/src/coordinator.rs index a2867b61..9618e30d 100644 --- a/openless-all/app/src-tauri/src/coordinator.rs +++ b/openless-all/app/src-tauri/src/coordinator.rs @@ -1363,9 +1363,11 @@ async fn end_session(inner: &Arc) -> Result<(), String> { if let Err(e) = asr.send_last_frame().await { log::error!("[coord] send last frame failed: {e}"); } - match asr.await_final_result().await { - Ok(r) => r, - Err(e) => { + // 添加全局超时保护：防止 await_final_result() 永远挂起 + let timeout_duration = std::time::Duration::from_secs(COORDINATOR_GLOBAL_TIMEOUT_SECS); + match tokio::time::timeout(timeout_duration, asr.await_final_result()).await { + Ok(Ok(r)) => r, + Ok(Err(e)) => { log::error!("[coord] await final failed: {e}"); emit_capsule( inner, @@ -1380,26 +1382,67 @@ async fn end_session(inner: &Arc) -> Result<(), String> { schedule_capsule_idle(inner, CAPSULE_AUTO_HIDE_DELAY_MS); return Err(e.to_string()); } + Err(_) => { + // 全局超时：最后的防线 + log::error!( + "[coord] 全局超时 {} 秒 - 强制恢复", + COORDINATOR_GLOBAL_TIMEOUT_SECS + ); + emit_capsule( + inner, + CapsuleState::Error, + 0.0, + elapsed, + Some("识别超时".to_string()), + None, + ); + restore_prepared_windows_ime_session(inner, current_session_id); + inner.state.lock().phase = SessionPhase::Idle; + schedule_capsule_idle(inner, CAPSULE_AUTO_HIDE_DELAY_MS); + return Err("global timeout".to_string()); + } } } - ActiveAsr::Whisper(w) => match w.transcribe().await { - Ok(r) => r, - Err(e) => { - log::error!("[coord] whisper transcribe failed: {e}"); - emit_capsule( - inner, - CapsuleState::Error, - 0.0, - elapsed, - Some(format!("识别失败: {e}")), - None, - ); - restore_prepared_windows_ime_session(inner, current_session_id); - inner.state.lock().phase = SessionPhase::Idle; - schedule_capsule_idle(inner, CAPSULE_AUTO_HIDE_DELAY_MS); - return Err(e.to_string()); + ActiveAsr::Whisper(w) => { + // Whisper 也添加类似的超时保护 + let timeout_duration = std::time::Duration::from_secs(COORDINATOR_GLOBAL_TIMEOUT_SECS); + match tokio::time::timeout(timeout_duration, w.transcribe()).await { + Ok(Ok(r)) => r, + Ok(Err(e)) => { + log::error!("[coord] whisper transcribe failed: {e}"); + emit_capsule( + inner, + CapsuleState::Error, + 0.0, + elapsed, + Some(format!("识别失败: {e}")), + None, + ); + restore_prepared_windows_ime_session(inner, current_session_id); + inner.state.lock().phase = SessionPhase::Idle; + schedule_capsule_idle(inner, CAPSULE_AUTO_HIDE_DELAY_MS); + return Err(e.to_string()); + } + Err(_) => { + log::error!( + "[coord] whisper 全局超时 {} 秒", + COORDINATOR_GLOBAL_TIMEOUT_SECS + ); + emit_capsule( + inner, + CapsuleState::Error, + 0.0, + elapsed, + Some("识别超时".to_string()), + None, + ); + restore_prepared_windows_ime_session(inner, current_session_id); + inner.state.lock().phase = SessionPhase::Idle; + schedule_capsule_idle(inner, CAPSULE_AUTO_HIDE_DELAY_MS); + return Err("whisper global timeout".to_string()); + } } - }, + } }; // ASR 完成后 cancel 检查：用户在 transcribe 进行中按 Esc 时，这里就会命中。 @@ -2907,6 +2950,11 @@ fn enabled_phrases(inner: &Arc) -> Vec { /// 用户点 ✕ / ✓ / 中途出错 / 按 Esc 都走这里，统一 2 秒。 const CAPSULE_AUTO_HIDE_DELAY_MS: u64 = 2000; +/// Coordinator 全局超时保护：防止 ASR await_final_result() 永远挂起。 +/// 设置为 15 秒（比 ASR 的 12 秒 FINAL_RESULT_TIMEOUT 稍长）， +/// 只在 ASR 超时机制失效时作为最后的防线触发。 +const COORDINATOR_GLOBAL_TIMEOUT_SECS: u64 = 15; + /// begin_session 中各 await 之间的 cancel race 检查结果。 enum BeginOutcome { /// 启动 continuation 属于旧 session；不能改动当前 session 状态。 diff --git a/openless-all/app/src-tauri/src/recorder.rs b/openless-all/app/src-tauri/src/recorder.rs index 4ab04aa6..e52fb3f3 100644 --- a/openless-all/app/src-tauri/src/recorder.rs +++ b/openless-all/app/src-tauri/src/recorder.rs @@ -115,7 +115,7 @@ fn run_audio_thread( startup_tx: Sender>, runtime_error_tx: Sender, ) { - let stream = match build_input_stream(consumer, level_handler, runtime_error_tx) { + let (stream, state) = match build_input_stream(consumer, level_handler, runtime_error_tx.clone()) { Ok(s) => s, Err(err) => { // 启动失败：通知主线程后即退出。 @@ -132,6 +132,37 @@ fn run_audio_thread( // 启动成功。 let _ = startup_tx.send(Ok(())); + // 启动 liveness watchdog 线程：检测录音回调是否静默停止 + const WATCHDOG_CHECK_INTERVAL_MS: u64 = 1000; // 每秒检查一次 + const CALLBACK_TIMEOUT_SECS: u64 = 3; // 3 秒没有回调视为异常 + + let stop_flag_for_watchdog = Arc::clone(&stop_flag); + let state_for_watchdog = Arc::clone(&state); + let runtime_error_tx_for_watchdog = runtime_error_tx.clone(); + + let watchdog_handle = thread::Builder::new() + .name("openless-recorder-watchdog".into()) + .spawn(move || { + while !stop_flag_for_watchdog.load(Ordering::SeqCst) { + thread::sleep(std::time::Duration::from_millis(WATCHDOG_CHECK_INTERVAL_MS)); + + if let Some(last_time) = *state_for_watchdog.last_callback_time.lock() { + let elapsed = last_time.elapsed(); + if elapsed.as_secs() > CALLBACK_TIMEOUT_SECS { + log::error!( + "[recorder] watchdog: 录音回调已停止 {} 秒，触发错误恢复", + elapsed.as_secs() + ); + let _ = runtime_error_tx_for_watchdog.send(RecorderError::EngineFailed( + format!("录音回调静默停止 {} 秒", elapsed.as_secs()) + )); + break; // 只报告一次 + } + } + } + }) + .ok(); + // 自旋等待停止信号——cpal 自身没有 wait API，sleep 50ms 完全够用。 while !stop_flag.load(Ordering::SeqCst) { thread::sleep(std::time::Duration::from_millis(50)); @@ -139,6 +170,11 @@ fn run_audio_thread( // Stream 在 drop 时自动停止。 drop(stream); + + // 等待 watchdog 线程退出 + if let Some(handle) = watchdog_handle { + let _ = handle.join(); + } } /// 选默认输入设备 + 默认配置 + 构造 Stream。 @@ -146,7 +182,7 @@ fn build_input_stream( consumer: Arc, level_handler: Arc, runtime_error_tx: Sender, -) -> Result { +) -> Result<(cpal::Stream, Arc), RecorderError> { let host = cpal::default_host(); let device = host .default_input_device() @@ -169,17 +205,18 @@ fn build_input_stream( ); let state = Arc::new(StreamState::new()); - build_stream_for_format( + let stream = build_stream_for_format( &device, &config, sample_format, consumer, level_handler, - state, + Arc::clone(&state), input_sr, channels, runtime_error_tx, - ) + )?; + Ok((stream, state)) } /// 启动期 default_input_config 失败：依靠错误字符串关键字粗判权限问题。 @@ -281,6 +318,8 @@ struct StreamState { callback_count: AtomicUsize, peak_input_rms_milli: AtomicUsize, peak_output_rms_milli: AtomicUsize, + /// 最后一次成功调用 consumer 的时间戳（用于 liveness 检测） + last_callback_time: Mutex>, } impl StreamState { @@ -291,6 +330,7 @@ impl StreamState { callback_count: AtomicUsize::new(0), peak_input_rms_milli: AtomicUsize::new(0), peak_output_rms_milli: AtomicUsize::new(0), + last_callback_time: Mutex::new(Some(std::time::Instant::now())), } } } @@ -322,6 +362,9 @@ fn process_callback( consumer.consume_pcm_chunk(&pcm_bytes); level_handler(level); + // 更新最后一次成功调用的时间戳（用于 liveness 检测） + *state.last_callback_time.lock() = Some(std::time::Instant::now()); + // 诊断：峰值 + 周期性日志。 let count = state.callback_count.fetch_add(1, Ordering::Relaxed) + 1; update_peak(&state.peak_input_rms_milli, input_rms);