feat(dictation): recover Esc-cancelled recordings

openless-all/app/crates/openless-core/src/asr/qwen_realtime.rs

@@ -123,6 +123,8 @@ struct SyncState {
     pending_audio: Vec<u8>,
     audio_scratch: Vec<u8>,
     bytes_received: u64,
+    /// 恢复录音会一次性排入旧 PCM；记录实际发送量，给收尾超时补足实时回放时长。
+    bytes_sent: u64,
     session_started: bool,
     session_finished: bool,
     session_start_error: Option<String>,
@@ -217,9 +219,18 @@ impl Qwen3RealtimeASR {
         let weak_self_for_worker = Arc::downgrade(self);
         let task_spawner = Arc::clone(&self.task_spawner);
         task_spawner.spawn(Box::pin(async move {
+            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
                         {
@@ -229,6 +240,12 @@ impl Qwen3RealtimeASR {
                             }
                             break;
                         }
+                        if let Some(this) = weak_self_for_worker.upgrade() {
+                            let mut state = this.state.lock();
+                            state.bytes_sent = state.bytes_sent.saturating_add(chunk.len() as u64);
+                        }
+                        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())
@@ -301,7 +318,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 state = self.state.lock();
+            let pending = state.bytes_received.saturating_sub(state.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();
@@ -607,6 +634,15 @@ 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 {
@@ -1004,4 +1040,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/crates/openless-core/src/audio.rs

@@ -57,6 +57,20 @@ pub fn encode_dictation_wav(pcm_i16_le: &[u8]) -> Result<Vec<u8>, crate::Backend
     Ok(wav)
 }
 
+pub fn decode_dictation_wav(wav: &[u8]) -> Result<Vec<u8>, crate::BackendError> {
+    if wav.len() <= 44
+        || &wav[..4] != b"RIFF"
+        || &wav[8..12] != b"WAVE"
+        || !(wav.len() - 44).is_multiple_of(2)
+    {
+        return Err(crate::BackendError::new(
+            crate::BackendErrorCode::Persistence,
+            "dictation recording archive is not canonical PCM WAV",
+        ));
+    }
+    Ok(wav[44..].to_vec())
+}
+
 impl PcmNormalizer {
     pub fn process(
         &mut self,
@@ -188,9 +202,14 @@ mod tests {
         assert_eq!(&wav[8..12], b"WAVE");
         assert_eq!(u32::from_le_bytes(wav[24..28].try_into().unwrap()), 16_000);
         assert_eq!(&wav[44..], &pcm);
+        assert_eq!(decode_dictation_wav(&wav).unwrap(), pcm);
         assert_eq!(
             encode_dictation_wav(&[1]).unwrap_err().code,
             crate::BackendErrorCode::InvalidArgument
         );
+        assert_eq!(
+            decode_dictation_wav(&wav[..44]).unwrap_err().code,
+            crate::BackendErrorCode::Persistence
+        );
     }
 }

openless-all/app/crates/openless-core/src/cloud_providers.rs

@@ -1730,6 +1730,9 @@ impl DictationEngine for SharedOmniDictationEngine {
                 }
                 return Ok(());
             }
+            if let Some(prefix) = context.recording.prefix_pcm.as_deref() {
+                pcm.consume_pcm_chunk(prefix);
+            }
             let consumer: Arc<dyn AudioConsumer> = pcm;
             let level_progress: Arc<dyn RecordingProgressSink> = Arc::new(OmniRecordingProgress {
                 session_id,
@@ -1996,19 +1999,29 @@ impl DictationEngine for SharedOmniDictationEngine {
     }
 
     fn cancel(&self, session_id: SessionId) -> BoxFuture<'static, Result<(), BackendError>> {
+        let cancelled = self.cancel_preserving_archive(session_id);
+        Box::pin(async move { cancelled.await.map(|_| ()) })
+    }
+
+    fn cancel_preserving_archive(
+        &self,
+        session_id: SessionId,
+    ) -> BoxFuture<'static, Result<Option<Arc<dyn crate::ports::RecordingArchive>>, BackendError>>
+    {
         let session = self.sessions.lock().get(&session_id).cloned();
         let sessions = Arc::clone(&self.sessions);
         Box::pin(async move {
             let Some(session) = session else {
-                return Ok(());
+                return Ok(None);
             };
             session.cancellation.cancel();
             let recording = session.recording.lock().take();
+            let archive = recording.as_ref().and_then(|recording| recording.archive());
             if let Some(recording) = recording {
                 recording.stop().await?;
             }
             remove_omni_session(&sessions, session_id, &session);
-            Ok(())
+            Ok(archive.filter(|archive| archive.is_available()))
         })
     }
 }

openless-all/app/crates/openless-core/src/dictation_context.rs

@@ -133,6 +133,9 @@ pub struct DictationInsertionContext {
 pub struct RecordingPlan {
     pub microphone_device_name: Option<String>,
     pub mute_during_recording: bool,
+    /// PCM from a recoverable cancelled recording, replayed before live input.
+    pub prefix_pcm: Option<Vec<u8>>,
+    pub elapsed_offset_ms: u64,
     /// Whether the Host may create an audio archive at all. QA/Selection Voice
     /// keep PCM in memory; successful-recording retention is a separate policy.
     pub archive_enabled: bool,
@@ -246,6 +249,8 @@ impl DictationContext {
             recording: RecordingPlan {
                 microphone_device_name: non_blank(&preferences.microphone_device_name),
                 mute_during_recording: preferences.mute_during_recording,
+                prefix_pcm: None,
+                elapsed_offset_ms: 0,
                 archive_enabled: true,
                 archive_successful_recording: preferences.record_audio_for_debug,
                 retention_days: preferences.history_retention_days,

openless-all/app/crates/openless-core/src/dictation_engine.rs

@@ -172,6 +172,9 @@ impl DictationEngine for PipelineDictationEngine {
             let audio_consumer: Arc<dyn AudioConsumer> = Arc::new(SessionAudioConsumer {
                 session: Arc::clone(&transcription),
             });
+            if let Some(prefix) = context.recording.prefix_pcm.as_deref() {
+                audio_consumer.consume_pcm_chunk(prefix);
+            }
             let recording_progress: Arc<dyn RecordingProgressSink> =
                 Arc::new(RecordingProgressForwarder {
                     session_id,
@@ -618,6 +621,15 @@ impl DictationEngine for PipelineDictationEngine {
     }
 
     fn cancel(&self, session_id: SessionId) -> BoxFuture<'static, Result<(), BackendError>> {
+        let cancelled = self.cancel_preserving_archive(session_id);
+        Box::pin(async move { cancelled.await.map(|_| ()) })
+    }
+
+    fn cancel_preserving_archive(
+        &self,
+        session_id: SessionId,
+    ) -> BoxFuture<'static, Result<Option<Arc<dyn crate::ports::RecordingArchive>>, BackendError>>
+    {
         let sessions = Arc::clone(&self.sessions);
         let polisher = Arc::clone(&self.polisher);
         Box::pin(async move {
@@ -627,10 +639,10 @@ impl DictationEngine for PipelineDictationEngine {
                 .get(&session_id)
                 .cloned()
             else {
-                return Ok(());
+                return Ok(None);
             };
             if session.cancelled.swap(true, Ordering::AcqRel) {
-                return Ok(());
+                return Ok(None);
             }
 
             let (recording, transcription) = {
@@ -640,6 +652,7 @@ impl DictationEngine for PipelineDictationEngine {
                     .expect("pipeline resource lock poisoned");
                 (resources.recording.take(), resources.transcription.clone())
             };
+            let archive = recording.as_ref().and_then(|recording| recording.archive());
             let mut first_error = None;
             if let Some(recording) = recording {
                 retain_first_error(&mut first_error, recording.stop().await);
@@ -659,7 +672,7 @@ impl DictationEngine for PipelineDictationEngine {
             remove_session(&sessions, session_id, &session);
             match first_error {
                 Some(error) => Err(error),
-                None => Ok(()),
+                None => Ok(archive.filter(|archive| archive.is_available())),
             }
         })
     }
@@ -1259,6 +1272,35 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn recovery_prefix_reaches_asr_and_cancel_returns_the_archive() {
+        let fixture = fixture_engine(
+            false,
+            Ok(crate::ports::PolishOutput::text("unused")),
+            None,
+            None,
+        );
+        let session_id = SessionId::new();
+        let mut context = DictationContext::default();
+        context.recording.prefix_pcm = Some(vec![9, 0, 8, 0]);
+        fixture
+            .engine
+            .start(session_id, Arc::new(context), fixture.progress.clone())
+            .await
+            .unwrap();
+
+        assert_eq!(*fixture.pcm.lock().unwrap(), vec![9, 0, 8, 0, 1, 0, 2, 0]);
+        let archive = fixture
+            .engine
+            .cancel_preserving_archive(session_id)
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(archive.read_pcm().await.unwrap(), vec![1, 0, 2, 0]);
+        assert_eq!(fixture.recorder_stops.load(Ordering::Acquire), 1);
+        assert_eq!(fixture.transcription_cancels.load(Ordering::Acquire), 1);
+    }
+
     #[tokio::test]
     async fn pipeline_streams_pcm_progress_and_terminal_deltas() {
         let fixture = fixture_engine(

openless-all/app/crates/openless-core/src/history.rs

@@ -40,6 +40,28 @@ impl HistoryStore {
         let _guard = self.lock_store()?;
         let mut sessions = self.read_locked()?;
         sessions.insert(0, session);
+        self.write_with_retention(sessions, retention_days, max_entries)
+    }
+
+    pub fn upsert_with_retention(
+        &self,
+        session: DictationSession,
+        retention_days: u32,
+        max_entries: Option<u32>,
+    ) -> Result<(), BackendError> {
+        let _guard = self.lock_store()?;
+        let mut sessions = self.read_locked()?;
+        sessions.retain(|existing| existing.id != session.id);
+        sessions.insert(0, session);
+        self.write_with_retention(sessions, retention_days, max_entries)
+    }
+
+    fn write_with_retention(
+        &self,
+        mut sessions: Vec<DictationSession>,
+        retention_days: u32,
+        max_entries: Option<u32>,
+    ) -> Result<(), BackendError> {
         if retention_days > 0 {
             let cutoff = chrono::Utc::now() - chrono::Duration::days(i64::from(retention_days));
             sessions.retain(|session| {
@@ -200,4 +222,20 @@ mod tests {
         assert!(store.recent_within_minutes(0).unwrap().is_empty());
         let _ = std::fs::remove_file(path);
     }
+
+    #[test]
+    fn upsert_replaces_the_same_session_without_duplicates() {
+        let path = std::env::temp_dir().join(format!(
+            "openless-core-history-upsert-{}.json",
+            uuid::Uuid::new_v4().simple()
+        ));
+        let store = HistoryStore::at_path(path.clone());
+        let mut entry = session("same", chrono::Utc::now().to_rfc3339());
+        store.upsert_with_retention(entry.clone(), 0, None).unwrap();
+        entry.error_code = Some("recordingCancelled".into());
+        store.upsert_with_retention(entry.clone(), 0, None).unwrap();
+
+        assert_eq!(store.list().unwrap(), vec![entry]);
+        let _ = std::fs::remove_file(path);
+    }
 }

openless-all/app/crates/openless-core/src/lib.rs

@@ -211,7 +211,10 @@ pub use api::{
     DictationHotkeyEdge, LessComputerHotkeyAction, LessComputerVoiceSession, OpenLessBackend,
     QaVoiceCaptureResult, QaVoiceCaptureSession, StartupSnapshot, VoiceTranscriptionSession,
 };
-pub use audio::{encode_dictation_wav, NormalizedPcmChunk, PcmNormalizer, DICTATION_SAMPLE_RATE};
+pub use audio::{
+    decode_dictation_wav, encode_dictation_wav, NormalizedPcmChunk, PcmNormalizer,
+    DICTATION_SAMPLE_RATE,
+};
 pub use auxiliary::{
     AsrCallLabel, AuxiliaryApi, RepolishRequest, RetranscriptionFailure, RetranscriptionResult,
 };
@@ -346,10 +349,10 @@ pub use style_pack_store::{
 pub use style_packs::*;
 pub use types::InsertStatus as DictationInsertStatus;
 pub use types::{
-    CorrectionRule, DictationPhase, DictationResult, DictationSession, DictationStateSnapshot,
-    DictionaryEntry, DownloadProgress, HistoryChange, HistoryInsertStatus, HistorySource,
-    InsertFallbackPayload, NotificationLevel, NotificationPayload, PermissionSnapshot,
-    PermissionState, PolishDelta, PolishMode, PreferencesChange, RuleSource,
+    CorrectionRule, DictationPhase, DictationRecordingRecovery, DictationResult, DictationSession,
+    DictationStateSnapshot, DictionaryEntry, DownloadProgress, HistoryChange, HistoryInsertStatus,
+    HistorySource, InsertFallbackPayload, NotificationLevel, NotificationPayload,
+    PermissionSnapshot, PermissionState, PolishDelta, PolishMode, PreferencesChange, RuleSource,
     SelectionVoiceIntentMode, SelectionVoiceManualIntent, SessionId, StylePackChange,
     TranscriptAccumulator, TranscriptDelta, VocabPreset, VocabPresetStore, VocabularyChange,
 };

openless-all/app/crates/openless-core/src/ports.rs

@@ -299,6 +299,19 @@ pub trait DictationEngine: Send + Sync {
         ))
     }
 
+    /// Cancel while retaining the stopped recording archive for an opt-in
+    /// recovery flow. Engines without archive support keep normal cancellation.
+    fn cancel_preserving_archive(
+        &self,
+        session_id: SessionId,
+    ) -> BoxFuture<'static, Result<Option<Arc<dyn RecordingArchive>>, BackendError>> {
+        let cancelled = self.cancel(session_id);
+        Box::pin(async move {
+            cancelled.await?;
+            Ok(None)
+        })
+    }
+
     fn cancel(&self, session_id: SessionId) -> BoxFuture<'static, Result<(), BackendError>>;
 }