From ab106923af2813c1203ec9828c215843c133cff6 Mon Sep 17 00:00:00 2001 From: jisongniu <529058747@qq.com> Date: Fri, 10 Jul 2026 10:40:49 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat(asr):=20=E6=96=B0=E5=A2=9E=E7=99=BE?= =?UTF-8?q?=E7=82=BC=20Qwen3-ASR-Flash=20=E5=AE=9E=E6=97=B6=E4=BE=9B?= =?UTF-8?q?=E5=BA=94=E5=95=86=EF=BC=88OpenAI=20Realtime=20WS=20=E5=8D=8F?= =?UTF-8?q?=E8=AE=AE=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit bailian.rs 头注释预留的 follow-up 协议线：/api-ws/v1/realtime，与经典 /api-ws/v1/inference 并列为独立供应商 bailian-qwen3-realtime。 - asr/qwen_realtime.rs：新客户端。base64 JSON 事件流 + server_vad 自动断句， 按序收集 completed 句段；session.finish 冲刷尾段后回 session.finished。 关键协议行为均经 2026-07 线上实测：尾段冲刷（说到一半松手不丢字）、 纯静音验证干净返回、language 省略走自动检测、业务空间专属域名 （wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com）同样承载此路径。 - coordinator / asr_wiring / dictation / qa_session / resources：与 Bailian 分支同构接线（流式 WS + DeferredAsrBridge）。 - 验证与拉取模型：真实 WS 连通性检查；协议头校验（qwen3EndpointSchemeInvalid）。 - 前端：ASR 预设下拉新增一项；i18n 5 语言。 测试：cargo test 573 passed（新增 18 个客户端单测）；npm run build 通过。 --- openless-all/app/src-tauri/src/asr/mod.rs | 2 + .../app/src-tauri/src/asr/qwen_realtime.rs | 802 ++++++++++++++++++ .../app/src-tauri/src/commands/credentials.rs | 4 +- .../app/src-tauri/src/commands/providers.rs | 56 ++ openless-all/app/src-tauri/src/coordinator.rs | 43 +- .../src-tauri/src/coordinator/asr_wiring.rs | 24 + .../src-tauri/src/coordinator/dictation.rs | 104 +++ .../src-tauri/src/coordinator/qa_session.rs | 39 + .../src-tauri/src/coordinator/resources.rs | 1 + openless-all/app/src/i18n/en.ts | 2 + openless-all/app/src/i18n/ja.ts | 2 + openless-all/app/src/i18n/ko.ts | 2 + openless-all/app/src/i18n/zh-CN.ts | 2 + openless-all/app/src/i18n/zh-TW.ts | 2 + .../src/pages/settings/ProvidersSection.tsx | 5 + .../app/src/pages/settings/shared.tsx | 1 + 16 files changed, 1088 insertions(+), 3 deletions(-) create mode 100644 openless-all/app/src-tauri/src/asr/qwen_realtime.rs diff --git a/openless-all/app/src-tauri/src/asr/mod.rs b/openless-all/app/src-tauri/src/asr/mod.rs index e1c761c4..4eb94097 100644 --- a/openless-all/app/src-tauri/src/asr/mod.rs +++ b/openless-all/app/src-tauri/src/asr/mod.rs @@ -10,12 +10,14 @@ mod frame; pub mod local; pub mod mimo; pub mod pcm; +pub mod qwen_realtime; pub mod volcengine; pub mod wav; pub mod whisper; pub use bailian::{BailianCredentials, BailianRealtimeASR}; pub use mimo::MimoBatchASR; +pub use qwen_realtime::{Qwen3RealtimeASR, Qwen3RealtimeCredentials}; pub use volcengine::{VolcengineCredentials, VolcengineStreamingASR}; pub use whisper::WhisperBatchASR; diff --git a/openless-all/app/src-tauri/src/asr/qwen_realtime.rs b/openless-all/app/src-tauri/src/asr/qwen_realtime.rs new file mode 100644 index 00000000..f3d95e82 --- /dev/null +++ b/openless-all/app/src-tauri/src/asr/qwen_realtime.rs @@ -0,0 +1,802 @@ +//! Alibaba Cloud Bailian Qwen3-ASR-Flash realtime client. +//! +//! Speaks the OpenAI Realtime-style WebSocket protocol +//! (`/api-ws/v1/realtime?model=...`) — the protocol line `bailian.rs` left as +//! a follow-up. 与经典 `/api-ws/v1/inference` 不同：音频以 base64 JSON 事件 +//! （`input_audio_buffer.append`）发送，服务端以 `server_vad` 自动断句，每句 +//! 产生一个 `conversation.item.input_audio_transcription.completed`。 +//! +//! 2026-07 线上实测确认的关键行为： +//! - `session.finish` 会先冲刷 VAD 尚未关闭的尾段（补发 completed）再回 +//! `session.finished`，说到一半松手不会丢尾巴； +//! - 纯静音 + finish 正常返回 `session.finished`（连接检查可用，无经典协议 +//! 的 EmptyAudio 问题）； +//! - `session.update` 省略 `input_audio_transcription.language` 时自动检测语种； +//! - 业务空间专属域名（`wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com`） +//! 同样承载此路径，经典 inference 路径则只在公共网关。 + +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use base64::Engine; +use futures_util::{SinkExt, StreamExt}; +use parking_lot::Mutex as ParkingMutex; +use serde_json::{json, Value}; +use tokio::net::TcpStream; +use tokio::runtime::Handle; +use tokio::sync::{mpsc, oneshot, Mutex as AsyncMutex, Notify}; +use tokio_tungstenite::tungstenite::client::IntoClientRequest; +use tokio_tungstenite::tungstenite::http::header::HeaderValue; +use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream}; + +use super::{AudioConsumer, RawTranscript}; + +pub const PROVIDER_ID: &str = "bailian-qwen3-realtime"; +pub const DEFAULT_ENDPOINT: &str = "wss://dashscope.aliyuncs.com/api-ws/v1/realtime"; +pub const DEFAULT_MODEL: &str = "qwen3-asr-flash-realtime"; + +/// 100 ms of 16 kHz / 16-bit / mono PCM，与 recorder 输出及官方示例一致。 +pub const TARGET_AUDIO_CHUNK_BYTES: usize = 3_200; +const BYTES_PER_MS: u64 = 32; +const FINAL_RESULT_TIMEOUT: Duration = Duration::from_secs(12); +/// server_vad 断句静默阈值。官方默认 400ms；取 500ms 降低说话中途换气被切断的概率。 +const VAD_SILENCE_DURATION_MS: u32 = 500; + +type WsStream = WebSocketStream>; +type WsSink = futures_util::stream::SplitSink; +type SharedWriter = Arc>>; + +#[derive(Clone, Debug)] +pub struct Qwen3RealtimeCredentials { + pub api_key: String, + pub endpoint: String, + pub model: String, +} + +impl Qwen3RealtimeCredentials { + pub fn normalized_endpoint(&self) -> String { + if self.endpoint.trim().is_empty() { + return DEFAULT_ENDPOINT.to_string(); + } + self.endpoint.trim().to_string() + } + + pub fn normalized_model(&self) -> String { + let model = self.model.trim(); + if model.is_empty() { + DEFAULT_MODEL.to_string() + } else { + model.to_string() + } + } + + /// 连接 URL：`{endpoint}?model={model}`。用户若已在 endpoint 里带了 + /// `model=` 查询参数则原样使用，不重复拼接。 + pub fn connect_url(&self) -> String { + let endpoint = self.normalized_endpoint(); + if endpoint.contains("model=") { + return endpoint; + } + let sep = if endpoint.contains('?') { '&' } else { '?' }; + format!( + "{}{}model={}", + endpoint.trim_end_matches('/'), + sep, + self.normalized_model() + ) + } +} + +#[derive(Debug, thiserror::Error)] +pub enum Qwen3ASRError { + #[error("credentials missing")] + CredentialsMissing, + #[error("connection failed: {0}")] + ConnectionFailed(String), + #[error("send failed: {0}")] + SendFailed(String), + #[error("task failed: {0}")] + TaskFailed(String), + #[error("no final result")] + NoFinalResult, + #[error("final result timed out")] + FinalResultTimeout, +} + +enum SendItem { + Audio(Vec), + Finish(oneshot::Sender>), +} + +#[derive(Default)] +struct SyncState { + pending_audio: Vec, + audio_scratch: Vec, + bytes_received: u64, + session_started: bool, + session_finished: bool, + runtime: Option, + start: Option, + final_tx: Option>>, + send_tx: Option>, + /// VAD 断句后按到达顺序累积的已完成句段（completed.transcript）。 + completed_segments: Vec, + /// 当前未完成句段的最新 interim 文本；completed 到达后清空。 + /// 服务端在句段开放期把累积文本放 `stash`、精修期放 `text`，取非空者。 + partial_text: String, +} + +pub struct Qwen3RealtimeASR { + credentials: Qwen3RealtimeCredentials, + state: ParkingMutex, + writer: SharedWriter, + final_rx: ParkingMutex>>>, + session_started: Arc, +} + +impl Qwen3RealtimeASR { + pub fn new(credentials: Qwen3RealtimeCredentials) -> Self { + Self { + credentials, + state: ParkingMutex::new(SyncState::default()), + writer: Arc::new(AsyncMutex::new(None)), + final_rx: ParkingMutex::new(None), + session_started: Arc::new(Notify::new()), + } + } + + pub async fn open_session(self: &Arc) -> Result<(), Qwen3ASRError> { + if self.credentials.api_key.trim().is_empty() { + return Err(Qwen3ASRError::CredentialsMissing); + } + + let url = self.credentials.connect_url(); + let mut request = url + .into_client_request() + .map_err(|e| Qwen3ASRError::ConnectionFailed(e.to_string()))?; + request.headers_mut().insert( + "Authorization", + HeaderValue::from_str(&format!("Bearer {}", self.credentials.api_key.trim())) + .map_err(|e| Qwen3ASRError::ConnectionFailed(e.to_string()))?, + ); + request.headers_mut().insert( + "OpenAI-Beta", + HeaderValue::from_static("realtime=v1"), + ); + + let (ws, _resp) = connect_async(request) + .await + .map_err(|e| Qwen3ASRError::ConnectionFailed(e.to_string()))?; + let (write, read) = ws.split(); + *self.writer.lock().await = Some(write); + + let (final_tx, final_rx) = oneshot::channel(); + let (send_tx, mut send_rx) = mpsc::unbounded_channel::(); + { + let mut st = self.state.lock(); + *st = SyncState::default(); + st.runtime = Some(Handle::current()); + st.start = Some(Instant::now()); + st.final_tx = Some(final_tx); + st.send_tx = Some(send_tx); + } + *self.final_rx.lock() = Some(final_rx); + + let writer_for_worker = Arc::clone(&self.writer); + tokio::spawn(async move { + while let Some(item) = send_rx.recv().await { + match item { + SendItem::Audio(chunk) => { + if let Err(e) = + send_text(&writer_for_worker, append_audio_message(&chunk)).await + { + log::error!("[qwen3-asr] audio frame send failed: {e}"); + } + } + SendItem::Finish(done) => { + let result = send_text(&writer_for_worker, finish_session_message()) + .await + .map_err(|e| Qwen3ASRError::SendFailed(e.to_string())); + let _ = done.send(result); + } + } + } + }); + + send_text(&self.writer, session_update_message()).await?; + + let weak_self = Arc::downgrade(self); + tokio::spawn(async move { + let mut read = read; + while let Some(msg) = read.next().await { + let Some(this) = weak_self.upgrade() else { + break; + }; + match msg { + Ok(Message::Text(text)) => { + if !this.handle_text_message(&text) { + break; + } + } + Ok(Message::Close(_)) => { + this.finish_with_partial_or_error(Qwen3ASRError::NoFinalResult); + break; + } + Ok(_) => {} + Err(e) => { + log::error!("[qwen3-asr] receive loop error: {e}"); + this.finish_with_partial_or_error(Qwen3ASRError::ConnectionFailed( + e.to_string(), + )); + break; + } + } + } + }); + + Ok(()) + } + + pub async fn send_last_frame(&self) -> Result<(), Qwen3ASRError> { + let started = self.session_started.notified(); + tokio::pin!(started); + started.as_mut().enable(); + let ready = { + let st = self.state.lock(); + st.session_started || st.session_finished + }; + if !ready { + tokio::time::timeout(Duration::from_secs(5), started) + .await + .map_err(|_| Qwen3ASRError::FinalResultTimeout)?; + } + let (send_tx, tail_chunks) = { + let mut st = self.state.lock(); + let send_tx = st.send_tx.clone(); + if !st.pending_audio.is_empty() { + let pending = std::mem::take(&mut st.pending_audio); + st.audio_scratch.extend_from_slice(&pending); + } + let tail = if st.audio_scratch.is_empty() { + Vec::new() + } else { + vec![std::mem::take(&mut st.audio_scratch)] + }; + (send_tx, tail) + }; + let Some(send_tx) = send_tx else { + return Ok(()); + }; + for chunk in tail_chunks { + let _ = send_tx.send(SendItem::Audio(chunk)); + } + let (done_tx, done_rx) = oneshot::channel(); + send_tx + .send(SendItem::Finish(done_tx)) + .map_err(|_| Qwen3ASRError::SendFailed("send worker closed".to_string()))?; + done_rx + .await + .map_err(|_| Qwen3ASRError::SendFailed("finish ack dropped".to_string()))? + } + + pub async fn await_final_result(&self) -> Result { + let rx = self.final_rx.lock().take(); + let Some(rx) = rx else { + return Err(Qwen3ASRError::NoFinalResult); + }; + tokio::time::timeout(FINAL_RESULT_TIMEOUT, rx) + .await + .map_err(|_| Qwen3ASRError::FinalResultTimeout)? + .map_err(|_| Qwen3ASRError::NoFinalResult)? + } + + pub fn cancel(&self) { + let mut st = self.state.lock(); + st.pending_audio.clear(); + st.audio_scratch.clear(); + st.send_tx.take(); + st.final_tx.take(); + st.session_finished = true; + drop(st); + let writer = Arc::clone(&self.writer); + if let Ok(handle) = Handle::try_current() { + handle.spawn(async move { + let _ = close_writer(&writer).await; + }); + } else { + std::thread::spawn(move || { + if let Ok(rt) = tokio::runtime::Runtime::new() { + rt.block_on(async move { + let _ = close_writer(&writer).await; + }); + } + }); + } + } + + fn handle_text_message(&self, text: &str) -> bool { + let value: Value = match serde_json::from_str(text) { + Ok(v) => v, + Err(e) => { + log::warn!("[qwen3-asr] invalid json event: {e}"); + return true; + } + }; + let event = value.get("type").and_then(Value::as_str).unwrap_or_default(); + match event { + "session.created" => { + self.mark_session_started(); + true + } + "conversation.item.input_audio_transcription.text" => { + self.record_partial(&value); + true + } + "conversation.item.input_audio_transcription.completed" => { + self.record_completed(&value); + true + } + "session.finished" => { + self.finish_success(); + false + } + "error" => { + let message = value + .get("error") + .and_then(|e| e.get("message")) + .and_then(Value::as_str) + .unwrap_or("realtime session error") + .to_string(); + self.finish_with_partial_or_error(Qwen3ASRError::TaskFailed(message)); + false + } + _ => true, + } + } + + fn mark_session_started(&self) { + let (send_tx, chunks) = { + let mut st = self.state.lock(); + st.session_started = true; + if !st.pending_audio.is_empty() { + let pending = std::mem::take(&mut st.pending_audio); + st.audio_scratch.extend_from_slice(&pending); + } + let send_tx = st.send_tx.clone(); + let chunks = drain_audio_chunks(&mut st.audio_scratch); + (send_tx, chunks) + }; + if let Some(tx) = send_tx { + for chunk in chunks { + let _ = tx.send(SendItem::Audio(chunk)); + } + } + self.session_started.notify_waiters(); + } + + fn record_partial(&self, value: &Value) { + // 句段开放期服务端把累积文本放 `stash`，随后的精修 pass 放 `text`； + // 两者互斥出现，取非空者作为当前句段的 interim 文本。 + let text = value + .get("text") + .and_then(Value::as_str) + .filter(|s| !s.trim().is_empty()) + .or_else(|| { + value + .get("stash") + .and_then(Value::as_str) + .filter(|s| !s.trim().is_empty()) + }); + if let Some(text) = text { + self.state.lock().partial_text = text.trim().to_string(); + } + } + + fn record_completed(&self, value: &Value) { + let Some(transcript) = value.get("transcript").and_then(Value::as_str) else { + return; + }; + let trimmed = transcript.trim(); + let mut st = self.state.lock(); + if !trimmed.is_empty() { + st.completed_segments.push(trimmed.to_string()); + } + st.partial_text.clear(); + } + + fn finish_success(&self) { + let (tx, text, duration_ms) = { + let mut st = self.state.lock(); + if st.session_finished { + return; + } + st.session_finished = true; + st.send_tx.take(); + let mut segments = std::mem::take(&mut st.completed_segments); + // session.finished 前若还有未 completed 的 interim 尾巴（理论上 + // finish 会冲刷出 completed，防御性兜底），拼在最后。 + if !st.partial_text.is_empty() { + segments.push(std::mem::take(&mut st.partial_text)); + } + let text = join_segments(&segments); + let duration_ms = if st.bytes_received > 0 { + st.bytes_received / BYTES_PER_MS + } else { + st.start + .map(|start| start.elapsed().as_millis() as u64) + .unwrap_or_default() + }; + (st.final_tx.take(), text, duration_ms) + }; + if let Some(tx) = tx { + let _ = tx.send(Ok(RawTranscript { text, duration_ms })); + } + self.close_on_runtime(); + } + + fn finish_with_partial_or_error(&self, error: Qwen3ASRError) { + let has_partial = { + let st = self.state.lock(); + !st.completed_segments.is_empty() || !st.partial_text.trim().is_empty() + }; + if has_partial { + // 与 Bailian / Volcengine 保持一致：连接异常但已有结果时兜底返回。 + self.finish_success(); + } else { + self.finish_error(error); + } + } + + fn finish_error(&self, error: Qwen3ASRError) { + let tx = { + let mut st = self.state.lock(); + if st.session_finished { + return; + } + st.session_finished = true; + st.send_tx.take(); + st.final_tx.take() + }; + if let Some(tx) = tx { + let _ = tx.send(Err(error)); + } + self.close_on_runtime(); + } + + fn close_on_runtime(&self) { + let writer = Arc::clone(&self.writer); + if let Some(handle) = self.state.lock().runtime.clone() { + handle.spawn(async move { + let _ = close_writer(&writer).await; + }); + } + } +} + +impl AudioConsumer for Qwen3RealtimeASR { + fn consume_pcm_chunk(&self, pcm: &[u8]) { + if pcm.is_empty() { + return; + } + let (send_tx, chunks) = { + let mut st = self.state.lock(); + st.bytes_received = st.bytes_received.saturating_add(pcm.len() as u64); + if !st.session_started { + st.pending_audio.extend_from_slice(pcm); + return; + } + st.audio_scratch.extend_from_slice(pcm); + let chunks = drain_audio_chunks(&mut st.audio_scratch); + (st.send_tx.clone(), chunks) + }; + if let Some(tx) = send_tx { + for chunk in chunks { + let _ = tx.send(SendItem::Audio(chunk)); + } + } + } +} + +fn drain_audio_chunks(buffer: &mut Vec) -> Vec> { + let mut chunks = Vec::new(); + while buffer.len() >= TARGET_AUDIO_CHUNK_BYTES { + chunks.push(buffer.drain(..TARGET_AUDIO_CHUNK_BYTES).collect()); + } + chunks +} + +/// VAD 句段拼接：CJK 之间直接相连；拉丁词之间补空格，避免英文句段黏连。 +fn join_segments(segments: &[String]) -> String { + let mut joined = String::new(); + for seg in segments.iter().map(|s| s.trim()) { + if seg.is_empty() { + continue; + } + if let (Some(prev), Some(next)) = (joined.chars().last(), seg.chars().next()) { + if prev.is_ascii_alphanumeric() && next.is_ascii_alphanumeric() { + joined.push(' '); + } + } + joined.push_str(seg); + } + joined +} + +fn session_update_message() -> String { + // language 省略 => 服务端自动检测语种（2026-07 实测可用）。 + json!({ + "type": "session.update", + "session": { + "modalities": ["text"], + "input_audio_format": "pcm", + "sample_rate": 16000, + "turn_detection": { + "type": "server_vad", + "silence_duration_ms": VAD_SILENCE_DURATION_MS, + }, + }, + }) + .to_string() +} + +fn append_audio_message(pcm: &[u8]) -> String { + json!({ + "type": "input_audio_buffer.append", + "audio": base64::engine::general_purpose::STANDARD.encode(pcm), + }) + .to_string() +} + +fn finish_session_message() -> String { + json!({ "type": "session.finish" }).to_string() +} + +async fn send_text(writer: &SharedWriter, text: String) -> Result<(), Qwen3ASRError> { + let mut guard = writer.lock().await; + let Some(ws) = guard.as_mut() else { + return Err(Qwen3ASRError::ConnectionFailed( + "websocket writer not available".to_string(), + )); + }; + ws.send(Message::Text(text)) + .await + .map_err(|e| Qwen3ASRError::SendFailed(e.to_string())) +} + +async fn close_writer(writer: &SharedWriter) -> Result<(), Qwen3ASRError> { + let mut guard = writer.lock().await; + if let Some(mut ws) = guard.take() { + let _ = ws.close().await; + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn create_test_asr() -> Qwen3RealtimeASR { + Qwen3RealtimeASR::new(Qwen3RealtimeCredentials { + api_key: "sk-test".to_string(), + endpoint: String::new(), + model: String::new(), + }) + } + + // ---- credentials / URL ---- + + #[test] + fn credentials_apply_default_endpoint_and_model() { + let creds = Qwen3RealtimeCredentials { + api_key: "sk-test".to_string(), + endpoint: String::new(), + model: String::new(), + }; + assert_eq!(creds.normalized_endpoint(), DEFAULT_ENDPOINT); + assert_eq!(creds.normalized_model(), DEFAULT_MODEL); + assert_eq!( + creds.connect_url(), + "wss://dashscope.aliyuncs.com/api-ws/v1/realtime?model=qwen3-asr-flash-realtime" + ); + } + + #[test] + fn connect_url_keeps_existing_model_query() { + let creds = Qwen3RealtimeCredentials { + api_key: "sk-test".to_string(), + endpoint: "wss://dashscope.aliyuncs.com/api-ws/v1/realtime?model=custom".to_string(), + model: "ignored".to_string(), + }; + assert_eq!( + creds.connect_url(), + "wss://dashscope.aliyuncs.com/api-ws/v1/realtime?model=custom" + ); + } + + #[test] + fn connect_url_supports_dedicated_workspace_domain() { + let creds = Qwen3RealtimeCredentials { + api_key: "sk-test".to_string(), + endpoint: "wss://llm-xxx.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime/".to_string(), + model: String::new(), + }; + assert_eq!( + creds.connect_url(), + "wss://llm-xxx.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime?model=qwen3-asr-flash-realtime" + ); + } + + // ---- message builders ---- + + #[test] + fn session_update_uses_pcm_16k_and_server_vad() { + let value: Value = serde_json::from_str(&session_update_message()).unwrap(); + assert_eq!(value["type"], "session.update"); + assert_eq!(value["session"]["input_audio_format"], "pcm"); + assert_eq!(value["session"]["sample_rate"], 16000); + assert_eq!(value["session"]["turn_detection"]["type"], "server_vad"); + // language 省略走服务端自动检测 + assert!(value["session"]["input_audio_transcription"].is_null()); + } + + #[test] + fn append_message_base64_encodes_audio() { + let value: Value = serde_json::from_str(&append_audio_message(b"\x01\x02\x03")).unwrap(); + assert_eq!(value["type"], "input_audio_buffer.append"); + assert_eq!(value["audio"], "AQID"); + } + + #[test] + fn finish_message_shape() { + let value: Value = serde_json::from_str(&finish_session_message()).unwrap(); + assert_eq!(value["type"], "session.finish"); + } + + // ---- event handling ---- + + fn text_event(text: &str, stash: &str) -> Value { + json!({ + "type": "conversation.item.input_audio_transcription.text", + "text": text, + "stash": stash, + }) + } + + fn completed_event(transcript: &str) -> Value { + json!({ + "type": "conversation.item.input_audio_transcription.completed", + "transcript": transcript, + }) + } + + #[test] + fn partial_prefers_text_falls_back_to_stash() { + let asr = create_test_asr(); + // 句段开放期：text 为空、stash 有累积文本 + asr.record_partial(&text_event("", "今天")); + assert_eq!(asr.state.lock().partial_text, "今天"); + // 精修期：text 有值优先 + asr.record_partial(&text_event("今天天气", "旧stash")); + assert_eq!(asr.state.lock().partial_text, "今天天气"); + // 两者皆空不覆盖已有 partial + asr.record_partial(&text_event("", "")); + assert_eq!(asr.state.lock().partial_text, "今天天气"); + } + + #[test] + fn completed_appends_segment_and_clears_partial() { + let asr = create_test_asr(); + asr.record_partial(&text_event("", "第一句")); + asr.handle_text_message(&completed_event("第一句话说完了。").to_string()); + asr.handle_text_message(&completed_event("第二句也说完了。").to_string()); + let st = asr.state.lock(); + assert_eq!( + st.completed_segments, + vec!["第一句话说完了。", "第二句也说完了。"] + ); + assert!(st.partial_text.is_empty()); + } + + #[test] + fn finish_success_joins_segments_with_trailing_partial() { + let asr = create_test_asr(); + asr.record_completed(&completed_event("第一句。")); + asr.record_partial(&text_event("", "没说完的尾巴")); + + let (tx, mut rx) = oneshot::channel(); + { + let mut st = asr.state.lock(); + st.final_tx = Some(tx); + st.bytes_received = 32_000; + } + asr.finish_success(); + let result = rx.try_recv().unwrap().unwrap(); + assert_eq!(result.text, "第一句。没说完的尾巴"); + assert_eq!(result.duration_ms, 1_000); + } + + #[test] + fn session_finished_event_stops_loop_with_result() { + let asr = create_test_asr(); + asr.record_completed(&completed_event("你好。")); + let (tx, mut rx) = oneshot::channel(); + asr.state.lock().final_tx = Some(tx); + let keep_going = + asr.handle_text_message(&json!({"type": "session.finished"}).to_string()); + assert!(!keep_going); + assert_eq!(rx.try_recv().unwrap().unwrap().text, "你好。"); + } + + #[test] + fn error_event_with_partial_returns_partial() { + let asr = create_test_asr(); + asr.record_completed(&completed_event("已识别内容。")); + let (tx, mut rx) = oneshot::channel(); + asr.state.lock().final_tx = Some(tx); + let keep_going = asr.handle_text_message( + &json!({"type": "error", "error": {"message": "boom"}}).to_string(), + ); + assert!(!keep_going); + assert_eq!(rx.try_recv().unwrap().unwrap().text, "已识别内容。"); + } + + #[test] + fn error_event_without_partial_returns_error() { + let asr = create_test_asr(); + let (tx, mut rx) = oneshot::channel(); + asr.state.lock().final_tx = Some(tx); + asr.handle_text_message(&json!({"type": "error", "error": {"message": "boom"}}).to_string()); + let err = rx.try_recv().unwrap().unwrap_err(); + assert!(matches!(err, Qwen3ASRError::TaskFailed(m) if m == "boom")); + } + + #[test] + fn empty_session_finishes_with_empty_text() { + // 连接检查场景：纯静音无任何 completed，finish 后应返回空文本成功。 + let asr = create_test_asr(); + let (tx, mut rx) = oneshot::channel(); + asr.state.lock().final_tx = Some(tx); + asr.handle_text_message(&json!({"type": "session.finished"}).to_string()); + assert_eq!(rx.try_recv().unwrap().unwrap().text, ""); + } + + // ---- join_segments ---- + + #[test] + fn join_segments_cjk_concatenates_directly() { + let segments = vec!["第一句。".to_string(), "第二句。".to_string()]; + assert_eq!(join_segments(&segments), "第一句。第二句。"); + } + + #[test] + fn join_segments_latin_inserts_space() { + let segments = vec!["hello".to_string(), "world".to_string()]; + assert_eq!(join_segments(&segments), "hello world"); + } + + #[test] + fn join_segments_skips_empty() { + let segments = vec!["".to_string(), "内容".to_string(), " ".to_string()]; + assert_eq!(join_segments(&segments), "内容"); + } + + // ---- audio buffering ---- + + #[test] + fn audio_buffered_before_session_created() { + let asr = create_test_asr(); + asr.consume_pcm_chunk(&[0u8; 100]); + let st = asr.state.lock(); + assert_eq!(st.pending_audio.len(), 100); + assert_eq!(st.bytes_received, 100); + } + + #[test] + fn drain_audio_chunks_keeps_tail_buffered() { + let mut buffer = vec![1u8; TARGET_AUDIO_CHUNK_BYTES * 2 + 17]; + let chunks = drain_audio_chunks(&mut buffer); + assert_eq!(chunks.len(), 2); + assert_eq!(buffer.len(), 17); + } +} diff --git a/openless-all/app/src-tauri/src/commands/credentials.rs b/openless-all/app/src-tauri/src/commands/credentials.rs index d5b2980b..9c72a5de 100644 --- a/openless-all/app/src-tauri/src/commands/credentials.rs +++ b/openless-all/app/src-tauri/src/commands/credentials.rs @@ -46,7 +46,9 @@ pub(crate) fn asr_configured_for_provider(provider: &str, snap: &CredentialsSnap // 本地 ASR 不依赖云端凭据。 return true; } - if provider == crate::asr::bailian::PROVIDER_ID { + if provider == crate::asr::bailian::PROVIDER_ID + || provider == crate::asr::qwen_realtime::PROVIDER_ID + { return configured(&snap.asr_api_key); } if provider == crate::asr::mimo::PROVIDER_ID { diff --git a/openless-all/app/src-tauri/src/commands/providers.rs b/openless-all/app/src-tauri/src/commands/providers.rs index 9a376ffd..31cd7a87 100644 --- a/openless-all/app/src-tauri/src/commands/providers.rs +++ b/openless-all/app/src-tauri/src/commands/providers.rs @@ -37,6 +37,20 @@ pub async fn list_provider_models(kind: String) -> Result Result<(), String> { if active_asr == crate::asr::bailian::PROVIDER_ID { return validate_bailian_asr_provider().await; } + if active_asr == crate::asr::qwen_realtime::PROVIDER_ID { + return validate_qwen3_realtime_asr_provider().await; + } if active_asr == crate::asr::mimo::PROVIDER_ID { return validate_mimo_asr_provider().await; } @@ -270,6 +287,45 @@ async fn validate_bailian_asr_provider() -> Result<(), String> { .map_err(|e| e.to_string()) } +async fn validate_qwen3_realtime_asr_provider() -> Result<(), String> { + let api_key = CredentialsVault::get(CredentialAccount::AsrApiKey) + .map_err(|e| e.to_string())? + .unwrap_or_default(); + if api_key.trim().is_empty() { + return Err("API Key 为空".to_string()); + } + let endpoint = CredentialsVault::get(CredentialAccount::AsrEndpoint) + .map_err(|e| e.to_string())? + .filter(|s| !s.trim().is_empty()) + .unwrap_or_else(|| crate::asr::qwen_realtime::DEFAULT_ENDPOINT.to_string()); + if !crate::asr::bailian::endpoint_scheme_is_websocket(&endpoint) { + return Err("qwen3EndpointSchemeInvalid".to_string()); + } + let model = CredentialsVault::get(CredentialAccount::AsrModel) + .map_err(|e| e.to_string())? + .filter(|s| !s.trim().is_empty()) + .unwrap_or_else(|| crate::asr::qwen_realtime::DEFAULT_MODEL.to_string()); + let asr = std::sync::Arc::new(crate::asr::Qwen3RealtimeASR::new( + crate::asr::Qwen3RealtimeCredentials { + api_key, + endpoint, + model, + }, + )); + asr.open_session().await.map_err(|e| e.to_string())?; + // Realtime 协议对纯静音 + finish 干净返回 session.finished（2026-07 实测）， + // 无经典协议 <200ms 必报 EmptyAudio 的问题；发 500ms 与 bailian 验证对齐。 + crate::asr::AudioConsumer::consume_pcm_chunk( + &*asr, + &vec![0u8; crate::asr::qwen_realtime::TARGET_AUDIO_CHUNK_BYTES * 5], + ); + asr.send_last_frame().await.map_err(|e| e.to_string())?; + asr.await_final_result() + .await + .map(|_| ()) + .map_err(|e| e.to_string()) +} + pub(crate) fn active_asr_is_keyless_for_validation(provider: &str) -> bool { if cfg!(mobile) { return false; diff --git a/openless-all/app/src-tauri/src/coordinator.rs b/openless-all/app/src-tauri/src/coordinator.rs index 99b07d86..4bc80f66 100644 --- a/openless-all/app/src-tauri/src/coordinator.rs +++ b/openless-all/app/src-tauri/src/coordinator.rs @@ -25,8 +25,9 @@ use crate::asr::local::{ foundry, sherpa, FoundryLocalRuntime, FoundryLocalWhisperAsr, SherpaOnnxAsr, SherpaOnnxRuntime, }; use crate::asr::{ - BailianCredentials, BailianRealtimeASR, DictionaryHotword, MimoBatchASR, RawTranscript, - VolcengineCredentials, VolcengineStreamingASR, WhisperBatchASR, + BailianCredentials, BailianRealtimeASR, DictionaryHotword, MimoBatchASR, Qwen3RealtimeASR, + Qwen3RealtimeCredentials, RawTranscript, VolcengineCredentials, VolcengineStreamingASR, + WhisperBatchASR, }; use crate::combo_hotkey::{ComboHotkeyError, ComboHotkeyEvent, ComboHotkeyMonitor}; use crate::coordinator_state::{ @@ -180,6 +181,8 @@ enum ActiveAsr { Whisper(Arc), Mimo(Arc), Bailian(Arc), + /// 百炼 Qwen3-ASR-Flash 实时（OpenAI Realtime 风格 WS 协议）。 + Qwen3Realtime(Arc), #[cfg(target_os = "windows")] FoundryLocalWhisper(Arc), /// Windows sherpa-onnx 本地 ASR（offline batch + 实验 online streaming）。 @@ -208,6 +211,7 @@ fn asr_transcribe_uses_global_timeout(asr: &ActiveAsr) -> bool { #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum ActiveAsrProviderKind { Bailian, + Qwen3Realtime, Mimo, WhisperCompatible, Volcengine, @@ -216,6 +220,8 @@ enum ActiveAsrProviderKind { fn active_asr_provider_kind(id: &str) -> ActiveAsrProviderKind { if is_bailian_provider(id) { ActiveAsrProviderKind::Bailian + } else if is_qwen3_realtime_provider(id) { + ActiveAsrProviderKind::Qwen3Realtime } else if is_mimo_provider(id) { ActiveAsrProviderKind::Mimo } else if is_whisper_compatible_provider(id) { @@ -1566,6 +1572,13 @@ impl Coordinator { .map_err(|_| "重新转录超时".to_string())? .map_err(|e| e.to_string())? } + ActiveAsr::Qwen3Realtime(asr) => { + asr.send_last_frame().await.map_err(|e| e.to_string())?; + tokio::time::timeout(timeout, asr.await_final_result()) + .await + .map_err(|_| "重新转录超时".to_string())? + .map_err(|e| e.to_string())? + } ActiveAsr::Whisper(w) => tokio::time::timeout(timeout, w.transcribe()) .await .map_err(|_| "重新转录超时".to_string())? @@ -1989,6 +2002,28 @@ fn read_bailian_credentials() -> BailianCredentials { } } +fn read_qwen3_realtime_credentials() -> Qwen3RealtimeCredentials { + let api_key = CredentialsVault::get(CredentialAccount::AsrApiKey) + .ok() + .flatten() + .unwrap_or_default(); + let endpoint = CredentialsVault::get(CredentialAccount::AsrEndpoint) + .ok() + .flatten() + .filter(|s| !s.trim().is_empty()) + .unwrap_or_else(|| crate::asr::qwen_realtime::DEFAULT_ENDPOINT.to_string()); + let model = CredentialsVault::get(CredentialAccount::AsrModel) + .ok() + .flatten() + .filter(|s| !s.trim().is_empty()) + .unwrap_or_else(|| crate::asr::qwen_realtime::DEFAULT_MODEL.to_string()); + Qwen3RealtimeCredentials { + api_key, + endpoint, + model, + } +} + fn read_volc_credentials() -> VolcengineCredentials { let app_id = CredentialsVault::get(CredentialAccount::VolcengineAppKey) .ok() @@ -2339,6 +2374,10 @@ mod tests { active_asr_provider_kind(crate::asr::bailian::PROVIDER_ID), ActiveAsrProviderKind::Bailian ); + assert_eq!( + active_asr_provider_kind(crate::asr::qwen_realtime::PROVIDER_ID), + ActiveAsrProviderKind::Qwen3Realtime + ); assert_eq!( active_asr_provider_kind("whisper"), ActiveAsrProviderKind::WhisperCompatible diff --git a/openless-all/app/src-tauri/src/coordinator/asr_wiring.rs b/openless-all/app/src-tauri/src/coordinator/asr_wiring.rs index 8f949f40..fcbc7c5e 100644 --- a/openless-all/app/src-tauri/src/coordinator/asr_wiring.rs +++ b/openless-all/app/src-tauri/src/coordinator/asr_wiring.rs @@ -106,6 +106,7 @@ pub(super) fn ensure_asr_credentials() -> Result<(), String> { if is_whisper_compatible_provider(&active_asr) || is_bailian_provider(&active_asr) + || is_qwen3_realtime_provider(&active_asr) || is_mimo_provider(&active_asr) { let api_key = CredentialsVault::get(CredentialAccount::AsrApiKey) @@ -351,6 +352,10 @@ pub(super) fn is_bailian_provider(id: &str) -> bool { id == crate::asr::bailian::PROVIDER_ID } +pub(super) fn is_qwen3_realtime_provider(id: &str) -> bool { + id == crate::asr::qwen_realtime::PROVIDER_ID +} + pub(super) fn is_mimo_provider(id: &str) -> bool { id == crate::asr::mimo::PROVIDER_ID } @@ -385,6 +390,10 @@ pub(super) enum QaAsrStart { asr: Arc, bridge: Arc, }, + Qwen3Realtime { + asr: Arc, + bridge: Arc, + }, Ready { active: ActiveAsr, consumer: Arc, @@ -396,6 +405,7 @@ impl QaAsrStart { match self { QaAsrStart::Volcengine { asr, .. } => ActiveAsr::Volcengine(Arc::clone(asr)), QaAsrStart::Bailian { asr, .. } => ActiveAsr::Bailian(Arc::clone(asr)), + QaAsrStart::Qwen3Realtime { asr, .. } => ActiveAsr::Qwen3Realtime(Arc::clone(asr)), QaAsrStart::Ready { active, .. } => active.clone(), } } @@ -404,6 +414,7 @@ impl QaAsrStart { match self { QaAsrStart::Volcengine { bridge, .. } => Arc::clone(bridge) as _, QaAsrStart::Bailian { bridge, .. } => Arc::clone(bridge) as _, + QaAsrStart::Qwen3Realtime { bridge, .. } => Arc::clone(bridge) as _, QaAsrStart::Ready { consumer, .. } => Arc::clone(consumer), } } @@ -426,6 +437,15 @@ impl QaAsrStart { ); Ok(()) } + QaAsrStart::Qwen3Realtime { asr, bridge } => { + asr.open_session().await.map_err(|e| e.to_string())?; + let target: Arc = Arc::clone(asr) as _; + let flushed = bridge.attach(target); + log::info!( + "[coord] QA Qwen3 realtime ASR connected; flushed {flushed} deferred audio bytes" + ); + Ok(()) + } QaAsrStart::Ready { .. } => Ok(()), } } @@ -515,6 +535,10 @@ pub(super) async fn build_qa_asr_start(inner: &Arc, active_asr: &str) -> asr: Arc::new(BailianRealtimeASR::new(read_bailian_credentials())), bridge: Arc::new(DeferredAsrBridge::new()), }), + ActiveAsrProviderKind::Qwen3Realtime => Ok(QaAsrStart::Qwen3Realtime { + asr: Arc::new(Qwen3RealtimeASR::new(read_qwen3_realtime_credentials())), + bridge: Arc::new(DeferredAsrBridge::new()), + }), ActiveAsrProviderKind::Mimo => { let (api_key, base_url, model) = read_mimo_credentials(); let mimo = Arc::new(MimoBatchASR::new(api_key, base_url, model)); diff --git a/openless-all/app/src-tauri/src/coordinator/dictation.rs b/openless-all/app/src-tauri/src/coordinator/dictation.rs index 01384017..c6c7caf3 100644 --- a/openless-all/app/src-tauri/src/coordinator/dictation.rs +++ b/openless-all/app/src-tauri/src/coordinator/dictation.rs @@ -1609,6 +1609,83 @@ pub(super) async fn begin_session_as( let flushed_bytes = bridge.attach(target); log::info!("[coord] Bailian ASR connected; flushed {flushed_bytes} deferred audio bytes"); finish_starting_session(inner, current_session_id).await; + } else if is_qwen3_realtime_provider(&active_asr) { + // 与 Bailian 分支同构：流式 WS 会话 + DeferredAsrBridge 缓冲开链前音频。 + let asr = Arc::new(Qwen3RealtimeASR::new(read_qwen3_realtime_credentials())); + let bridge = Arc::new(DeferredAsrBridge::new()); + let consumer: Arc = bridge.clone(); + store_asr_for_session( + inner, + current_session_id, + ActiveAsr::Qwen3Realtime(Arc::clone(&asr)), + ); + start_recorder_for_starting(inner, current_session_id, &active_asr, consumer).await?; + + if let Err(e) = asr.open_session().await { + log::error!("[coord] open Qwen3 realtime ASR session failed: {e}"); + match startup_race_status_for_starting(inner, current_session_id) { + StartupRaceStatus::StaleContinuation => { + log::info!( + "[coord] stale Qwen3 realtime ASR open_session error from session {current_session_id} — ignoring" + ); + asr.cancel(); + discard_startup_resources_for_session(inner, current_session_id); + restore_prepared_windows_ime_session(inner, current_session_id); + return Ok(()); + } + StartupRaceStatus::CancelRaced => { + asr.cancel(); + discard_startup_resources_for_session(inner, current_session_id); + restore_prepared_windows_ime_session(inner, current_session_id); + set_phase_idle_if_session_matches(inner, current_session_id); + return Ok(()); + } + StartupRaceStatus::ActiveStarting => { + asr.cancel(); + } + } + discard_startup_resources_for_session(inner, current_session_id); + emit_capsule( + inner, + CapsuleState::Error, + 0.0, + 0, + Some(format!("ASR 连接失败: {e}")), + None, + ); + restore_prepared_windows_ime_session(inner, current_session_id); + set_phase_idle_if_session_matches(inner, current_session_id); + schedule_capsule_idle(inner, CAPSULE_AUTO_HIDE_DELAY_MS); + return Err(e.to_string()); + } + match startup_race_status_for_starting(inner, current_session_id) { + StartupRaceStatus::ActiveStarting => {} + StartupRaceStatus::CancelRaced => { + log::info!( + "[coord] cancel raced during Qwen3 realtime ASR open_session — aborting begin" + ); + asr.cancel(); + discard_startup_resources_for_session(inner, current_session_id); + restore_prepared_windows_ime_session(inner, current_session_id); + set_phase_idle_if_session_matches(inner, current_session_id); + return Ok(()); + } + StartupRaceStatus::StaleContinuation => { + log::info!( + "[coord] stale Qwen3 realtime ASR open_session continuation from session {current_session_id} — ignoring" + ); + asr.cancel(); + discard_startup_resources_for_session(inner, current_session_id); + restore_prepared_windows_ime_session(inner, current_session_id); + return Ok(()); + } + } + let target: Arc = asr; + let flushed_bytes = bridge.attach(target); + log::info!( + "[coord] Qwen3 realtime ASR connected; flushed {flushed_bytes} deferred audio bytes" + ); + finish_starting_session(inner, current_session_id).await; } else if is_mimo_provider(&active_asr) { let (api_key, base_url, model) = read_mimo_credentials(); let mimo = Arc::new(MimoBatchASR::new(api_key, base_url, model)); @@ -2244,6 +2321,33 @@ pub(super) async fn end_session(inner: &Arc) -> Result<(), 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); diff --git a/openless-all/app/src-tauri/src/coordinator/qa_session.rs b/openless-all/app/src-tauri/src/coordinator/qa_session.rs index 094a1c6c..a9746e07 100644 --- a/openless-all/app/src-tauri/src/coordinator/qa_session.rs +++ b/openless-all/app/src-tauri/src/coordinator/qa_session.rs @@ -242,6 +242,21 @@ pub(super) async fn transcribe_overlay_dictation_asr( } } } + ActiveAsr::Qwen3Realtime(asr) => { + debug_assert!(uses_global_timeout); + if let Err(error) = asr.send_last_frame().await { + log::error!("[coord] overlay QA: Qwen3 realtime send last frame failed: {error}"); + } + 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(raw)) => Ok(raw), + Ok(Err(error)) => Err(error.to_string()), + Err(_) => { + asr.cancel(); + Err("qwen3 realtime global timeout".to_string()) + } + } + } ActiveAsr::Whisper(whisper) => { debug_assert!(uses_global_timeout); let timeout_duration = std::time::Duration::from_secs(COORDINATOR_GLOBAL_TIMEOUT_SECS); @@ -783,6 +798,30 @@ pub(super) async fn end_qa_session(inner: &Arc) -> Result<(), String> { } } } + ActiveAsr::Qwen3Realtime(asr) => { + debug_assert!(uses_global_timeout); + if let Err(e) = asr.send_last_frame().await { + log::error!("[coord] QA: 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)) => r, + Ok(Err(e)) => { + log::error!("[coord] QA: Qwen3 realtime await final failed: {e}"); + finish_qa_with_error(inner, format!("识别失败: {e}")); + return Err(e.to_string()); + } + Err(_) => { + log::error!( + "[coord] QA: Qwen3 realtime 全局超时 {} 秒", + COORDINATOR_GLOBAL_TIMEOUT_SECS + ); + asr.cancel(); + finish_qa_with_error(inner, "识别超时".to_string()); + return Err("qwen3 realtime global timeout".to_string()); + } + } + } ActiveAsr::Whisper(w) => { debug_assert!(uses_global_timeout); let timeout_duration = std::time::Duration::from_secs(COORDINATOR_GLOBAL_TIMEOUT_SECS); diff --git a/openless-all/app/src-tauri/src/coordinator/resources.rs b/openless-all/app/src-tauri/src/coordinator/resources.rs index 03184636..30fd3bdc 100644 --- a/openless-all/app/src-tauri/src/coordinator/resources.rs +++ b/openless-all/app/src-tauri/src/coordinator/resources.rs @@ -69,6 +69,7 @@ pub(super) fn cancel_active_asr(asr: ActiveAsr) { ActiveAsr::Whisper(w) => w.cancel(), ActiveAsr::Mimo(m) => m.cancel(), ActiveAsr::Bailian(b) => b.cancel(), + ActiveAsr::Qwen3Realtime(q) => q.cancel(), #[cfg(target_os = "windows")] ActiveAsr::FoundryLocalWhisper(local) => local.cancel(), #[cfg(target_os = "windows")] diff --git a/openless-all/app/src/i18n/en.ts b/openless-all/app/src/i18n/en.ts index 200bd48f..f270b2a0 100644 --- a/openless-all/app/src/i18n/en.ts +++ b/openless-all/app/src/i18n/en.ts @@ -778,6 +778,7 @@ export const en: typeof zhCN = { custom: 'Custom', asrVolcengine: 'Volcengine bigasr', asrBailian: 'Alibaba Bailian realtime ASR', + asrBailianQwen3: 'Bailian Qwen3 Realtime ASR', asrSiliconflow: 'SiliconFlow SenseVoice', asrZhipu: 'Zhipu GLM-ASR', asrGroq: 'Groq Whisper-large-v3', @@ -838,6 +839,7 @@ export const en: typeof zhCN = { endpointMustUseHttps: 'HTTP endpoints are allowed, but API keys and audio content may leak in transit.', endpointInvalid: 'Endpoint format is invalid.', bailianEndpointSchemeInvalid: 'Bailian realtime ASR uses the DashScope WebSocket gateway: the endpoint must start with wss:// (default: wss://dashscope.aliyuncs.com/api-ws/v1/inference/). An https:// compatible-mode URL will not work here.', + qwen3EndpointSchemeInvalid: 'Qwen3 realtime ASR uses the DashScope Realtime WebSocket gateway: the endpoint must start with wss:// (default: wss://dashscope.aliyuncs.com/api-ws/v1/realtime). An https:// URL will not work here.', responseTooLarge: 'Provider response is too large to validate safely.', asrInvalidJson: 'ASR response is not valid JSON.', asrMissingTextField: 'ASR response is missing the text field.', diff --git a/openless-all/app/src/i18n/ja.ts b/openless-all/app/src/i18n/ja.ts index 77142167..49ad2cd7 100644 --- a/openless-all/app/src/i18n/ja.ts +++ b/openless-all/app/src/i18n/ja.ts @@ -780,6 +780,7 @@ export const ja: typeof zhCN = { custom: 'カスタム', asrVolcengine: 'Volcengine bigasr', asrBailian: 'Alibaba Bailian リアルタイム ASR', + asrBailianQwen3: 'Bailian Qwen3 リアルタイム ASR', asrSiliconflow: 'SiliconFlow SenseVoice', asrZhipu: 'Zhipu GLM-ASR', asrGroq: 'Groq Whisper-large-v3', @@ -840,6 +841,7 @@ export const ja: typeof zhCN = { endpointMustUseHttps: 'HTTP Endpoint も使用できますが、API Key と音声内容が通信中に漏えいする可能性があります。', endpointInvalid: 'Endpoint の形式が無効です。', bailianEndpointSchemeInvalid: 'Bailian リアルタイム ASR は DashScope の WebSocket ゲートウェイを使用します。エンドポイントは wss:// で始まる必要があります（既定: wss://dashscope.aliyuncs.com/api-ws/v1/inference/）。https:// の互換モード URL はここでは使用できません。', + qwen3EndpointSchemeInvalid: 'Qwen3 リアルタイム ASR は DashScope Realtime WebSocket ゲートウェイを使用します。エンドポイントは wss:// で始まる必要があります（既定: wss://dashscope.aliyuncs.com/api-ws/v1/realtime）。https:// の URL はここでは使用できません。', responseTooLarge: 'サプライヤーの応答が大きすぎるため、安全のため検証を停止しました。', asrInvalidJson: 'ASR の応答が有効な JSON ではありません。', asrMissingTextField: 'ASR の応答に text フィールドがありません。', diff --git a/openless-all/app/src/i18n/ko.ts b/openless-all/app/src/i18n/ko.ts index 699966d4..ce86b5a4 100644 --- a/openless-all/app/src/i18n/ko.ts +++ b/openless-all/app/src/i18n/ko.ts @@ -780,6 +780,7 @@ export const ko: typeof zhCN = { custom: '사용자 정의', asrVolcengine: 'Volcengine bigasr', asrBailian: 'Alibaba Bailian 실시간 ASR', + asrBailianQwen3: 'Bailian Qwen3 실시간 ASR', asrSiliconflow: 'SiliconFlow SenseVoice', asrZhipu: 'Zhipu GLM-ASR', asrGroq: 'Groq Whisper-large-v3', @@ -840,6 +841,7 @@ export const ko: typeof zhCN = { endpointMustUseHttps: 'HTTP Endpoint 를 사용할 수 있지만, API Key 와 음성 내용이 전송 중 유출될 수 있습니다.', endpointInvalid: 'Endpoint 형식이 올바르지 않습니다.', bailianEndpointSchemeInvalid: 'Bailian 실시간 ASR은 DashScope WebSocket 게이트웨이를 사용합니다. 엔드포인트는 wss://로 시작해야 합니다(기본값: wss://dashscope.aliyuncs.com/api-ws/v1/inference/). https:// 호환 모드 주소는 여기서 사용할 수 없습니다.', + qwen3EndpointSchemeInvalid: 'Qwen3 실시간 ASR은 DashScope Realtime WebSocket 게이트웨이를 사용합니다. 엔드포인트는 wss://로 시작해야 합니다(기본값: wss://dashscope.aliyuncs.com/api-ws/v1/realtime). https:// 주소는 여기서 사용할 수 없습니다.', responseTooLarge: '공급자 응답이 너무 커서 안전을 위해 검증을 중단했습니다.', asrInvalidJson: 'ASR 응답이 유효한 JSON 이 아닙니다.', asrMissingTextField: 'ASR 응답에 text 필드가 없습니다.', diff --git a/openless-all/app/src/i18n/zh-CN.ts b/openless-all/app/src/i18n/zh-CN.ts index 04b5a8df..b0e9ac2f 100644 --- a/openless-all/app/src/i18n/zh-CN.ts +++ b/openless-all/app/src/i18n/zh-CN.ts @@ -776,6 +776,7 @@ export const zhCN = { custom: '自定义', asrVolcengine: '火山引擎 bigasr', asrBailian: '阿里云百炼实时 ASR', + asrBailianQwen3: '阿里云百炼 Qwen3 实时 ASR', asrSiliconflow: '硅基流动 SenseVoice', asrZhipu: '智谱 GLM-ASR', asrGroq: 'Groq Whisper-large-v3', @@ -836,6 +837,7 @@ export const zhCN = { endpointMustUseHttps: '允许使用 HTTP Endpoint，但请注意：API Key 和音频内容可能在传输中泄漏。', endpointInvalid: 'Endpoint 格式不合法。', bailianEndpointSchemeInvalid: '百炼实时 ASR 走 DashScope WebSocket 网关，接口地址必须以 wss:// 开头（默认 wss://dashscope.aliyuncs.com/api-ws/v1/inference/）；https:// 的兼容模式地址在此不可用。', + qwen3EndpointSchemeInvalid: 'Qwen3 实时 ASR 走 DashScope Realtime WebSocket 网关，接口地址必须以 wss:// 开头（默认 wss://dashscope.aliyuncs.com/api-ws/v1/realtime）；https:// 地址在此不可用。', responseTooLarge: '供应商响应过大，已停止验证以保证安全。', asrInvalidJson: 'ASR 响应不是有效 JSON。', asrMissingTextField: 'ASR 响应缺少 text 字段。', diff --git a/openless-all/app/src/i18n/zh-TW.ts b/openless-all/app/src/i18n/zh-TW.ts index 0e316d29..976d7100 100644 --- a/openless-all/app/src/i18n/zh-TW.ts +++ b/openless-all/app/src/i18n/zh-TW.ts @@ -778,6 +778,7 @@ export const zhTW: typeof zhCN = { custom: '自定義', asrVolcengine: '火山引擎 bigasr', asrBailian: '阿里雲百煉即時 ASR', + asrBailianQwen3: '阿里雲百煉 Qwen3 即時 ASR', asrSiliconflow: '硅基流動 SenseVoice', asrZhipu: '智譜 GLM-ASR', asrGroq: 'Groq Whisper-large-v3', @@ -838,6 +839,7 @@ export const zhTW: typeof zhCN = { endpointMustUseHttps: '允許使用 HTTP Endpoint，但請注意：API Key 和音訊內容可能在傳輸中外洩。', endpointInvalid: 'Endpoint 格式不合法。', bailianEndpointSchemeInvalid: '百煉即時 ASR 走 DashScope WebSocket 閘道，接口地址必須以 wss:// 開頭（預設 wss://dashscope.aliyuncs.com/api-ws/v1/inference/）；https:// 的相容模式地址在此不可用。', + qwen3EndpointSchemeInvalid: 'Qwen3 即時 ASR 走 DashScope Realtime WebSocket 閘道，接口地址必須以 wss:// 開頭（預設 wss://dashscope.aliyuncs.com/api-ws/v1/realtime）；https:// 地址在此不可用。', responseTooLarge: '供應商響應過大，已停止驗證以保證安全。', asrInvalidJson: 'ASR 響應不是有效 JSON。', asrMissingTextField: 'ASR 響應缺少 text 字段。', diff --git a/openless-all/app/src/pages/settings/ProvidersSection.tsx b/openless-all/app/src/pages/settings/ProvidersSection.tsx index 63012cce..bd6b5da0 100644 --- a/openless-all/app/src/pages/settings/ProvidersSection.tsx +++ b/openless-all/app/src/pages/settings/ProvidersSection.tsx @@ -155,6 +155,10 @@ const ASR_DEFAULT_RESOURCE_ID = 'volc.seedasr.sauc.duration'; const ASR_PRESETS: ReadonlyArray<{ id: AsrPresetId; nameKey: string; baseUrl: string; model: string }> = [ { id: 'volcengine', nameKey: 'asrVolcengine', baseUrl: '', model: '' }, { id: 'bailian', nameKey: 'asrBailian', baseUrl: 'wss://dashscope.aliyuncs.com/api-ws/v1/inference/', model: 'fun-asr-realtime' }, + // Qwen3-ASR-Flash 实时：OpenAI Realtime 风格 WS（/api-ws/v1/realtime）， + // 与上面经典 inference 协议不同，由 asr/qwen_realtime.rs 专用 client 处理。 + // 业务空间专属域名（wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime）同样可用。 + { id: 'bailian-qwen3-realtime', nameKey: 'asrBailianQwen3', baseUrl: 'wss://dashscope.aliyuncs.com/api-ws/v1/realtime', model: 'qwen3-asr-flash-realtime' }, { id: 'siliconflow', nameKey: 'asrSiliconflow', baseUrl: 'https://api.siliconflow.cn/v1', model: 'FunAudioLLM/SenseVoiceSmall' }, { id: 'zhipu', nameKey: 'asrZhipu', baseUrl: 'https://open.bigmodel.cn/api/paas/v4', model: 'glm-asr-2512' }, { id: 'groq', nameKey: 'asrGroq', baseUrl: 'https://api.groq.com/openai/v1', model: 'whisper-large-v3-turbo' }, @@ -622,6 +626,7 @@ function providerErrorMessage(error: unknown, t: ReturnType Date: Sun, 12 Jul 2026 18:37:13 +0800 Subject: [PATCH 2/2] fix(asr): harden qwen realtime protocol --- .../app/src-tauri/src/asr/qwen_realtime.rs | 284 ++++++++++++++---- .../app/src-tauri/src/commands/providers.rs | 5 +- 2 files changed, 228 insertions(+), 61 deletions(-) diff --git a/openless-all/app/src-tauri/src/asr/qwen_realtime.rs b/openless-all/app/src-tauri/src/asr/qwen_realtime.rs index f3d95e82..577f61cd 100644 --- a/openless-all/app/src-tauri/src/asr/qwen_realtime.rs +++ b/openless-all/app/src-tauri/src/asr/qwen_realtime.rs @@ -29,6 +29,7 @@ use tokio_tungstenite::tungstenite::client::IntoClientRequest; use tokio_tungstenite::tungstenite::http::header::HeaderValue; use tokio_tungstenite::tungstenite::Message; use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream}; +use uuid::Uuid; use super::{AudioConsumer, RawTranscript}; @@ -40,6 +41,8 @@ pub const DEFAULT_MODEL: &str = "qwen3-asr-flash-realtime"; pub const TARGET_AUDIO_CHUNK_BYTES: usize = 3_200; const BYTES_PER_MS: u64 = 32; const FINAL_RESULT_TIMEOUT: Duration = Duration::from_secs(12); +const SESSION_READY_TIMEOUT: Duration = Duration::from_secs(5); +const WRITE_TIMEOUT: Duration = Duration::from_secs(5); /// server_vad 断句静默阈值。官方默认 400ms；取 500ms 降低说话中途换气被切断的概率。 const VAD_SILENCE_DURATION_MS: u32 = 500; @@ -116,6 +119,7 @@ struct SyncState { bytes_received: u64, session_started: bool, session_finished: bool, + session_start_error: Option, runtime: Option, start: Option, final_tx: Option>>, @@ -133,6 +137,7 @@ pub struct Qwen3RealtimeASR { writer: SharedWriter, final_rx: ParkingMutex>>>, session_started: Arc, + session_finished: Arc, } impl Qwen3RealtimeASR { @@ -143,6 +148,7 @@ impl Qwen3RealtimeASR { writer: Arc::new(AsyncMutex::new(None)), final_rx: ParkingMutex::new(None), session_started: Arc::new(Notify::new()), + session_finished: Arc::new(Notify::new()), } } @@ -160,10 +166,9 @@ impl Qwen3RealtimeASR { HeaderValue::from_str(&format!("Bearer {}", self.credentials.api_key.trim())) .map_err(|e| Qwen3ASRError::ConnectionFailed(e.to_string()))?, ); - request.headers_mut().insert( - "OpenAI-Beta", - HeaderValue::from_static("realtime=v1"), - ); + request + .headers_mut() + .insert("OpenAI-Beta", HeaderValue::from_static("realtime=v1")); let (ws, _resp) = connect_async(request) .await @@ -184,6 +189,7 @@ impl Qwen3RealtimeASR { *self.final_rx.lock() = Some(final_rx); let writer_for_worker = Arc::clone(&self.writer); + let weak_self_for_worker = Arc::downgrade(self); tokio::spawn(async move { while let Some(item) = send_rx.recv().await { match item { @@ -192,6 +198,10 @@ impl Qwen3RealtimeASR { send_text(&writer_for_worker, append_audio_message(&chunk)).await { log::error!("[qwen3-asr] audio frame send failed: {e}"); + if let Some(this) = weak_self_for_worker.upgrade() { + this.finish_error(e); + } + break; } } SendItem::Finish(done) => { @@ -204,8 +214,6 @@ impl Qwen3RealtimeASR { } }); - send_text(&self.writer, session_update_message()).await?; - let weak_self = Arc::downgrade(self); tokio::spawn(async move { let mut read = read; @@ -220,12 +228,16 @@ impl Qwen3RealtimeASR { } } Ok(Message::Close(_)) => { + this.fail_session_start( + "websocket closed before session configuration completed", + ); this.finish_with_partial_or_error(Qwen3ASRError::NoFinalResult); break; } Ok(_) => {} Err(e) => { log::error!("[qwen3-asr] receive loop error: {e}"); + this.fail_session_start(&e.to_string()); this.finish_with_partial_or_error(Qwen3ASRError::ConnectionFailed( e.to_string(), )); @@ -235,49 +247,78 @@ impl Qwen3RealtimeASR { } }); - Ok(()) - } - - pub async fn send_last_frame(&self) -> Result<(), Qwen3ASRError> { let started = self.session_started.notified(); tokio::pin!(started); started.as_mut().enable(); - let ready = { - let st = self.state.lock(); - st.session_started || st.session_finished - }; - if !ready { - tokio::time::timeout(Duration::from_secs(5), started) + if let Err(error) = send_text(&self.writer, session_update_message()).await { + self.cancel(); + return Err(error); + } + let ready_result = if !self.state.lock().session_started { + tokio::time::timeout(SESSION_READY_TIMEOUT, started) .await - .map_err(|_| Qwen3ASRError::FinalResultTimeout)?; + .map_err(|_| Qwen3ASRError::FinalResultTimeout) + } else { + Ok(()) + }; + if let Err(error) = ready_result { + self.cancel(); + return Err(error); } - let (send_tx, tail_chunks) = { - let mut st = self.state.lock(); - let send_tx = st.send_tx.clone(); - if !st.pending_audio.is_empty() { - let pending = std::mem::take(&mut st.pending_audio); - st.audio_scratch.extend_from_slice(&pending); - } - let tail = if st.audio_scratch.is_empty() { - Vec::new() - } else { - vec![std::mem::take(&mut st.audio_scratch)] + if let Some(error) = self.state.lock().session_start_error.clone() { + self.cancel(); + return Err(Qwen3ASRError::TaskFailed(error)); + } + + Ok(()) + } + + pub async fn send_last_frame(&self) -> Result<(), Qwen3ASRError> { + let result = tokio::time::timeout(FINAL_RESULT_TIMEOUT, async { + let finished = self.session_finished.notified(); + tokio::pin!(finished); + finished.as_mut().enable(); + let (send_tx, tail_chunks) = { + let mut st = self.state.lock(); + let send_tx = st.send_tx.clone(); + if !st.pending_audio.is_empty() { + let pending = std::mem::take(&mut st.pending_audio); + st.audio_scratch.extend_from_slice(&pending); + } + let tail = if st.audio_scratch.is_empty() { + Vec::new() + } else { + vec![std::mem::take(&mut st.audio_scratch)] + }; + (send_tx, tail) }; - (send_tx, tail) - }; - let Some(send_tx) = send_tx else { - return Ok(()); - }; - for chunk in tail_chunks { - let _ = send_tx.send(SendItem::Audio(chunk)); + let Some(send_tx) = send_tx else { + return Ok(()); + }; + for chunk in tail_chunks { + send_tx + .send(SendItem::Audio(chunk)) + .map_err(|_| Qwen3ASRError::SendFailed("send worker closed".to_string()))?; + } + let (done_tx, done_rx) = oneshot::channel(); + send_tx + .send(SendItem::Finish(done_tx)) + .map_err(|_| Qwen3ASRError::SendFailed("send worker closed".to_string()))?; + done_rx + .await + .map_err(|_| Qwen3ASRError::SendFailed("finish ack dropped".to_string()))??; + if !self.state.lock().session_finished { + finished.await; + } + Ok(()) + }) + .await + .map_err(|_| Qwen3ASRError::FinalResultTimeout) + .and_then(|result| result); + if result.is_err() { + self.cancel(); } - let (done_tx, done_rx) = oneshot::channel(); - send_tx - .send(SendItem::Finish(done_tx)) - .map_err(|_| Qwen3ASRError::SendFailed("send worker closed".to_string()))?; - done_rx - .await - .map_err(|_| Qwen3ASRError::SendFailed("finish ack dropped".to_string()))? + result } pub async fn await_final_result(&self) -> Result { @@ -323,9 +364,12 @@ impl Qwen3RealtimeASR { return true; } }; - let event = value.get("type").and_then(Value::as_str).unwrap_or_default(); + let event = value + .get("type") + .and_then(Value::as_str) + .unwrap_or_default(); match event { - "session.created" => { + "session.updated" => { self.mark_session_started(); true } @@ -337,6 +381,20 @@ impl Qwen3RealtimeASR { self.record_completed(&value); true } + "conversation.item.input_audio_transcription.failed" => { + let item_id = value + .get("item_id") + .and_then(Value::as_str) + .unwrap_or("unknown item"); + let message = value + .get("error") + .and_then(|error| error.get("message")) + .and_then(Value::as_str) + .or_else(|| value.get("message").and_then(Value::as_str)) + .unwrap_or("audio transcription failed"); + self.finish_error(Qwen3ASRError::TaskFailed(format!("{item_id}: {message}"))); + false + } "session.finished" => { self.finish_success(); false @@ -375,6 +433,14 @@ impl Qwen3RealtimeASR { self.session_started.notify_waiters(); } + fn fail_session_start(&self, error: &str) { + let mut st = self.state.lock(); + if !st.session_started && st.session_start_error.is_none() { + st.session_start_error = Some(error.to_string()); + self.session_started.notify_waiters(); + } + } + fn record_partial(&self, value: &Value) { // 句段开放期服务端把累积文本放 `stash`，随后的精修 pass 放 `text`； // 两者互斥出现，取非空者作为当前句段的 interim 文本。 @@ -432,6 +498,7 @@ impl Qwen3RealtimeASR { if let Some(tx) = tx { let _ = tx.send(Ok(RawTranscript { text, duration_ms })); } + self.session_finished.notify_waiters(); self.close_on_runtime(); } @@ -449,6 +516,7 @@ impl Qwen3RealtimeASR { } fn finish_error(&self, error: Qwen3ASRError) { + self.fail_session_start(&error.to_string()); let tx = { let mut st = self.state.lock(); if st.session_finished { @@ -461,6 +529,7 @@ impl Qwen3RealtimeASR { if let Some(tx) = tx { let _ = tx.send(Err(error)); } + self.session_finished.notify_waiters(); self.close_on_runtime(); } @@ -514,7 +583,10 @@ fn join_segments(segments: &[String]) -> String { continue; } if let (Some(prev), Some(next)) = (joined.chars().last(), seg.chars().next()) { - if prev.is_ascii_alphanumeric() && next.is_ascii_alphanumeric() { + if next.is_ascii_alphanumeric() + && (prev.is_ascii_alphanumeric() + || matches!(prev, '.' | ',' | '?' | '!' | ':' | ';')) + { joined.push(' '); } } @@ -527,6 +599,7 @@ fn session_update_message() -> String { // language 省略 => 服务端自动检测语种（2026-07 实测可用）。 json!({ "type": "session.update", + "event_id": event_id(), "session": { "modalities": ["text"], "input_audio_format": "pcm", @@ -543,25 +616,40 @@ fn session_update_message() -> String { fn append_audio_message(pcm: &[u8]) -> String { json!({ "type": "input_audio_buffer.append", + "event_id": event_id(), "audio": base64::engine::general_purpose::STANDARD.encode(pcm), }) .to_string() } fn finish_session_message() -> String { - json!({ "type": "session.finish" }).to_string() + json!({ "type": "session.finish", "event_id": event_id() }).to_string() +} + +fn event_id() -> String { + format!("event_{}", Uuid::new_v4()) +} + +pub fn endpoint_scheme_is_secure_websocket(endpoint: &str) -> bool { + url::Url::parse(endpoint.trim()) + .map(|url| url.scheme() == "wss") + .unwrap_or(false) } async fn send_text(writer: &SharedWriter, text: String) -> Result<(), Qwen3ASRError> { - let mut guard = writer.lock().await; - let Some(ws) = guard.as_mut() else { - return Err(Qwen3ASRError::ConnectionFailed( - "websocket writer not available".to_string(), - )); - }; - ws.send(Message::Text(text)) - .await - .map_err(|e| Qwen3ASRError::SendFailed(e.to_string())) + tokio::time::timeout(WRITE_TIMEOUT, async { + let mut guard = writer.lock().await; + let Some(ws) = guard.as_mut() else { + return Err(Qwen3ASRError::ConnectionFailed( + "websocket writer not available".to_string(), + )); + }; + ws.send(Message::Text(text)) + .await + .map_err(|e| Qwen3ASRError::SendFailed(e.to_string())) + }) + .await + .map_err(|_| Qwen3ASRError::SendFailed("websocket write timed out".to_string()))? } async fn close_writer(writer: &SharedWriter) -> Result<(), Qwen3ASRError> { @@ -638,6 +726,9 @@ mod tests { assert_eq!(value["session"]["turn_detection"]["type"], "server_vad"); // language 省略走服务端自动检测 assert!(value["session"]["input_audio_transcription"].is_null()); + assert!(value["event_id"] + .as_str() + .is_some_and(|id| id.starts_with("event_"))); } #[test] @@ -645,12 +736,27 @@ mod tests { let value: Value = serde_json::from_str(&append_audio_message(b"\x01\x02\x03")).unwrap(); assert_eq!(value["type"], "input_audio_buffer.append"); assert_eq!(value["audio"], "AQID"); + assert!(value["event_id"] + .as_str() + .is_some_and(|id| id.starts_with("event_"))); } #[test] fn finish_message_shape() { let value: Value = serde_json::from_str(&finish_session_message()).unwrap(); assert_eq!(value["type"], "session.finish"); + assert!(value["event_id"] + .as_str() + .is_some_and(|id| id.starts_with("event_"))); + } + + #[test] + fn client_event_ids_are_unique() { + let update: Value = serde_json::from_str(&session_update_message()).unwrap(); + let append: Value = serde_json::from_str(&append_audio_message(b"audio")).unwrap(); + let finish: Value = serde_json::from_str(&finish_session_message()).unwrap(); + assert_ne!(update["event_id"], append["event_id"]); + assert_ne!(append["event_id"], finish["event_id"]); } // ---- event handling ---- @@ -722,8 +828,7 @@ mod tests { asr.record_completed(&completed_event("你好。")); let (tx, mut rx) = oneshot::channel(); asr.state.lock().final_tx = Some(tx); - let keep_going = - asr.handle_text_message(&json!({"type": "session.finished"}).to_string()); + let keep_going = asr.handle_text_message(&json!({"type": "session.finished"}).to_string()); assert!(!keep_going); assert_eq!(rx.try_recv().unwrap().unwrap().text, "你好。"); } @@ -746,11 +851,52 @@ mod tests { let asr = create_test_asr(); let (tx, mut rx) = oneshot::channel(); asr.state.lock().final_tx = Some(tx); - asr.handle_text_message(&json!({"type": "error", "error": {"message": "boom"}}).to_string()); + asr.handle_text_message( + &json!({"type": "error", "error": {"message": "boom"}}).to_string(), + ); let err = rx.try_recv().unwrap().unwrap_err(); assert!(matches!(err, Qwen3ASRError::TaskFailed(m) if m == "boom")); } + #[test] + fn transcription_failure_returns_error_instead_of_silent_success() { + let asr = create_test_asr(); + let (tx, mut rx) = oneshot::channel(); + asr.state.lock().final_tx = Some(tx); + let keep_going = asr.handle_text_message( + &json!({ + "type": "conversation.item.input_audio_transcription.failed", + "item_id": "item_123", + "error": { "message": "transcription rejected" }, + }) + .to_string(), + ); + assert!(!keep_going); + let err = rx.try_recv().unwrap().unwrap_err(); + assert!( + matches!(err, Qwen3ASRError::TaskFailed(message) if message == "item_123: transcription rejected") + ); + } + + #[test] + fn append_write_failure_returns_error_instead_of_partial_success() { + let asr = create_test_asr(); + let (tx, mut rx) = oneshot::channel(); + asr.state.lock().final_tx = Some(tx); + asr.finish_error(Qwen3ASRError::SendFailed("websocket write timed out".to_string())); + let err = rx.try_recv().unwrap().unwrap_err(); + assert!(matches!(err, Qwen3ASRError::SendFailed(message) if message == "websocket write timed out")); + } + + #[test] + fn only_session_updated_marks_session_ready() { + let asr = create_test_asr(); + asr.handle_text_message(&json!({"type": "session.created"}).to_string()); + assert!(!asr.state.lock().session_started); + asr.handle_text_message(&json!({"type": "session.updated"}).to_string()); + assert!(asr.state.lock().session_started); + } + #[test] fn empty_session_finishes_with_empty_text() { // 连接检查场景：纯静音无任何 completed，finish 后应返回空文本成功。 @@ -775,6 +921,28 @@ mod tests { assert_eq!(join_segments(&segments), "hello world"); } + #[test] + fn join_segments_latin_punctuation_inserts_space() { + for punctuation in ['.', ',', '?', '!'] { + let segments = vec![format!("Hello{punctuation}"), "World".to_string()]; + assert_eq!( + join_segments(&segments), + format!("Hello{punctuation} World") + ); + } + } + + #[test] + fn qwen3_endpoint_requires_wss() { + assert!(endpoint_scheme_is_secure_websocket(DEFAULT_ENDPOINT)); + assert!(!endpoint_scheme_is_secure_websocket( + "ws://localhost:9000/realtime" + )); + assert!(!endpoint_scheme_is_secure_websocket( + "https://api.example.com/realtime" + )); + } + #[test] fn join_segments_skips_empty() { let segments = vec!["".to_string(), "内容".to_string(), " ".to_string()]; diff --git a/openless-all/app/src-tauri/src/commands/providers.rs b/openless-all/app/src-tauri/src/commands/providers.rs index 31cd7a87..1823e3a1 100644 --- a/openless-all/app/src-tauri/src/commands/providers.rs +++ b/openless-all/app/src-tauri/src/commands/providers.rs @@ -37,8 +37,7 @@ pub async fn list_provider_models(kind: String) -> Result Result<(), String> { .map_err(|e| e.to_string())? .filter(|s| !s.trim().is_empty()) .unwrap_or_else(|| crate::asr::qwen_realtime::DEFAULT_ENDPOINT.to_string()); - if !crate::asr::bailian::endpoint_scheme_is_websocket(&endpoint) { + if !crate::asr::qwen_realtime::endpoint_scheme_is_secure_websocket(&endpoint) { return Err("qwen3EndpointSchemeInvalid".to_string()); } let model = CredentialsVault::get(CredentialAccount::AsrModel)