diff --git a/openless-all/app/src-tauri/src/coordinator.rs b/openless-all/app/src-tauri/src/coordinator.rs index c62db45f..773d9556 100644 --- a/openless-all/app/src-tauri/src/coordinator.rs +++ b/openless-all/app/src-tauri/src/coordinator.rs @@ -56,6 +56,24 @@ enum ActiveAsr { Whisper(Arc), } +struct SessionResource { + session_id: u64, + resource: T, +} + +impl SessionResource { + fn new(session_id: u64, resource: T) -> Self { + Self { + session_id, + resource, + } + } + + fn into_inner(self) -> T { + self.resource + } +} + struct SessionState { phase: SessionPhase, started_at: Instant, @@ -104,8 +122,8 @@ struct Inner { #[cfg(target_os = "windows")] prepared_windows_ime_session: Arc>>, state: Mutex, - asr: Mutex>, - recorder: Mutex>, + asr: Mutex>>, + recorder: Mutex>>, hotkey: Mutex>, hotkey_status: Mutex, hotkey_trigger_held: AtomicBool, @@ -755,15 +773,72 @@ fn request_stop_during_starting(inner: &Arc, reason: &str) { stop_recorder_if_pending_start_stop(inner); } +fn take_session_resource(slot: &mut Option>, session_id: u64) -> Option { + if slot + .as_ref() + .map(|resource| resource.session_id == session_id) + .unwrap_or(false) + { + slot.take().map(SessionResource::into_inner) + } else { + None + } +} + +fn store_asr_for_session(inner: &Arc, session_id: u64, asr: ActiveAsr) { + *inner.asr.lock() = Some(SessionResource::new(session_id, asr)); +} + +fn take_asr_for_session(inner: &Arc, session_id: u64) -> Option { + let mut slot = inner.asr.lock(); + take_session_resource(&mut slot, session_id) +} + +fn cancel_active_asr(asr: ActiveAsr) { + match asr { + ActiveAsr::Volcengine(v) => v.cancel(), + ActiveAsr::Whisper(w) => w.cancel(), + } +} + +fn cancel_asr_for_session(inner: &Arc, session_id: u64) { + if let Some(asr) = take_asr_for_session(inner, session_id) { + cancel_active_asr(asr); + } +} + +fn store_recorder_for_session(inner: &Arc, session_id: u64, recorder: Recorder) { + *inner.recorder.lock() = Some(SessionResource::new(session_id, recorder)); +} + +fn take_recorder_for_session(inner: &Arc, session_id: u64) -> Option { + let mut slot = inner.recorder.lock(); + take_session_resource(&mut slot, session_id) +} + +fn stop_recorder_for_session(inner: &Arc, session_id: u64) { + if let Some(recorder) = take_recorder_for_session(inner, session_id) { + recorder.stop(); + } +} + +fn discard_startup_resources_for_session(inner: &Arc, session_id: u64) { + stop_recorder_for_session(inner, session_id); + cancel_asr_for_session(inner, session_id); +} + fn stop_recorder_if_pending_start_stop(inner: &Arc) { - let should_stop = { + let (should_stop, session_id) = { let state = inner.state.lock(); - state.phase == SessionPhase::Starting && state.pending_stop + ( + state.phase == SessionPhase::Starting && state.pending_stop, + state.session_id, + ) }; if !should_stop { return; } - if let Some(rec) = inner.recorder.lock().take() { + if let Some(rec) = take_recorder_for_session(inner, session_id) { rec.stop(); let elapsed = inner.state.lock().started_at.elapsed().as_millis() as u64; emit_capsule(inner, CapsuleState::Transcribing, 0.0, elapsed, None, None); @@ -938,7 +1013,11 @@ async fn begin_session(inner: &Arc) -> Result<(), String> { if active_asr == "whisper" { let (api_key, base_url, model) = read_whisper_credentials(); let whisper = Arc::new(WhisperBatchASR::new(api_key, base_url, model)); - *inner.asr.lock() = Some(ActiveAsr::Whisper(Arc::clone(&whisper))); + store_asr_for_session( + inner, + current_session_id, + ActiveAsr::Whisper(Arc::clone(&whisper)), + ); let consumer: Arc = whisper; start_recorder_and_enter_listening(inner, current_session_id, &active_asr, consumer) .await?; @@ -948,7 +1027,11 @@ async fn begin_session(inner: &Arc) -> Result<(), String> { let asr = Arc::new(VolcengineStreamingASR::new(creds, hotwords)); let bridge = Arc::new(DeferredAsrBridge::new()); let consumer: Arc = bridge.clone(); - *inner.asr.lock() = Some(ActiveAsr::Volcengine(Arc::clone(&asr))); + store_asr_for_session( + inner, + current_session_id, + ActiveAsr::Volcengine(Arc::clone(&asr)), + ); start_recorder_for_starting(inner, current_session_id, &active_asr, consumer)?; if let Err(e) = asr.open_session().await { @@ -959,26 +1042,20 @@ async fn begin_session(inner: &Arc) -> Result<(), String> { "[coord] stale 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 => {} } - if let Some(rec) = inner.recorder.lock().take() { - rec.stop(); - } - if let Some(asr) = inner.asr.lock().take() { - match asr { - ActiveAsr::Volcengine(v) => v.cancel(), - ActiveAsr::Whisper(w) => w.cancel(), - } - } + discard_startup_resources_for_session(inner, current_session_id); emit_capsule( inner, CapsuleState::Error, @@ -1000,9 +1077,7 @@ async fn begin_session(inner: &Arc) -> Result<(), String> { StartupRaceStatus::CancelRaced => { log::info!("[coord] cancel raced during ASR open_session — aborting begin"); asr.cancel(); - if let Some(rec) = inner.recorder.lock().take() { - rec.stop(); - } + 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(()); @@ -1012,6 +1087,7 @@ async fn begin_session(inner: &Arc) -> Result<(), String> { "[coord] stale 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(()); } @@ -1070,19 +1146,14 @@ fn start_recorder_for_starting( match Recorder::start(consumer, level_handler) { Ok((rec, runtime_errors)) => { - *inner.recorder.lock() = Some(rec); + store_recorder_for_session(inner, session_id, rec); spawn_recorder_error_monitor(inner, runtime_errors); stop_recorder_if_pending_start_stop(inner); log::info!("[coord] recorder started (asr={active_asr}, phase=Starting)"); } Err(e) => { log::error!("[coord] recorder start failed: {e}"); - if let Some(asr) = inner.asr.lock().take() { - match asr { - ActiveAsr::Volcengine(v) => v.cancel(), - ActiveAsr::Whisper(w) => w.cancel(), - } - } + cancel_asr_for_session(inner, session_id); emit_capsule( inner, CapsuleState::Error, @@ -1173,15 +1244,7 @@ fn abort_recording_with_error(inner: &Arc, message: String) { ) }; - if let Some(rec) = inner.recorder.lock().take() { - rec.stop(); - } - if let Some(asr) = inner.asr.lock().take() { - match asr { - ActiveAsr::Volcengine(v) => v.cancel(), - ActiveAsr::Whisper(w) => w.cancel(), - } - } + discard_startup_resources_for_session(inner, session_id); restore_prepared_windows_ime_session(inner, session_id); emit_capsule( @@ -1231,19 +1294,12 @@ async fn finish_starting_session(inner: &Arc, session_id: u64) { log::info!( "[coord] stale recorder/ASR startup continuation from session {session_id} — ignoring" ); + discard_startup_resources_for_session(inner, session_id); restore_prepared_windows_ime_session(inner, session_id); } BeginOutcome::CancelRaced => { log::info!("[coord] cancel raced during recorder/ASR startup — aborting begin"); - if let Some(rec) = inner.recorder.lock().take() { - rec.stop(); - } - if let Some(asr) = inner.asr.lock().take() { - match asr { - ActiveAsr::Volcengine(v) => v.cancel(), - ActiveAsr::Whisper(w) => w.cancel(), - } - } + discard_startup_resources_for_session(inner, session_id); restore_prepared_windows_ime_session(inner, session_id); set_phase_idle_if_session_matches(inner, session_id); } @@ -1270,11 +1326,11 @@ async fn end_session(inner: &Arc) -> Result<(), String> { let elapsed = inner.state.lock().started_at.elapsed().as_millis() as u64; emit_capsule(inner, CapsuleState::Transcribing, 0.0, elapsed, None, None); - if let Some(rec) = inner.recorder.lock().take() { + if let Some(rec) = take_recorder_for_session(inner, current_session_id) { rec.stop(); } - let asr_opt = inner.asr.lock().take(); + let asr_opt = take_asr_for_session(inner, current_session_id); let asr = match asr_opt { Some(a) => a, None => { @@ -1581,15 +1637,8 @@ fn cancel_session(inner: &Arc) { (phase, state.session_id) }; - if let Some(rec) = inner.recorder.lock().take() { - rec.stop(); - } - if let Some(asr) = inner.asr.lock().take() { - match asr { - ActiveAsr::Volcengine(v) => v.cancel(), - ActiveAsr::Whisper(w) => w.cancel(), - } - } + stop_recorder_for_session(inner, session_id); + cancel_asr_for_session(inner, session_id); restore_prepared_windows_ime_session(inner, session_id); // Processing 阶段保持 phase=Processing 让 end_session 自己走完检查 + 收尾； // 其他阶段直接转 Idle。 @@ -2622,6 +2671,36 @@ mod tests { StartupRaceStatus::StaleContinuation ); } + + #[test] + fn stale_startup_cleanup_keeps_newer_asr_resource() { + let coordinator = Coordinator::new(); + let newer_asr = Arc::new(WhisperBatchASR::new( + "key".to_string(), + "http://localhost".to_string(), + "model".to_string(), + )); + *coordinator.inner.asr.lock() = Some(SessionResource::new( + 2, + ActiveAsr::Whisper(Arc::clone(&newer_asr)), + )); + + discard_startup_resources_for_session(&coordinator.inner, 1); + + assert_eq!( + coordinator + .inner + .asr + .lock() + .as_ref() + .map(|resource| resource.session_id), + Some(2) + ); + + discard_startup_resources_for_session(&coordinator.inner, 2); + + assert!(coordinator.inner.asr.lock().is_none()); + } } fn enabled_phrases(inner: &Arc) -> Vec { diff --git a/openless-all/app/src-tauri/src/windows_ime_profile.rs b/openless-all/app/src-tauri/src/windows_ime_profile.rs index 253af515..57d55c5f 100644 --- a/openless-all/app/src-tauri/src/windows_ime_profile.rs +++ b/openless-all/app/src-tauri/src/windows_ime_profile.rs @@ -243,14 +243,31 @@ mod windows_impl { if profile.dwProfileType == TF_PROFILETYPE_INPUTPROCESSOR { Ok(ImeProfileSnapshot::text_service( profile.langid, - format!("{:?}", profile.clsid), - format!("{:?}", profile.guidProfile), + guid_to_braced_string(profile.clsid), + guid_to_braced_string(profile.guidProfile), )) } else { keyboard_layout_snapshot_from_tsf(profile.langid, profile.hkl) } } + pub(super) fn guid_to_braced_string(guid: GUID) -> String { + format!( + "{{{:08X}-{:04X}-{:04X}-{:02X}{:02X}-{:02X}{:02X}{:02X}{:02X}{:02X}{:02X}}}", + guid.data1, + guid.data2, + guid.data3, + guid.data4[0], + guid.data4[1], + guid.data4[2], + guid.data4[3], + guid.data4[4], + guid.data4[5], + guid.data4[6], + guid.data4[7], + ) + } + pub fn activate_openless_profile() -> WindowsImeProfileResult<()> { let clsid = parse_guid(OPENLESS_TEXT_SERVICE_CLSID_BRACED)?; let profile_guid = parse_guid(OPENLESS_PROFILE_GUID_BRACED)?; @@ -657,4 +674,13 @@ mod windows_tests { .to_string() .contains("active keyboard layout profile has no HKL")); } + + #[test] + fn guid_snapshot_strings_are_canonical_and_parseable() { + let guid = windows::core::GUID::from_u128(0x6b9f3f4f_5ee7_42d6_9c61_9f80b03a5d7d); + let formatted = windows_impl::guid_to_braced_string(guid); + + assert_eq!(formatted, "{6B9F3F4F-5EE7-42D6-9C61-9F80B03A5D7D}"); + assert!(parse_guid(&formatted).is_ok()); + } }