fix(dictation): 转写阶段可中断——按 Esc 立即取消在途 ASR 请求 (#798)
* fix(dictation): 转写阶段可中断——按 Esc 立即取消在途 ASR 请求 进入 Processing（转写）后，end_session 已把 ASR 句柄从 inner.asr 槽 take 走， cancel_session 走的 cancel_asr_for_session 是 no-op，够不到在途请求：用户按 Esc 后仍要等整段转写跑完才停，且结果被静默丢弃，白等一场。 改为把转写与「用户取消」轮询用 tokio::select! 赛跑：命中取消即 drop 掉 transcribe future（中断 reqwest HTTP / 停止等待流式最终结果 / 停止本地转写），再显式 cancel_active_asr 促流式 WebSocket 立即关闭，然后 finish_cancelled_processing 收尾 静默返回。覆盖云端流式 / 云端批量 / 本地引擎全部 provider。 顺带删除 Foundry / Sherpa / Apple Speech 分支里各自的 if cancelled 早退补丁—— 取消现在由外层 select! 统一处理，那些分支已冗余。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * fix(dictation): clean up cancelled ASR sessions --------- Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com> Co-authored-by: Chris233 <h-chris233@outlook.com>
This commit is contained in:
4 files changed
+564
-334
No files matched your search
@@ -7,6 +7,8 @@
|
||||
//! engine 现在由 `LocalAsrCache` 提供——Coordinator 在 build_local_qwen3 里
|
||||
//! 取已缓存的引擎再传进来，避免每次会话都重加载 1.2GB+ 模型。
|
||||
|
||||
#[cfg(target_os = "macos")]
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
#[cfg(target_os = "macos")]
|
||||
use std::sync::Arc;
|
||||
|
||||
@@ -25,6 +27,9 @@ use crate::asr::RawTranscript;
|
||||
#[cfg(target_os = "macos")]
|
||||
pub struct LocalQwenAsr {
|
||||
engine: Arc<QwenAsrEngine>,
|
||||
/// `spawn_blocking` 已经启动后不能被 JoinHandle 强制终止；取消先关闭
|
||||
/// 当前会话的 token 门控，确保 native worker 返回前不会向新 UI 会话泄漏旧 token。
|
||||
cancelled: Arc<AtomicBool>,
|
||||
/// 16-bit LE PCM 字节缓冲（recorder 推什么我们存什么），在 transcribe 时再
|
||||
/// 转 f32 喂给 C 端。一次会话最多几 MB，clone 一次成本可接受。
|
||||
buffer: Mutex<Vec<u8>>,
|
||||
@@ -36,6 +41,7 @@ impl LocalQwenAsr {
|
||||
pub fn new(app: AppHandle, engine: Arc<QwenAsrEngine>) -> Self {
|
||||
Self {
|
||||
engine,
|
||||
cancelled: Arc::new(AtomicBool::new(false)),
|
||||
buffer: Mutex::new(Vec::new()),
|
||||
app,
|
||||
}
|
||||
@@ -51,6 +57,7 @@ impl LocalQwenAsr {
|
||||
/// stop 时调用：把 buffer 的 i16 PCM 转 f32，跑流式转写，token 实时
|
||||
/// 通过事件吐到前端胶囊；最终文本一起返回供 polish/insert。
|
||||
pub async fn transcribe(self: Arc<Self>) -> Result<RawTranscript> {
|
||||
self.cancelled.store(false, Ordering::Release);
|
||||
let pcm_bytes = std::mem::take(&mut *self.buffer.lock());
|
||||
if pcm_bytes.is_empty() {
|
||||
return Ok(RawTranscript {
|
||||
@@ -68,32 +75,35 @@ impl LocalQwenAsr {
|
||||
samples_f32.extend(std::iter::repeat(0.0f32).take(8_000));
|
||||
|
||||
// 注册 token 回调：每个稳定 token 抛 `local-asr-token` 事件。
|
||||
// capsule 前端按 sessionId 累积显示。
|
||||
// capsule 前端按 sessionId 累积显示。回调安装、native 调用和解绑
|
||||
// 在 QwenAsrEngine 的同一把 context 锁内完成。
|
||||
let app = self.app.clone();
|
||||
self.engine.set_token_handler(Some(move |piece: &str| {
|
||||
if let Err(e) = app.emit("local-asr-token", piece.to_string()) {
|
||||
log::warn!("[local-asr] emit token failed: {e}");
|
||||
}
|
||||
}));
|
||||
let cancelled = Arc::clone(&self.cancelled);
|
||||
|
||||
// qwen_transcribe_stream 是阻塞调用；用 spawn_blocking 防止占住 tokio runtime。
|
||||
// 用 tauri::async_runtime::spawn_blocking 而非 tokio 的 —— 同 download.rs 注释，
|
||||
// 走 Tauri 持有的 runtime handle，不依赖调用方上下文（虽然这里目前都在 async 路径上调，
|
||||
// 但保持一致更稳）。
|
||||
let engine = Arc::clone(&self.engine);
|
||||
let text =
|
||||
tauri::async_runtime::spawn_blocking(move || engine.transcribe_stream(&samples_f32))
|
||||
.await
|
||||
.context("transcribe spawn_blocking join 失败")?
|
||||
.context("qwen_transcribe_stream 失败")?;
|
||||
|
||||
// 解绑回调，避免 idle 期 C 端任何后续触发。
|
||||
self.engine.set_token_handler::<fn(&str)>(None);
|
||||
let text = tauri::async_runtime::spawn_blocking(move || {
|
||||
engine.transcribe_stream_with_handler(&samples_f32, move |piece: &str| {
|
||||
if cancelled.load(Ordering::Acquire) {
|
||||
return;
|
||||
}
|
||||
if let Err(e) = app.emit("local-asr-token", piece.to_string()) {
|
||||
log::warn!("[local-asr] emit token failed: {e}");
|
||||
}
|
||||
})
|
||||
})
|
||||
.await
|
||||
.context("transcribe spawn_blocking join 失败")?
|
||||
.context("qwen_transcribe_stream 失败")?;
|
||||
|
||||
Ok(RawTranscript { text, duration_ms })
|
||||
}
|
||||
|
||||
pub fn cancel(&self) {
|
||||
self.cancelled.store(true, Ordering::Release);
|
||||
self.buffer.lock().clear();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -7,7 +7,7 @@ use std::ffi::{CStr, CString};
|
||||
use std::os::raw::{c_char, c_void};
|
||||
use std::path::Path;
|
||||
use std::ptr;
|
||||
use std::sync::Mutex;
|
||||
use std::sync::{Mutex, MutexGuard};
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
|
||||
@@ -22,14 +22,18 @@ type TokenHandlerBox = Box<Box<TokenHandler>>;
|
||||
|
||||
pub struct QwenAsrEngine {
|
||||
ctx: *mut QwenCtx,
|
||||
/// 同一个 C context 不能并发转写，也不能在转写期间替换 token 回调。
|
||||
/// 取消时上层只会丢弃 `spawn_blocking` 的 JoinHandle，已经开始运行的
|
||||
/// native worker 仍会继续到返回；这把锁保证它返回前不会被下一次会话复用。
|
||||
run_lock: Mutex<()>,
|
||||
/// 持有 token 回调的所有权；C 端拿到的是 `&**handler` 派生出来的 raw ptr，
|
||||
/// 只要这个 Box 还活着，那个 raw ptr 就有效。Mutex 防止并发 set。
|
||||
token_handler: Mutex<Option<TokenHandlerBox>>,
|
||||
}
|
||||
|
||||
/// SAFETY: `qwen_ctx_t` 内部的 pthread/buffer 仅在单次 transcribe 期间被 C 端
|
||||
/// 自己用；外层不会从两个 Rust 线程并发调进同一个 ctx（由 coordinator 串行
|
||||
/// 化保证）。Send/Sync 在这一约束下成立。
|
||||
/// 自己用；`run_lock` 保证同一 context 不会被两个 Rust 线程并发调用。Send/Sync
|
||||
/// 在这一约束下成立。
|
||||
unsafe impl Send for QwenAsrEngine {}
|
||||
unsafe impl Sync for QwenAsrEngine {}
|
||||
|
||||
@@ -50,12 +54,27 @@ impl QwenAsrEngine {
|
||||
|
||||
Ok(Self {
|
||||
ctx,
|
||||
run_lock: Mutex::new(()),
|
||||
token_handler: Mutex::new(None),
|
||||
})
|
||||
}
|
||||
|
||||
/// 注册流式 token 回调；传 `None` 清空。重新注册会先解绑再装新回调。
|
||||
pub fn set_token_handler<F>(&self, handler: Option<F>)
|
||||
where
|
||||
F: FnMut(&str) + Send + 'static,
|
||||
{
|
||||
let _run_guard = self.lock_run();
|
||||
self.set_token_handler_locked(handler);
|
||||
}
|
||||
|
||||
fn lock_run(&self) -> MutexGuard<'_, ()> {
|
||||
self.run_lock
|
||||
.lock()
|
||||
.unwrap_or_else(|poisoned| poisoned.into_inner())
|
||||
}
|
||||
|
||||
fn set_token_handler_locked<F>(&self, handler: Option<F>)
|
||||
where
|
||||
F: FnMut(&str) + Send + 'static,
|
||||
{
|
||||
@@ -84,6 +103,11 @@ impl QwenAsrEngine {
|
||||
|
||||
/// 批式转写：一次性给完整音频（mono f32 16kHz）。
|
||||
pub fn transcribe_audio(&self, samples: &[f32]) -> Result<String> {
|
||||
let _run_guard = self.lock_run();
|
||||
self.transcribe_audio_locked(samples)
|
||||
}
|
||||
|
||||
fn transcribe_audio_locked(&self, samples: &[f32]) -> Result<String> {
|
||||
if self.ctx.is_null() {
|
||||
anyhow::bail!("engine already freed — cannot transcribe");
|
||||
}
|
||||
@@ -103,6 +127,11 @@ impl QwenAsrEngine {
|
||||
/// 流式转写：内部按 2s chunk 切片，token 通过 `set_token_handler` 注册的
|
||||
/// 回调实时吐出；返回值是最终完整文本。
|
||||
pub fn transcribe_stream(&self, samples: &[f32]) -> Result<String> {
|
||||
let _run_guard = self.lock_run();
|
||||
self.transcribe_stream_locked(samples)
|
||||
}
|
||||
|
||||
fn transcribe_stream_locked(&self, samples: &[f32]) -> Result<String> {
|
||||
if self.ctx.is_null() {
|
||||
anyhow::bail!("engine already freed — cannot transcribe");
|
||||
}
|
||||
@@ -117,6 +146,22 @@ impl QwenAsrEngine {
|
||||
unsafe { libc::free(raw as *mut c_void) };
|
||||
Ok(text)
|
||||
}
|
||||
|
||||
/// 在同一把 context 锁内安装回调、运行 native 转写并解绑回调。
|
||||
///
|
||||
/// `spawn_blocking` 的 JoinHandle 被取消时，已经启动的 blocking closure
|
||||
/// 仍可能运行；把回调解绑放进 closure 自身的同步收尾路径，避免旧会话
|
||||
/// 在下一次会话期间继续持有或调用失效的 userdata。
|
||||
pub fn transcribe_stream_with_handler<F>(&self, samples: &[f32], handler: F) -> Result<String>
|
||||
where
|
||||
F: FnMut(&str) + Send + 'static,
|
||||
{
|
||||
let _run_guard = self.lock_run();
|
||||
self.set_token_handler_locked(Some(handler));
|
||||
let result = self.transcribe_stream_locked(samples);
|
||||
self.set_token_handler_locked::<fn(&str)>(None);
|
||||
result
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for QwenAsrEngine {
|
||||
|
||||
@@ -370,6 +370,42 @@ struct PreparedWindowsImeSessionSlot {
|
||||
prepared: PreparedWindowsImeSession,
|
||||
}
|
||||
|
||||
/// 历史音频静默重试的 ASR 资源护栏。
|
||||
///
|
||||
/// 重试 future 被 select 丢弃时，局部 QaAsrStart 不会再经过正常的
|
||||
/// end_session 收尾；这里用 Drop 补 cancel 和本地模型释放，尤其覆盖
|
||||
/// spawn_blocking 已经开始运行的本地 ASR。
|
||||
struct CancellableRetranscribeGuard {
|
||||
inner: Arc<Inner>,
|
||||
asr: Option<ActiveAsr>,
|
||||
session_id: SessionId,
|
||||
}
|
||||
|
||||
impl CancellableRetranscribeGuard {
|
||||
fn new(inner: Arc<Inner>, asr: ActiveAsr, session_id: SessionId) -> Self {
|
||||
Self {
|
||||
inner,
|
||||
asr: Some(asr),
|
||||
session_id,
|
||||
}
|
||||
}
|
||||
|
||||
fn disarm(mut self) {
|
||||
self.asr.take();
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for CancellableRetranscribeGuard {
|
||||
fn drop(&mut self) {
|
||||
let Some(asr) = self.asr.take() else {
|
||||
return;
|
||||
};
|
||||
let asr_for_release = asr.clone();
|
||||
cancel_active_asr(asr);
|
||||
dictation::schedule_cancelled_asr_release(&self.inner, &asr_for_release, self.session_id);
|
||||
}
|
||||
}
|
||||
|
||||
impl Coordinator {
|
||||
pub fn new() -> Self {
|
||||
#[cfg(target_os = "windows")]
|
||||
@@ -1556,9 +1592,33 @@ impl Coordinator {
|
||||
}
|
||||
|
||||
pub async fn retranscribe_pcm(&self, pcm: Vec<u8>) -> Result<String, String> {
|
||||
self.retranscribe_pcm_inner(pcm, false).await
|
||||
}
|
||||
|
||||
pub(super) async fn retranscribe_pcm_until_cancelled(
|
||||
&self,
|
||||
pcm: Vec<u8>,
|
||||
) -> Result<String, String> {
|
||||
self.retranscribe_pcm_inner(pcm, true).await
|
||||
}
|
||||
|
||||
async fn retranscribe_pcm_inner(
|
||||
&self,
|
||||
pcm: Vec<u8>,
|
||||
cancel_on_drop: bool,
|
||||
) -> Result<String, String> {
|
||||
let inner = &self.inner;
|
||||
let active_asr = CredentialsVault::get_active_asr();
|
||||
let start = build_qa_asr_start(inner, &active_asr).await?;
|
||||
let retry_guard = if cancel_on_drop {
|
||||
Some(CancellableRetranscribeGuard::new(
|
||||
Arc::clone(inner),
|
||||
start.active_asr(),
|
||||
inner.state.lock().session_id,
|
||||
))
|
||||
} else {
|
||||
None
|
||||
};
|
||||
start.open_streaming_session().await?;
|
||||
let consumer = start.recorder_consumer();
|
||||
consumer.consume_pcm_chunk(&pcm);
|
||||
@@ -1627,6 +1687,9 @@ impl Coordinator {
|
||||
.map_err(|_| "重新转录超时".to_string())?
|
||||
.map_err(|e| e.to_string())?,
|
||||
};
|
||||
if let Some(guard) = retry_guard {
|
||||
guard.disarm();
|
||||
}
|
||||
Ok(raw.text)
|
||||
}
|
||||
|
||||
|
||||
@@ -2174,6 +2174,12 @@ const SILENT_RETRY_MAX: u32 = 2;
|
||||
/// 网络/服务端一点缓冲再打。
|
||||
const SILENT_RETRY_BACKOFF_MS: u64 = 500;
|
||||
|
||||
enum SilentRetryOutcome {
|
||||
Transcript(RawTranscript),
|
||||
Exhausted,
|
||||
Cancelled,
|
||||
}
|
||||
|
||||
/// 归档 wav 是 16k/mono/16-bit、固定 44 字节标准头（asr::wav::encode_wav_16k_mono）；取出
|
||||
/// PCM 负载。长度 <= 44（空/损坏）返回 None。
|
||||
fn pcm_from_wav_bytes(wav: &[u8]) -> Option<Vec<u8>> {
|
||||
@@ -2196,49 +2202,70 @@ async fn retranscribe_pcm_via_inner(inner: &Arc<Inner>, pcm: Vec<u8>) -> Result<
|
||||
Coordinator {
|
||||
inner: Arc::clone(inner),
|
||||
}
|
||||
.retranscribe_pcm(pcm)
|
||||
.retranscribe_pcm_until_cancelled(pcm)
|
||||
.await
|
||||
}
|
||||
|
||||
/// 自动静默重试：从刚归档的 wav 读 PCM，用当前 provider 重转最多 SILENT_RETRY_MAX 次（线性
|
||||
/// 退避）。任一次拿到非空文本立即 Some（当作正常转写继续走润色/插入）；没有归档音频、读不到、
|
||||
/// 或全部失败则 None（交回 fail_dictation 做「失败保留 + 报错」）。全程不改胶囊文案——对用户
|
||||
/// 静默，只是「转写中」多停留一会儿。
|
||||
async fn try_silent_retranscribe(
|
||||
inner: &Arc<Inner>,
|
||||
session_id: SessionId,
|
||||
) -> Option<RawTranscript> {
|
||||
if !inner.audio_archive_active.load(Ordering::Relaxed) {
|
||||
return None; // 没归档音频，无从重试
|
||||
/// 退避）。任一次拿到非空文本立即返回 Transcript（当作正常转写继续走润色/插入）；没有归档
|
||||
/// 音频、读不到或全部失败返回 Exhausted（交回 fail_dictation 做「失败保留 + 报错」）。如果
|
||||
/// 用户在退避或重试请求期间按 Esc，则返回 Cancelled，直接完成取消收尾。全程不改胶囊文案——
|
||||
/// 对用户静默，只是「转写中」多停留一会儿。
|
||||
async fn try_silent_retranscribe(inner: &Arc<Inner>, session_id: SessionId) -> SilentRetryOutcome {
|
||||
if inner.state.lock().cancelled {
|
||||
return SilentRetryOutcome::Cancelled;
|
||||
}
|
||||
let path = crate::persistence::recording_path_for_session(&session_id.to_string()).ok()?;
|
||||
let wav = tokio::fs::read(&path).await.ok()?;
|
||||
let pcm = pcm_from_wav_bytes(&wav)?;
|
||||
if !inner.audio_archive_active.load(Ordering::Relaxed) {
|
||||
return SilentRetryOutcome::Exhausted; // 没归档音频，无从重试
|
||||
}
|
||||
let Some(path) = crate::persistence::recording_path_for_session(&session_id.to_string()).ok()
|
||||
else {
|
||||
return SilentRetryOutcome::Exhausted;
|
||||
};
|
||||
let wav = tokio::select! {
|
||||
biased;
|
||||
_ = wait_for_processing_cancel(inner) => return SilentRetryOutcome::Cancelled,
|
||||
result = tokio::fs::read(&path) => match result {
|
||||
Ok(wav) => wav,
|
||||
Err(_) => return SilentRetryOutcome::Exhausted,
|
||||
},
|
||||
};
|
||||
let Some(pcm) = pcm_from_wav_bytes(&wav) else {
|
||||
return SilentRetryOutcome::Exhausted;
|
||||
};
|
||||
let duration_ms = pcm_duration_ms(pcm.len());
|
||||
for attempt in 1..=SILENT_RETRY_MAX {
|
||||
tokio::time::sleep(std::time::Duration::from_millis(
|
||||
SILENT_RETRY_BACKOFF_MS * attempt as u64,
|
||||
))
|
||||
.await;
|
||||
match retranscribe_pcm_via_inner(inner, pcm.clone()).await {
|
||||
tokio::select! {
|
||||
biased;
|
||||
_ = wait_for_processing_cancel(inner) => return SilentRetryOutcome::Cancelled,
|
||||
_ = tokio::time::sleep(std::time::Duration::from_millis(
|
||||
SILENT_RETRY_BACKOFF_MS * attempt as u64,
|
||||
)) => {}
|
||||
}
|
||||
let result = tokio::select! {
|
||||
biased;
|
||||
_ = wait_for_processing_cancel(inner) => return SilentRetryOutcome::Cancelled,
|
||||
result = retranscribe_pcm_via_inner(inner, pcm.clone()) => result,
|
||||
};
|
||||
match result {
|
||||
Ok(text) if !text.trim().is_empty() => {
|
||||
log::info!(
|
||||
"[coord] 自动静默重试第 {attempt}/{SILENT_RETRY_MAX} 次成功（{} 字）",
|
||||
text.chars().count()
|
||||
);
|
||||
return Some(RawTranscript { text, duration_ms });
|
||||
return SilentRetryOutcome::Transcript(RawTranscript { text, duration_ms });
|
||||
}
|
||||
Ok(_) => {
|
||||
// 重试得到空转写——多半真没说话，再重试无意义，省流量直接放弃。
|
||||
log::info!("[coord] 自动静默重试得到空转写，停止重试");
|
||||
return None;
|
||||
return SilentRetryOutcome::Exhausted;
|
||||
}
|
||||
Err(e) => {
|
||||
log::warn!("[coord] 自动静默重试第 {attempt}/{SILENT_RETRY_MAX} 次失败: {e}");
|
||||
}
|
||||
}
|
||||
}
|
||||
None
|
||||
SilentRetryOutcome::Exhausted
|
||||
}
|
||||
|
||||
fn finish_cancelled_processing(inner: &Arc<Inner>, session_id: SessionId) -> bool {
|
||||
@@ -2252,6 +2279,54 @@ fn finish_cancelled_processing(inner: &Arc<Inner>, session_id: SessionId) -> boo
|
||||
finished
|
||||
}
|
||||
|
||||
pub(super) fn schedule_cancelled_asr_release(
|
||||
inner: &Arc<Inner>,
|
||||
asr: &ActiveAsr,
|
||||
session_id: SessionId,
|
||||
) {
|
||||
match asr {
|
||||
#[cfg(target_os = "windows")]
|
||||
ActiveAsr::FoundryLocalWhisper(_) => {
|
||||
schedule_foundry_local_asr_release(inner, AsrReleaseSession::Dictation(session_id));
|
||||
}
|
||||
#[cfg(target_os = "windows")]
|
||||
ActiveAsr::SherpaOnnxLocal(_) => {
|
||||
schedule_sherpa_onnx_release(inner, AsrReleaseSession::Dictation(session_id));
|
||||
}
|
||||
#[cfg(target_os = "macos")]
|
||||
ActiveAsr::Local(_) => {
|
||||
inner.local_asr_cache.touch();
|
||||
schedule_local_asr_release(inner);
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
/// end_session 转写阶段与「用户取消」赛跑的结果。
|
||||
enum TranscribeRace {
|
||||
Done(Result<RawTranscript, TranscribeFail>),
|
||||
/// 用户在 Processing（转写）阶段按 Esc / 取消：drop 掉在途 transcribe future。
|
||||
Cancelled,
|
||||
}
|
||||
|
||||
/// 轮询 Processing 阶段的取消标志。用户在转写阶段按 Esc 时，cancel_session 只把
|
||||
/// `state.cancelled` 置 true —— 此刻 ASR 句柄已被 end_session 从 `inner.asr` 槽 take 走，
|
||||
/// cancel_session 走的 `cancel_asr_for_session` 是 no-op，够不到在途请求。end_session 用
|
||||
/// 本函数与在途 transcribe future 赛跑：命中即 drop future，从而中断 reqwest HTTP /
|
||||
/// 停止等待流式最终结果 / 停止本地转写。
|
||||
///
|
||||
/// 用 75ms 轮询而非 notify：转写通常 0.2–3s，几次定时器唤醒的开销可忽略，用户也感知不到
|
||||
/// 这点延迟；换来的是不依赖任何唤醒信号、没有「取消边沿在注册 waiter 之前触发就丢失」的
|
||||
/// 竞态，逻辑上更稳。
|
||||
async fn wait_for_processing_cancel(inner: &Arc<Inner>) {
|
||||
loop {
|
||||
if inner.state.lock().cancelled {
|
||||
return;
|
||||
}
|
||||
tokio::time::sleep(std::time::Duration::from_millis(75)).await;
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) async fn end_session(inner: &Arc<Inner>) -> Result<(), String> {
|
||||
let current_session_id = {
|
||||
let mut state = inner.state.lock();
|
||||
@@ -2282,306 +2357,333 @@ pub(super) async fn end_session(inner: &Arc<Inner>) -> Result<(), String> {
|
||||
};
|
||||
|
||||
let uses_global_timeout = asr_transcribe_uses_global_timeout(&asr);
|
||||
// ASR 句柄内部是 Arc，clone 只是 +1 引用。留一份给取消路径：transcribe future 会把
|
||||
// `asr` move 进去，命中取消时那个 future 会被 drop（连同它持有的 Arc），我们再用这份
|
||||
// clone 显式 cancel，促使流式 WebSocket 立刻关闭、不残留后台 worker。
|
||||
let asr_for_cancel = asr.clone();
|
||||
// 每个引擎分支产出 Ok(RawTranscript) 或 Err(TranscribeFail)；失败/超时不再就地 return，
|
||||
// 而是把失败值交给 match 之后统一处理：先自动静默重试（从归档音频重转，应对网络/服务端
|
||||
// 瞬时抖动），重试拿回文本就当正常转写继续；彻底失败才 fail_dictation 保留录音 + 报错。
|
||||
let transcribe_outcome: Result<RawTranscript, TranscribeFail> = match asr {
|
||||
ActiveAsr::Volcengine(asr) => {
|
||||
debug_assert!(uses_global_timeout);
|
||||
if let Err(e) = asr.send_last_frame().await {
|
||||
log::error!("[coord] send last frame failed: {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)) => Ok(r),
|
||||
Ok(Err(e)) => {
|
||||
log::error!("[coord] await final failed: {e}");
|
||||
// 关闭 WebSocket 连接，避免流式 ASR 资源泄漏
|
||||
asr.cancel();
|
||||
Err(TranscribeFail::new(format!("识别失败: {e}"), e.to_string()))
|
||||
}
|
||||
Err(_) => {
|
||||
// 全局超时：最后的防线
|
||||
log::error!(
|
||||
"[coord] 全局超时 {} 秒 - 强制恢复",
|
||||
COORDINATOR_GLOBAL_TIMEOUT_SECS
|
||||
);
|
||||
// 清理 ASR session，避免资源泄漏
|
||||
asr.cancel();
|
||||
Err(TranscribeFail::new(
|
||||
"识别超时".to_string(),
|
||||
"global timeout".to_string(),
|
||||
))
|
||||
}
|
||||
}
|
||||
}
|
||||
ActiveAsr::Whisper(w) => {
|
||||
debug_assert!(uses_global_timeout);
|
||||
// Whisper / OpenRouter 动态超时：音频越长、分片越多，给更多
|
||||
// HTTP round-trip 预算。公式见 `whisper_transcribe_timeout`。
|
||||
let audio_secs = (w.buffer_duration_ms() as f64) / 1000.0;
|
||||
let timeout_duration = whisper_transcribe_timeout(audio_secs);
|
||||
log::info!(
|
||||
"[coord] Whisper transcribe: audio={:.2}s timeout={}s",
|
||||
audio_secs,
|
||||
timeout_duration.as_secs()
|
||||
);
|
||||
match tokio::time::timeout(timeout_duration, w.transcribe()).await {
|
||||
Ok(Ok(r)) => Ok(r),
|
||||
Ok(Err(e)) => {
|
||||
log::error!("[coord] whisper transcribe failed: {e}");
|
||||
Err(TranscribeFail::new(format!("识别失败: {e}"), e.to_string()))
|
||||
}
|
||||
Err(_) => {
|
||||
log::error!(
|
||||
"[coord] Whisper 动态超时 {}s（音频 {:.2}s）",
|
||||
timeout_duration.as_secs(),
|
||||
audio_secs
|
||||
);
|
||||
Err(TranscribeFail::new(
|
||||
"识别超时".to_string(),
|
||||
"whisper global timeout".to_string(),
|
||||
))
|
||||
}
|
||||
}
|
||||
}
|
||||
ActiveAsr::Mimo(m) => {
|
||||
debug_assert!(uses_global_timeout);
|
||||
let timeout_duration = std::time::Duration::from_secs(COORDINATOR_GLOBAL_TIMEOUT_SECS);
|
||||
match tokio::time::timeout(timeout_duration, m.transcribe()).await {
|
||||
Ok(Ok(r)) => Ok(r),
|
||||
Ok(Err(e)) => {
|
||||
log::error!("[coord] MiMo ASR transcribe failed: {e}");
|
||||
Err(TranscribeFail::new(format!("识别失败: {e}"), e.to_string()))
|
||||
}
|
||||
Err(_) => {
|
||||
log::error!(
|
||||
"[coord] MiMo ASR 全局超时 {} 秒",
|
||||
COORDINATOR_GLOBAL_TIMEOUT_SECS
|
||||
);
|
||||
Err(TranscribeFail::new(
|
||||
"识别超时".to_string(),
|
||||
"mimo global timeout".to_string(),
|
||||
))
|
||||
}
|
||||
}
|
||||
}
|
||||
ActiveAsr::Bailian(asr) => {
|
||||
debug_assert!(uses_global_timeout);
|
||||
if let Err(e) = asr.send_last_frame().await {
|
||||
log::error!("[coord] Bailian send last frame failed: {e}");
|
||||
}
|
||||
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)) => Ok(r),
|
||||
Ok(Err(e)) => {
|
||||
log::error!("[coord] Bailian await final failed: {e}");
|
||||
// 关闭 WebSocket 连接，避免流式 ASR 资源泄漏
|
||||
asr.cancel();
|
||||
Err(TranscribeFail::new(format!("识别失败: {e}"), e.to_string()))
|
||||
}
|
||||
Err(_) => {
|
||||
log::error!(
|
||||
"[coord] Bailian 全局超时 {} 秒",
|
||||
COORDINATOR_GLOBAL_TIMEOUT_SECS
|
||||
);
|
||||
asr.cancel();
|
||||
Err(TranscribeFail::new(
|
||||
"识别超时".to_string(),
|
||||
"bailian global timeout".to_string(),
|
||||
))
|
||||
}
|
||||
}
|
||||
}
|
||||
ActiveAsr::Qwen3Realtime(asr) => {
|
||||
debug_assert!(uses_global_timeout);
|
||||
if let Err(e) = asr.send_last_frame().await {
|
||||
log::error!("[coord] Qwen3 realtime send last frame failed: {e}");
|
||||
}
|
||||
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)) => Ok(r),
|
||||
Ok(Err(e)) => {
|
||||
log::error!("[coord] Qwen3 realtime await final failed: {e}");
|
||||
// 关闭 WebSocket 连接，避免流式 ASR 资源泄漏
|
||||
asr.cancel();
|
||||
Err(TranscribeFail::new(format!("识别失败: {e}"), e.to_string()))
|
||||
}
|
||||
Err(_) => {
|
||||
log::error!(
|
||||
"[coord] Qwen3 realtime 全局超时 {} 秒",
|
||||
COORDINATOR_GLOBAL_TIMEOUT_SECS
|
||||
);
|
||||
asr.cancel();
|
||||
Err(TranscribeFail::new(
|
||||
"识别超时".to_string(),
|
||||
"qwen3 realtime global timeout".to_string(),
|
||||
))
|
||||
}
|
||||
}
|
||||
}
|
||||
#[cfg(target_os = "windows")]
|
||||
ActiveAsr::FoundryLocalWhisper(local) => {
|
||||
debug_assert!(!uses_global_timeout);
|
||||
let audio_secs = (local.buffer_duration_ms() as f64) / 1000.0;
|
||||
let timeout_duration = windows_local_asr_transcribe_timeout(audio_secs);
|
||||
log::info!(
|
||||
"[coord] Foundry Local Whisper transcribe: audio={:.2}s timeout={}s",
|
||||
audio_secs,
|
||||
timeout_duration.as_secs()
|
||||
);
|
||||
match local.transcribe(timeout_duration).await {
|
||||
Ok(r) => {
|
||||
schedule_foundry_local_asr_release(
|
||||
inner,
|
||||
AsrReleaseSession::Dictation(current_session_id),
|
||||
);
|
||||
Ok(r)
|
||||
}
|
||||
Err(e) => {
|
||||
if inner.state.lock().cancelled {
|
||||
log::info!(
|
||||
"[coord] Foundry Local Whisper transcribe cancelled — discarding transcript"
|
||||
);
|
||||
schedule_foundry_local_asr_release(
|
||||
inner,
|
||||
AsrReleaseSession::Dictation(current_session_id),
|
||||
);
|
||||
restore_prepared_windows_ime_session(inner, current_session_id);
|
||||
finish_cancelled_processing(inner, current_session_id);
|
||||
return Ok(());
|
||||
//
|
||||
// 整段转写与「用户在 Processing 阶段取消」赛跑：命中取消就直接 drop 掉 transcribe future
|
||||
// 中断在途请求，不再傻等它跑完（见 issue「转写中按 Esc 停不下来」）。
|
||||
let raced: TranscribeRace = {
|
||||
let transcribe_fut = async move {
|
||||
let transcribe_outcome: Result<RawTranscript, TranscribeFail> = match asr {
|
||||
ActiveAsr::Volcengine(asr) => {
|
||||
debug_assert!(uses_global_timeout);
|
||||
if let Err(e) = asr.send_last_frame().await {
|
||||
log::error!("[coord] send last frame failed: {e}");
|
||||
}
|
||||
log::error!("[coord] Foundry Local Whisper transcribe failed: {e:#}");
|
||||
schedule_foundry_local_asr_release(
|
||||
inner,
|
||||
AsrReleaseSession::Dictation(current_session_id),
|
||||
);
|
||||
Err(TranscribeFail::new(format!("本地识别失败: {e}"), e.to_string()))
|
||||
}
|
||||
}
|
||||
}
|
||||
// Windows sherpa-onnx offline batch：停止录音后整段转写，再复用现有
|
||||
// polish / insert / history 收尾路径。
|
||||
#[cfg(target_os = "windows")]
|
||||
ActiveAsr::SherpaOnnxLocal(local) => {
|
||||
debug_assert!(!uses_global_timeout);
|
||||
let audio_secs = (local.buffer_duration_ms() as f64) / 1000.0;
|
||||
let timeout_duration = windows_local_asr_transcribe_timeout(audio_secs);
|
||||
log::info!(
|
||||
"[coord] sherpa-onnx transcribe: audio={:.2}s timeout={}s",
|
||||
audio_secs,
|
||||
timeout_duration.as_secs()
|
||||
);
|
||||
match local.transcribe(timeout_duration).await {
|
||||
Ok(r) => {
|
||||
schedule_sherpa_onnx_release(
|
||||
inner,
|
||||
AsrReleaseSession::Dictation(current_session_id),
|
||||
);
|
||||
Ok(r)
|
||||
}
|
||||
Err(e) => {
|
||||
if inner.state.lock().cancelled {
|
||||
log::info!(
|
||||
"[coord] sherpa-onnx transcribe cancelled — discarding transcript"
|
||||
);
|
||||
schedule_sherpa_onnx_release(
|
||||
inner,
|
||||
AsrReleaseSession::Dictation(current_session_id),
|
||||
);
|
||||
restore_prepared_windows_ime_session(inner, current_session_id);
|
||||
finish_cancelled_processing(inner, current_session_id);
|
||||
return Ok(());
|
||||
// 添加全局超时保护：防止 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)) => Ok(r),
|
||||
Ok(Err(e)) => {
|
||||
log::error!("[coord] await final failed: {e}");
|
||||
// 关闭 WebSocket 连接，避免流式 ASR 资源泄漏
|
||||
asr.cancel();
|
||||
Err(TranscribeFail::new(format!("识别失败: {e}"), e.to_string()))
|
||||
}
|
||||
Err(_) => {
|
||||
// 全局超时：最后的防线
|
||||
log::error!(
|
||||
"[coord] 全局超时 {} 秒 - 强制恢复",
|
||||
COORDINATOR_GLOBAL_TIMEOUT_SECS
|
||||
);
|
||||
// 清理 ASR session，避免资源泄漏
|
||||
asr.cancel();
|
||||
Err(TranscribeFail::new(
|
||||
"识别超时".to_string(),
|
||||
"global timeout".to_string(),
|
||||
))
|
||||
}
|
||||
}
|
||||
log::error!("[coord] sherpa-onnx transcribe failed: {e:#}");
|
||||
schedule_sherpa_onnx_release(
|
||||
inner,
|
||||
AsrReleaseSession::Dictation(current_session_id),
|
||||
}
|
||||
ActiveAsr::Whisper(w) => {
|
||||
debug_assert!(uses_global_timeout);
|
||||
// Whisper / OpenRouter 动态超时：音频越长、分片越多，给更多
|
||||
// HTTP round-trip 预算。公式见 `whisper_transcribe_timeout`。
|
||||
let audio_secs = (w.buffer_duration_ms() as f64) / 1000.0;
|
||||
let timeout_duration = whisper_transcribe_timeout(audio_secs);
|
||||
log::info!(
|
||||
"[coord] Whisper transcribe: audio={:.2}s timeout={}s",
|
||||
audio_secs,
|
||||
timeout_duration.as_secs()
|
||||
);
|
||||
Err(TranscribeFail::new(format!("本地识别失败: {e}"), e.to_string()))
|
||||
}
|
||||
}
|
||||
}
|
||||
#[cfg(target_os = "macos")]
|
||||
ActiveAsr::Local(local) => {
|
||||
debug_assert!(uses_global_timeout);
|
||||
// 缓存命中时 transcribe 不含 load 时间；冷启动 load 已在 build_local_qwen3
|
||||
// 提前完成。但 transcribe 本身受音频长度影响：用户实测 RTF ≈ 0.3，慢机
|
||||
// 可达 0.5；15s 固定超时在 ≥ 30s 录音上会把整段结果丢掉。改用动态
|
||||
// 超时 max(15, ceil(audio_s × 0.6) + 10)，公式与单测见
|
||||
// `local_qwen_transcribe_timeout`。
|
||||
let audio_secs = (local.buffer_duration_ms() as f64) / 1000.0;
|
||||
let timeout_duration = local_qwen_transcribe_timeout(audio_secs);
|
||||
log::info!(
|
||||
"[coord] local Qwen3-ASR transcribe: audio={:.2}s timeout={}s",
|
||||
audio_secs,
|
||||
timeout_duration.as_secs()
|
||||
);
|
||||
let result = tokio::time::timeout(timeout_duration, local.transcribe()).await;
|
||||
inner.local_asr_cache.touch();
|
||||
schedule_local_asr_release(inner);
|
||||
match result {
|
||||
Ok(Ok(r)) => Ok(r),
|
||||
Ok(Err(e)) => {
|
||||
log::error!("[coord] local Qwen3-ASR transcribe failed: {e:#}");
|
||||
Err(TranscribeFail::new(format!("本地识别失败: {e}"), e.to_string()))
|
||||
}
|
||||
Err(_) => {
|
||||
log::error!(
|
||||
"[coord] local Qwen3-ASR 动态超时 {}s（音频 {:.2}s）",
|
||||
timeout_duration.as_secs(),
|
||||
audio_secs
|
||||
);
|
||||
Err(TranscribeFail::new(
|
||||
"识别超时".to_string(),
|
||||
"local global timeout".to_string(),
|
||||
))
|
||||
}
|
||||
}
|
||||
}
|
||||
// Apple Speech：系统语音识别，无模型加载耗时。批处理 transcribe 受音频
|
||||
// 长度影响，沿用 local_qwen_transcribe_timeout 的动态超时公式。
|
||||
#[cfg(target_os = "macos")]
|
||||
ActiveAsr::AppleSpeech(local) => {
|
||||
debug_assert!(uses_global_timeout);
|
||||
let audio_secs = (local.buffer_duration_ms() as f64) / 1000.0;
|
||||
let timeout_duration = local_qwen_transcribe_timeout(audio_secs);
|
||||
log::info!(
|
||||
"[coord] Apple Speech transcribe: audio={:.2}s timeout={}s",
|
||||
audio_secs,
|
||||
timeout_duration.as_secs()
|
||||
);
|
||||
match tokio::time::timeout(timeout_duration, local.transcribe()).await {
|
||||
Ok(Ok(r)) => Ok(r),
|
||||
Ok(Err(e)) => {
|
||||
if inner.state.lock().cancelled {
|
||||
log::info!(
|
||||
"[coord] Apple Speech transcribe cancelled - discarding transcript"
|
||||
);
|
||||
restore_prepared_windows_ime_session(inner, current_session_id);
|
||||
finish_cancelled_processing(inner, current_session_id);
|
||||
return Ok(());
|
||||
match tokio::time::timeout(timeout_duration, w.transcribe()).await {
|
||||
Ok(Ok(r)) => Ok(r),
|
||||
Ok(Err(e)) => {
|
||||
log::error!("[coord] whisper transcribe failed: {e}");
|
||||
Err(TranscribeFail::new(format!("识别失败: {e}"), e.to_string()))
|
||||
}
|
||||
Err(_) => {
|
||||
log::error!(
|
||||
"[coord] Whisper 动态超时 {}s（音频 {:.2}s）",
|
||||
timeout_duration.as_secs(),
|
||||
audio_secs
|
||||
);
|
||||
Err(TranscribeFail::new(
|
||||
"识别超时".to_string(),
|
||||
"whisper global timeout".to_string(),
|
||||
))
|
||||
}
|
||||
}
|
||||
log::error!("[coord] Apple Speech transcribe failed: {e:#}");
|
||||
Err(TranscribeFail::new(format!("本地识别失败: {e}"), e.to_string()))
|
||||
}
|
||||
Err(_) => {
|
||||
log::error!(
|
||||
"[coord] Apple Speech 动态超时 {}s（音频 {:.2}s）",
|
||||
timeout_duration.as_secs(),
|
||||
audio_secs
|
||||
ActiveAsr::Mimo(m) => {
|
||||
debug_assert!(uses_global_timeout);
|
||||
let timeout_duration =
|
||||
std::time::Duration::from_secs(COORDINATOR_GLOBAL_TIMEOUT_SECS);
|
||||
match tokio::time::timeout(timeout_duration, m.transcribe()).await {
|
||||
Ok(Ok(r)) => Ok(r),
|
||||
Ok(Err(e)) => {
|
||||
log::error!("[coord] MiMo ASR transcribe failed: {e}");
|
||||
Err(TranscribeFail::new(format!("识别失败: {e}"), e.to_string()))
|
||||
}
|
||||
Err(_) => {
|
||||
log::error!(
|
||||
"[coord] MiMo ASR 全局超时 {} 秒",
|
||||
COORDINATOR_GLOBAL_TIMEOUT_SECS
|
||||
);
|
||||
Err(TranscribeFail::new(
|
||||
"识别超时".to_string(),
|
||||
"mimo global timeout".to_string(),
|
||||
))
|
||||
}
|
||||
}
|
||||
}
|
||||
ActiveAsr::Bailian(asr) => {
|
||||
debug_assert!(uses_global_timeout);
|
||||
if let Err(e) = asr.send_last_frame().await {
|
||||
log::error!("[coord] Bailian send last frame failed: {e}");
|
||||
}
|
||||
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)) => Ok(r),
|
||||
Ok(Err(e)) => {
|
||||
log::error!("[coord] Bailian await final failed: {e}");
|
||||
// 关闭 WebSocket 连接，避免流式 ASR 资源泄漏
|
||||
asr.cancel();
|
||||
Err(TranscribeFail::new(format!("识别失败: {e}"), e.to_string()))
|
||||
}
|
||||
Err(_) => {
|
||||
log::error!(
|
||||
"[coord] Bailian 全局超时 {} 秒",
|
||||
COORDINATOR_GLOBAL_TIMEOUT_SECS
|
||||
);
|
||||
asr.cancel();
|
||||
Err(TranscribeFail::new(
|
||||
"识别超时".to_string(),
|
||||
"bailian global timeout".to_string(),
|
||||
))
|
||||
}
|
||||
}
|
||||
}
|
||||
ActiveAsr::Qwen3Realtime(asr) => {
|
||||
debug_assert!(uses_global_timeout);
|
||||
if let Err(e) = asr.send_last_frame().await {
|
||||
log::error!("[coord] Qwen3 realtime send last frame failed: {e}");
|
||||
}
|
||||
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)) => Ok(r),
|
||||
Ok(Err(e)) => {
|
||||
log::error!("[coord] Qwen3 realtime await final failed: {e}");
|
||||
// 关闭 WebSocket 连接，避免流式 ASR 资源泄漏
|
||||
asr.cancel();
|
||||
Err(TranscribeFail::new(format!("识别失败: {e}"), e.to_string()))
|
||||
}
|
||||
Err(_) => {
|
||||
log::error!(
|
||||
"[coord] Qwen3 realtime 全局超时 {} 秒",
|
||||
COORDINATOR_GLOBAL_TIMEOUT_SECS
|
||||
);
|
||||
asr.cancel();
|
||||
Err(TranscribeFail::new(
|
||||
"识别超时".to_string(),
|
||||
"qwen3 realtime global timeout".to_string(),
|
||||
))
|
||||
}
|
||||
}
|
||||
}
|
||||
#[cfg(target_os = "windows")]
|
||||
ActiveAsr::FoundryLocalWhisper(local) => {
|
||||
debug_assert!(!uses_global_timeout);
|
||||
let audio_secs = (local.buffer_duration_ms() as f64) / 1000.0;
|
||||
let timeout_duration = windows_local_asr_transcribe_timeout(audio_secs);
|
||||
log::info!(
|
||||
"[coord] Foundry Local Whisper transcribe: audio={:.2}s timeout={}s",
|
||||
audio_secs,
|
||||
timeout_duration.as_secs()
|
||||
);
|
||||
Err(TranscribeFail::new(
|
||||
"识别超时".to_string(),
|
||||
"apple-speech global timeout".to_string(),
|
||||
))
|
||||
match local.transcribe(timeout_duration).await {
|
||||
Ok(r) => {
|
||||
schedule_foundry_local_asr_release(
|
||||
inner,
|
||||
AsrReleaseSession::Dictation(current_session_id),
|
||||
);
|
||||
Ok(r)
|
||||
}
|
||||
Err(e) => {
|
||||
// 用户取消现在由外层 select! 统一处理（drop 掉本 future 中断在途转写），
|
||||
// 到这里的 Err 一律当作真失败：调度引擎释放 + 交给 match 后的重试/报错。
|
||||
log::error!("[coord] Foundry Local Whisper transcribe failed: {e:#}");
|
||||
schedule_foundry_local_asr_release(
|
||||
inner,
|
||||
AsrReleaseSession::Dictation(current_session_id),
|
||||
);
|
||||
Err(TranscribeFail::new(
|
||||
format!("本地识别失败: {e}"),
|
||||
e.to_string(),
|
||||
))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
// Windows sherpa-onnx offline batch：停止录音后整段转写，再复用现有
|
||||
// polish / insert / history 收尾路径。
|
||||
#[cfg(target_os = "windows")]
|
||||
ActiveAsr::SherpaOnnxLocal(local) => {
|
||||
debug_assert!(!uses_global_timeout);
|
||||
let audio_secs = (local.buffer_duration_ms() as f64) / 1000.0;
|
||||
let timeout_duration = windows_local_asr_transcribe_timeout(audio_secs);
|
||||
log::info!(
|
||||
"[coord] sherpa-onnx transcribe: audio={:.2}s timeout={}s",
|
||||
audio_secs,
|
||||
timeout_duration.as_secs()
|
||||
);
|
||||
match local.transcribe(timeout_duration).await {
|
||||
Ok(r) => {
|
||||
schedule_sherpa_onnx_release(
|
||||
inner,
|
||||
AsrReleaseSession::Dictation(current_session_id),
|
||||
);
|
||||
Ok(r)
|
||||
}
|
||||
Err(e) => {
|
||||
// 取消由外层 select! 统一处理，见 Foundry 分支同款注释。
|
||||
log::error!("[coord] sherpa-onnx transcribe failed: {e:#}");
|
||||
schedule_sherpa_onnx_release(
|
||||
inner,
|
||||
AsrReleaseSession::Dictation(current_session_id),
|
||||
);
|
||||
Err(TranscribeFail::new(
|
||||
format!("本地识别失败: {e}"),
|
||||
e.to_string(),
|
||||
))
|
||||
}
|
||||
}
|
||||
}
|
||||
#[cfg(target_os = "macos")]
|
||||
ActiveAsr::Local(local) => {
|
||||
debug_assert!(uses_global_timeout);
|
||||
// 缓存命中时 transcribe 不含 load 时间；冷启动 load 已在 build_local_qwen3
|
||||
// 提前完成。但 transcribe 本身受音频长度影响：用户实测 RTF ≈ 0.3，慢机
|
||||
// 可达 0.5；15s 固定超时在 ≥ 30s 录音上会把整段结果丢掉。改用动态
|
||||
// 超时 max(15, ceil(audio_s × 0.6) + 10)，公式与单测见
|
||||
// `local_qwen_transcribe_timeout`。
|
||||
let audio_secs = (local.buffer_duration_ms() as f64) / 1000.0;
|
||||
let timeout_duration = local_qwen_transcribe_timeout(audio_secs);
|
||||
log::info!(
|
||||
"[coord] local Qwen3-ASR transcribe: audio={:.2}s timeout={}s",
|
||||
audio_secs,
|
||||
timeout_duration.as_secs()
|
||||
);
|
||||
let result = tokio::time::timeout(timeout_duration, local.transcribe()).await;
|
||||
inner.local_asr_cache.touch();
|
||||
schedule_local_asr_release(inner);
|
||||
match result {
|
||||
Ok(Ok(r)) => Ok(r),
|
||||
Ok(Err(e)) => {
|
||||
log::error!("[coord] local Qwen3-ASR transcribe failed: {e:#}");
|
||||
Err(TranscribeFail::new(
|
||||
format!("本地识别失败: {e}"),
|
||||
e.to_string(),
|
||||
))
|
||||
}
|
||||
Err(_) => {
|
||||
log::error!(
|
||||
"[coord] local Qwen3-ASR 动态超时 {}s（音频 {:.2}s）",
|
||||
timeout_duration.as_secs(),
|
||||
audio_secs
|
||||
);
|
||||
Err(TranscribeFail::new(
|
||||
"识别超时".to_string(),
|
||||
"local global timeout".to_string(),
|
||||
))
|
||||
}
|
||||
}
|
||||
}
|
||||
// Apple Speech：系统语音识别，无模型加载耗时。批处理 transcribe 受音频
|
||||
// 长度影响，沿用 local_qwen_transcribe_timeout 的动态超时公式。
|
||||
#[cfg(target_os = "macos")]
|
||||
ActiveAsr::AppleSpeech(local) => {
|
||||
debug_assert!(uses_global_timeout);
|
||||
let audio_secs = (local.buffer_duration_ms() as f64) / 1000.0;
|
||||
let timeout_duration = local_qwen_transcribe_timeout(audio_secs);
|
||||
log::info!(
|
||||
"[coord] Apple Speech transcribe: audio={:.2}s timeout={}s",
|
||||
audio_secs,
|
||||
timeout_duration.as_secs()
|
||||
);
|
||||
match tokio::time::timeout(timeout_duration, local.transcribe()).await {
|
||||
Ok(Ok(r)) => Ok(r),
|
||||
Ok(Err(e)) => {
|
||||
// 取消由外层 select! 统一处理，见 Foundry 分支同款注释。
|
||||
log::error!("[coord] Apple Speech transcribe failed: {e:#}");
|
||||
Err(TranscribeFail::new(
|
||||
format!("本地识别失败: {e}"),
|
||||
e.to_string(),
|
||||
))
|
||||
}
|
||||
Err(_) => {
|
||||
log::error!(
|
||||
"[coord] Apple Speech 动态超时 {}s（音频 {:.2}s）",
|
||||
timeout_duration.as_secs(),
|
||||
audio_secs
|
||||
);
|
||||
Err(TranscribeFail::new(
|
||||
"识别超时".to_string(),
|
||||
"apple-speech global timeout".to_string(),
|
||||
))
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
transcribe_outcome
|
||||
};
|
||||
tokio::select! {
|
||||
// biased：每次先查取消标志，取消优先于「转写恰好同时完成」。
|
||||
biased;
|
||||
_ = wait_for_processing_cancel(inner) => TranscribeRace::Cancelled,
|
||||
outcome = transcribe_fut => TranscribeRace::Done(outcome),
|
||||
}
|
||||
};
|
||||
|
||||
// ASR 完成后 cancel 检查：用户在 transcribe 进行中按 Esc 时，这里就会命中。
|
||||
let transcribe_outcome: Result<RawTranscript, TranscribeFail> = match raced {
|
||||
TranscribeRace::Cancelled => {
|
||||
log::info!("[coord] cancel during transcribe — 中断在途 ASR 请求，丢弃转写");
|
||||
// 上面 select! 已把 transcribe_fut drop 掉（中断 reqwest / 停止等待流式结果 /
|
||||
// 停止本地转写）；这里再显式 cancel 一次，促使流式 WebSocket 立即关闭、不残留
|
||||
// 后台 worker。asr_for_cancel 与被 drop 的 future 共享同一 Arc 底层。
|
||||
let asr_for_release = asr_for_cancel.clone();
|
||||
cancel_active_asr(asr_for_cancel);
|
||||
// end_session 已经把 ASR 从 inner.asr 取走，cancel_session 无法再触发
|
||||
// provider 的释放调度；取消路径必须自己补上，否则本地模型会一直占用缓存。
|
||||
schedule_cancelled_asr_release(inner, &asr_for_release, current_session_id);
|
||||
restore_prepared_windows_ime_session(inner, current_session_id);
|
||||
// 与下方「ASR 完成后 cancel 检查」同款收尾（finish_cancelled_processing 负责
|
||||
// 把 phase 收回 Idle、清 focus_target）。
|
||||
finish_cancelled_processing(inner, current_session_id);
|
||||
return Ok(());
|
||||
}
|
||||
TranscribeRace::Done(outcome) => outcome,
|
||||
};
|
||||
|
||||
// ASR 完成后 cancel 检查：转写恰好跑完、用户几乎同时按 Esc（select! 走了 Done 分支）时
|
||||
// 这里兜底命中。上面赛跑分支处理的是「转写还在途中」的取消。
|
||||
// 优先级高于 empty 检查 — 用户取消 → 静默丢弃，不写失败历史也不弹错误胶囊。
|
||||
if inner.state.lock().cancelled {
|
||||
log::info!("[coord] cancel detected after ASR — discarding transcript");
|
||||
@@ -2600,15 +2702,25 @@ pub(super) async fn end_session(inner: &Arc<Inner>) -> Result<(), String> {
|
||||
let raw = match transcribe_outcome {
|
||||
Ok(raw) => raw,
|
||||
Err(fail) => match try_silent_retranscribe(inner, current_session_id).await {
|
||||
Some(raw) => raw,
|
||||
None => {
|
||||
return fail_dictation(
|
||||
inner,
|
||||
current_session_id,
|
||||
elapsed,
|
||||
fail.user_msg,
|
||||
fail.err,
|
||||
)
|
||||
SilentRetryOutcome::Transcript(raw) => raw,
|
||||
SilentRetryOutcome::Cancelled => {
|
||||
log::info!("[coord] cancel during silent ASR retry — discarding transcript");
|
||||
restore_prepared_windows_ime_session(inner, current_session_id);
|
||||
finish_cancelled_processing(inner, current_session_id);
|
||||
return Ok(());
|
||||
}
|
||||
SilentRetryOutcome::Exhausted => {
|
||||
// 处理最后一次重试结果时也复查一次取消标志，覆盖「重试刚返回
|
||||
// Exhausted 与用户同时按 Esc」的窄竞态，避免误走失败提示。
|
||||
if inner.state.lock().cancelled {
|
||||
log::info!(
|
||||
"[coord] cancel after silent ASR retry — discarding transcript"
|
||||
);
|
||||
restore_prepared_windows_ime_session(inner, current_session_id);
|
||||
finish_cancelled_processing(inner, current_session_id);
|
||||
return Ok(());
|
||||
}
|
||||
return fail_dictation(inner, current_session_id, elapsed, fail.user_msg, fail.err)
|
||||
}
|
||||
},
|
||||
};
|
||||
|
||||
Reference in new issue
Block a user