feat(dictation): recover Esc-cancelled recordings

openless-all/app/src-tauri/src/asr/qwen_realtime.rs

@@ -121,6 +121,9 @@ struct SyncState {
     pending_audio: Vec<u8>,
     audio_scratch: Vec<u8>,
     bytes_received: u64,
+    /// 已经真正写入 WebSocket 的 PCM 字节数。恢复录音时旧音频会一次性进入发送队列，
+    /// 需要用它估算队列还要按实时节奏排空多久，避免固定 12s 收尾超时提前中断。
+    bytes_sent: u64,
     session_started: bool,
     session_finished: bool,
     session_start_error: Option<String>,
@@ -201,9 +204,21 @@ impl Qwen3RealtimeASR {
         let writer_for_worker = Arc::clone(&self.writer);
         let weak_self_for_worker = Arc::downgrade(self);
         tokio::spawn(async move {
+            // Qwen realtime 按实时流消费音频。正常录音时 recorder 本来就约每 100ms
+            // 送来一帧；恢复录音则会把之前保存的 PCM 一次性灌进队列。若不节流，这批
+            // 帧会在几毫秒内全部写进 WebSocket，服务端会把它当成超实时突发而漏掉旧段。
+            let mut next_audio_send_at: Option<tokio::time::Instant> = None;
             while let Some(item) = send_rx.recv().await {
                 match item {
                     SendItem::Audio(chunk) => {
+                        let now = tokio::time::Instant::now();
+                        let frame_started_at = match next_audio_send_at {
+                            Some(deadline) if deadline > now => {
+                                tokio::time::sleep_until(deadline).await;
+                                deadline
+                            }
+                            _ => now,
+                        };
                         if let Err(e) =
                             send_text(&writer_for_worker, append_audio_message(&chunk)).await
                         {
@@ -213,6 +228,15 @@ impl Qwen3RealtimeASR {
                             }
                             break;
                         }
+                        if let Some(this) = weak_self_for_worker.upgrade() {
+                            let mut st = this.state.lock();
+                            st.bytes_sent = st.bytes_sent.saturating_add(chunk.len() as u64);
+                        }
+                        // 以上一帧的计划时刻为基准，避免把每次 WebSocket write 的耗时
+                        // 累加进节拍；网络偶发卡顿时则从当前时刻重新起算，不追赶突发。
+                        next_audio_send_at = Some(
+                            frame_started_at + realtime_audio_duration(chunk.len() as u64),
+                        );
                     }
                     SendItem::Finish(done) => {
                         let result = send_text(&writer_for_worker, finish_session_message())
@@ -284,7 +308,17 @@ impl Qwen3RealtimeASR {
     }
 
     pub async fn send_last_frame(&self) -> Result<(), Qwen3ASRError> {
-        let result = tokio::time::timeout(FINAL_RESULT_TIMEOUT, async {
+        let (finish_timeout, pending_audio_bytes) = {
+            let st = self.state.lock();
+            let pending = st.bytes_received.saturating_sub(st.bytes_sent);
+            (final_result_timeout(pending), pending)
+        };
+        if pending_audio_bytes > TARGET_AUDIO_CHUNK_BYTES as u64 {
+            log::info!(
+                "[qwen3-asr] draining {pending_audio_bytes} queued audio bytes at realtime cadence before finish"
+            );
+        }
+        let result = tokio::time::timeout(finish_timeout, async {
             let finished = self.session_finished.notified();
             tokio::pin!(finished);
             finished.as_mut().enable();
@@ -585,6 +619,16 @@ fn drain_audio_chunks(buffer: &mut Vec<u8>) -> Vec<Vec<u8>> {
     chunks
 }
 
+/// PCM 为 16kHz / 16-bit / mono，即每毫秒 32 字节。向上取整以免尾帧得到 0ms。
+fn realtime_audio_duration(bytes: u64) -> Duration {
+    Duration::from_millis(bytes.saturating_add(BYTES_PER_MS - 1) / BYTES_PER_MS)
+}
+
+/// 固定的服务端收尾窗口之外，为尚未发出的音频保留完整实时播放时长。
+fn final_result_timeout(pending_audio_bytes: u64) -> Duration {
+    FINAL_RESULT_TIMEOUT.saturating_add(realtime_audio_duration(pending_audio_bytes))
+}
+
 /// VAD 句段拼接：CJK 之间直接相连；拉丁词之间补空格，避免英文句段黏连。
 /// `stepfun_realtime` 的多句段收尾复用同一套拼接逻辑，故 `pub(crate)`。
 pub(crate) fn join_segments(segments: &[String]) -> String {
@@ -978,4 +1022,21 @@ mod tests {
         assert_eq!(chunks.len(), 2);
         assert_eq!(buffer.len(), 17);
     }
+
+    #[test]
+    fn realtime_pacing_uses_pcm_duration() {
+        assert_eq!(
+            realtime_audio_duration(TARGET_AUDIO_CHUNK_BYTES as u64),
+            Duration::from_millis(100)
+        );
+        assert_eq!(realtime_audio_duration(1), Duration::from_millis(1));
+    }
+
+    #[test]
+    fn finish_timeout_includes_queued_replay_duration() {
+        assert_eq!(
+            final_result_timeout(64_000),
+            FINAL_RESULT_TIMEOUT + Duration::from_secs(2)
+        );
+    }
 }

openless-all/app/src-tauri/src/commands/dictation.rs

@@ -15,6 +15,22 @@ pub fn cancel_dictation(coord: CoordinatorState<'_>) {
     coord.cancel_dictation();
 }
 
+#[tauri::command]
+pub async fn resume_cancelled_recording(
+    coord: CoordinatorState<'_>,
+    session_id: String,
+) -> Result<(), String> {
+    coord.resume_cancelled_recording(&session_id).await
+}
+
+#[tauri::command]
+pub fn dismiss_cancelled_recording_recovery(
+    coord: CoordinatorState<'_>,
+    session_id: Option<String>,
+) {
+    coord.dismiss_cancelled_recording_recovery(session_id.as_deref());
+}
+
 #[tauri::command]
 pub async fn handle_window_hotkey_event(
     coord: CoordinatorState<'_>,

openless-all/app/src-tauri/src/coordinator.rs

@@ -32,9 +32,9 @@ use crate::asr::{
 use crate::combo_hotkey::{ComboHotkeyError, ComboHotkeyEvent, ComboHotkeyMonitor};
 use crate::coordinator_state::{
     begin_cancel_session_state, begin_recording_abort_before_restore, begin_session_state,
-    finish_cancel_session_state, finish_starting_session_state, new_session_id,
-    publish_abort_idle_after_restore, start_processing_if_listening, startup_race_status,
-    BeginOutcome, SessionId, SessionPhase, SessionState, StartupRaceStatus,
+    begin_session_state_with_id, finish_cancel_session_state, finish_starting_session_state,
+    new_session_id, publish_abort_idle_after_restore, start_processing_if_listening,
+    startup_race_status, BeginOutcome, SessionId, SessionPhase, SessionState, StartupRaceStatus,
 };
 use crate::correction::apply_correction_rules;
 use crate::hotkey::{HotkeyEvent, HotkeyMonitor};
@@ -102,8 +102,9 @@ pub(super) fn qa_event_target() -> &'static str {
 #[cfg(test)]
 use dictation::dictation_error_code;
 use dictation::{
-    begin_session, begin_session_as, cancel_session, end_session, handle_pressed_edge,
-    handle_released_edge, handle_trigger_combined, request_stop_during_starting,
+    begin_session, begin_session_as, cancel_session, cancel_session_after_escape, end_session,
+    handle_pressed_edge, handle_released_edge, handle_trigger_combined,
+    request_stop_during_starting,
 };
 #[cfg(any(debug_assertions, test))]
 use dictation::{handle_pressed, handle_released};
@@ -1030,11 +1031,17 @@ struct Inner {
     #[cfg(target_os = "windows")]
     sherpa_onnx_runtime: Arc<SherpaOnnxRuntime>,
     recorder: Mutex<Option<SessionResource<Recorder>>>,
+    /// 恢复被 Esc 打断的录音时，旧 WAV 中的 PCM 前缀。新 ASR 启动后先消费这段，
+    /// Recorder 同时把它写回同一个归档，再追加新的麦克风音频。
+    resume_audio_pcm: Mutex<Option<SessionResource<Vec<u8>>>>,
     /// 当前 dictation / QA session 的 wav 归档是否真的被写到磁盘上。
     /// 由 Recorder::start 返回值 (archive_active) 写入；history.append 路径读取，
     /// 决定 DictationSession.has_audio_recording 字段。比单纯读 prefs.record_audio_for_debug
     /// 更准确：用户开了开关但路径无法创建（权限 / 磁盘满）也算 false。
     audio_archive_active: AtomicBool,
+    /// 当前 3 秒「是否继续」提示对应的录音 id。只影响 CapsulePayload；历史记录和 WAV
+    /// 已在显示提示前持久化，因此提示消失或应用退出都不会丢失恢复入口。
+    cancelled_recording_recovery: Mutex<Option<String>>,
     /// 上一次落字之后武装的手改监听（macOS）。
     ///
     /// 存在 `Inner` 上只为了「下一次听写开始时解除上一次的」这一条生命周期规则 ——
@@ -1407,7 +1414,9 @@ impl Coordinator {
                     asr_label: Mutex::new(None),
                     omni_pcm: Mutex::new(None),
                     recorder: Mutex::new(None),
+                    resume_audio_pcm: Mutex::new(None),
                     audio_archive_active: AtomicBool::new(false),
+                    cancelled_recording_recovery: Mutex::new(None),
                     edit_watcher: Mutex::new(None),
                     edit_watch_generation: std::sync::atomic::AtomicU64::new(0),
                     pending_corrections: Mutex::new(Vec::new()),
@@ -1539,7 +1548,9 @@ impl Coordinator {
                 asr_label: Mutex::new(None),
                 omni_pcm: Mutex::new(None),
                 recorder: Mutex::new(None),
+                resume_audio_pcm: Mutex::new(None),
                 audio_archive_active: AtomicBool::new(false),
+                cancelled_recording_recovery: Mutex::new(None),
                 edit_watcher: Mutex::new(None),
                 edit_watch_generation: std::sync::atomic::AtomicU64::new(0),
                 pending_corrections: Mutex::new(Vec::new()),
@@ -2507,6 +2518,98 @@ impl Coordinator {
         cancel_session(&self.inner);
     }
 
+    /// 从历史或 3 秒浮层恢复一条被 Esc 打断的录音。旧 PCM 会在新麦克风启动前喂给
+    /// 当前识别器，并作为新 WAV 的前缀，因此后续停止时处理的是完整语音。
+    pub async fn resume_cancelled_recording(&self, session_id: &str) -> Result<(), String> {
+        if self.inner.state.lock().phase != SessionPhase::Idle {
+            return Err("another dictation session is active".into());
+        }
+        let entry = self
+            .inner
+            .history
+            .list()
+            .map_err(|error| error.to_string())?
+            .into_iter()
+            .find(|entry| entry.id == session_id)
+            .ok_or_else(|| "history entry not found".to_string())?;
+        if entry.error_code.as_deref() != Some("recordingCancelled") {
+            return Err("history entry is not a cancelled recording".into());
+        }
+        if entry.has_audio_recording != Some(true) {
+            return Err("recording not found".into());
+        }
+        let parsed_id =
+            uuid::Uuid::parse_str(session_id).map_err(|_| "invalid session id".to_string())?;
+        let path = crate::persistence::recording_path_for_session(session_id)
+            .map_err(|error| error.to_string())?;
+        let wav = tokio::fs::read(path).await.map_err(|error| {
+            if error.kind() == std::io::ErrorKind::NotFound {
+                "recording not found".to_string()
+            } else {
+                format!("read wav failed: {error}")
+            }
+        })?;
+        if wav.len() <= 44 {
+            return Err("recording is empty or corrupt".into());
+        }
+        let pcm = wav[44..].to_vec();
+        // 先移除占位条目，再开启同 id 的新会话。若等会话已经启动后才删，用户在这段
+        // 极短窗口里再次按 Esc，取消路径会先 upsert 新占位，随后这里却会把它误删。
+        // 启动失败时恢复原条目，保证历史不会因恢复尝试而丢失。
+        self.inner
+            .history
+            .delete(session_id)
+            .map_err(|error| format!("remove history placeholder failed: {error}"))?;
+        *self.inner.cancelled_recording_recovery.lock() = None;
+        let resume_result = dictation::resume_cancelled_recording(
+            &self.inner,
+            parsed_id,
+            entry.duration_ms.unwrap_or_default(),
+            pcm,
+        )
+        .await;
+
+        let resumed = {
+            let state = self.inner.state.lock();
+            state.session_id == parsed_id
+                && matches!(
+                    state.phase,
+                    SessionPhase::Starting | SessionPhase::Listening
+                )
+        };
+        if let Err(error) = resume_result {
+            let prefs = self.inner.prefs.get();
+            if let Err(restore_error) = self.inner.history.insert_if_missing_with_retention(
+                entry,
+                prefs.history_retention_days,
+                prefs.history_max_entries,
+            ) {
+                log::error!(
+                    "[coord] failed to restore cancelled recording after resume error: {restore_error}"
+                );
+            }
+            return Err(error);
+        }
+        if !resumed {
+            let prefs = self.inner.prefs.get();
+            if let Err(error) = self.inner.history.insert_if_missing_with_retention(
+                entry,
+                prefs.history_retention_days,
+                prefs.history_max_entries,
+            ) {
+                log::error!(
+                    "[coord] failed to restore cancelled recording after resume race: {error}"
+                );
+            }
+            return Err("cancelled recording could not be resumed".into());
+        }
+        Ok(())
+    }
+
+    pub fn dismiss_cancelled_recording_recovery(&self, session_id: Option<&str>) {
+        dictation::dismiss_cancelled_recording_recovery(&self.inner, session_id);
+    }
+
     #[cfg(not(mobile))]
     pub fn set_remote_no_insert(&self, no_insert: bool) {
         self.inner
@@ -5942,6 +6045,11 @@ const CAPSULE_AUTO_HIDE_DELAY_MS: u64 = 2000;
 /// 不需要像 Done/Error 那样停留 2 秒给用户读——立刻回 Idle，由前端 capsule-out
 /// 淡出动画（520ms）负责优雅收尾，观感上「按下即消失」（对齐 Typeless）。
 const CAPSULE_CANCEL_HIDE_DELAY_MS: u64 = 0;
+/// 给连续双击 Esc 留出的判定窗口。期间录音已经停止、WAV 与历史已经保存，只延后展示
+/// 恢复入口；第二次 Esc 会直接清掉入口，因此不会发生浮层闪现。
+const CANCELLED_RECORDING_RECOVERY_PROMPT_DELAY_MS: u64 = 240;
+/// Esc 录音恢复提示停留时间。历史/WAV 在提示出现前已经保存，超时只收起提示。
+const CANCELLED_RECORDING_RECOVERY_TIMEOUT_MS: u64 = 3_000;
 
 /// Toggle 模式下，end_session 将 phase 设为 Idle 后在此时间内禁止新的 begin_session。
 /// 避免用户三连按时第 3 次按下误激活新听写（此时胶囊仍在离场动画周期内）。