diff --git a/openless-all/app/crates/openless-core/src/api.rs b/openless-all/app/crates/openless-core/src/api.rs index d1feab8f3..d75302e3c 100644 --- a/openless-all/app/crates/openless-core/src/api.rs +++ b/openless-all/app/crates/openless-core/src/api.rs @@ -131,7 +131,6 @@ pub struct LessComputerVoiceSession { less_computer: Arc, request: crate::domains::LessComputerRunRequest, partials: Arc, - received_bytes: AtomicU64, archive_successful_recording: bool, } @@ -358,6 +357,22 @@ impl crate::ports::RecordingProgressSink for LessComputerRecordingProgress { crate::ports::RecordingEvent::Level { elapsed_ms, level } => { self.publish_level(elapsed_ms, level) } + crate::ports::RecordingEvent::LimitReached => { + let session_id = self.session_id; + let less_computer = Arc::clone(&self.less_computer); + let control = Arc::clone(&self.control); + self.task_spawner.spawn(Box::pin(async move { + if less_computer.capture_cancelled(session_id) { + return; + } + if let Err(error) = + control.request(session_id, crate::events::RecordingControlAction::Stop) + { + log::warn!("failed to stop Less Computer capture at PCM limit: {error}"); + } + })); + Ok(()) + } crate::ports::RecordingEvent::Fatal(error) => { let session_id = self.session_id; let less_computer = Arc::clone(&self.less_computer); @@ -500,6 +515,10 @@ impl crate::ports::RecordingProgressSink for QaRecordingProgress { crate::ports::RecordingEvent::Level { elapsed_ms, level } => { self.publish_level(elapsed_ms, level) } + crate::ports::RecordingEvent::LimitReached => { + self.submit_terminal(QaRecordingTerminal::Stop); + Ok(()) + } crate::ports::RecordingEvent::Fatal(error) => { self.submit_terminal(QaRecordingTerminal::Fault(error)); Ok(()) @@ -556,6 +575,18 @@ impl crate::ports::RecordingProgressSink for SelectionVoiceRecordingProgress { crate::ports::RecordingEvent::Level { elapsed_ms, level } => { self.publish_level(elapsed_ms, level) } + crate::ports::RecordingEvent::LimitReached => { + let session_id = self.session_id; + let control = Arc::clone(&self.control); + self.task_spawner.spawn(Box::pin(async move { + if let Err(error) = + control.request(session_id, crate::events::RecordingControlAction::Stop) + { + log::warn!("failed to stop selection voice at PCM limit: {error}"); + } + })); + Ok(()) + } crate::ports::RecordingEvent::Fatal(error) => { let session_id = self.session_id; let selection_voice = Arc::clone(&self.selection_voice); @@ -915,19 +946,6 @@ impl LessComputerVoiceSession { "Less Computer PCM must be non-empty and contain complete 16-bit samples", )); } - const MAX_PCM_BYTES: u64 = 128 * 1024 * 1024; - let next = self - .received_bytes - .fetch_add(pcm.len() as u64, Ordering::AcqRel) - .saturating_add(pcm.len() as u64); - if next > MAX_PCM_BYTES { - self.received_bytes - .fetch_sub(pcm.len() as u64, Ordering::AcqRel); - return Err(BackendError::new( - BackendErrorCode::InvalidArgument, - "Less Computer PCM exceeds the provider limit", - )); - } self.control.transcription.consume_pcm_chunk(pcm); Ok(()) } @@ -1722,6 +1740,29 @@ impl EngineProgressSink for BackendEngineProgress { ); } } + EngineProgress::RecordingLimitReached => { + let state = self.state.read().expect("backend state lock poisoned"); + ensure_active_session(&state, session_id)?; + if !matches!( + state.dictation.phase, + DictationPhase::Starting | DictationPhase::Recording + ) { + return Err(BackendError::new( + BackendErrorCode::InvalidState, + "recording limit arrived after recording stopped", + )); + } + drop(state); + self.events.publish( + Some(session_id), + BackendEventKind::RecordingControlRequested( + crate::events::RecordingControlRequest { + session_id, + action: crate::events::RecordingControlAction::Stop, + }, + ), + ); + } EngineProgress::RecordingFault(error) => { let mut state = self.state.write().expect("backend state lock poisoned"); ensure_active_session(&state, session_id)?; @@ -2487,7 +2528,7 @@ impl OpenLessBackend { session_id, Arc::clone(&context), Arc::clone(&partials) as Arc, - recording_progress, + Arc::clone(&recording_progress) as Arc, resources.cancel.clone(), ), discard_voice_capture, @@ -2500,10 +2541,12 @@ impl OpenLessBackend { match own_voice_start( &self.deps.task_spawner, Arc::clone(&resources), - self.deps.dictation_engine.start_transcription( + Arc::clone(&self.deps.dictation_engine).start_transcription_with_progress( session_id, Arc::clone(&context), Arc::clone(&partials) as Arc, + Arc::clone(&recording_progress) + as Arc, ), |transcription| transcription.cancel(), ) @@ -2576,7 +2619,6 @@ impl OpenLessBackend { less_computer: Arc::clone(&self.deps.services.less_computer), request, partials, - received_bytes: AtomicU64::new(0), archive_successful_recording: context.recording.archive_successful_recording, }) } @@ -6617,6 +6659,7 @@ mod tests { struct VoiceTranscription { pcm: Mutex>, cancelled: std::sync::atomic::AtomicBool, + starts: AtomicU64, } impl crate::ports::AudioConsumer for VoiceTranscription { @@ -6660,6 +6703,7 @@ mod tests { _context: Arc, _partials: Arc, ) -> BoxFuture<'static, Result, BackendError>> { + self.0.starts.fetch_add(1, Ordering::AcqRel); let session: Arc = self.0.clone(); boxed(async move { Ok(session) }) } @@ -6859,6 +6903,7 @@ mod tests { .unwrap(); let mut preferences = backend.get_preferences(); preferences.coding_agent_enabled = true; + preferences.stable_transcription_enabled = true; backend.set_preferences(preferences).unwrap(); let mut events = backend.subscribe(); @@ -6867,6 +6912,7 @@ mod tests { .start_less_computer_voice(session_id, Arc::new(FakeRecordingControl::default())) .await .unwrap(); + assert_eq!(transcription.starts.load(Ordering::Acquire), 0); assert_eq!(host.actions(), vec![HostAction::ShowLessComputer]); assert_eq!( session.feed_pcm(&[]).unwrap_err().code, @@ -6881,6 +6927,7 @@ mod tests { assert_eq!(result.session_id, session_id); assert_eq!(*transcription.pcm.lock().unwrap(), vec![1, 0, 2, 0]); + assert_eq!(transcription.starts.load(Ordering::Acquire), 1); assert!(runtime.request.lock().unwrap().is_some()); let transcript_events = std::iter::from_fn(|| events.try_recv().ok()) .filter(|event| matches!(event.kind, BackendEventKind::TranscriptDelta(_))) @@ -6890,6 +6937,13 @@ mod tests { transcript_events[0].kind, BackendEventKind::TranscriptDelta(crate::TranscriptDelta { is_final: true, .. }) )); + + let cancelled = backend + .start_less_computer_voice(SessionId::new(), Arc::new(FakeRecordingControl::default())) + .await + .unwrap(); + cancelled.cancel().await.unwrap(); + assert_eq!(transcription.starts.load(Ordering::Acquire), 1); } #[tokio::test] @@ -7188,6 +7242,15 @@ mod tests { ))), }; use crate::ports::RecordingProgressSink; + progress + .publish(crate::ports::RecordingEvent::LimitReached) + .unwrap(); + tokio::task::yield_now().await; + assert_eq!( + *control.requests.lock().unwrap(), + vec![(stop_session, crate::events::RecordingControlAction::Stop)] + ); + control.requests.lock().unwrap().clear(); progress.publish_level(10, 0.1).unwrap(); progress.publish_level(20, 0.1).unwrap(); progress.publish_level(30, 0.1).unwrap(); @@ -7276,6 +7339,22 @@ mod tests { assert_eq!(qa.cancels.load(Ordering::Acquire), 0); assert_eq!(qa.faults.load(Ordering::Acquire), 0); + let limit_progress = QaRecordingProgress { + session_id: SessionId::new(), + qa: Arc::clone(&qa) as Arc, + progress: Arc::new(VoiceRecordingProgress), + task_spawner: Arc::new(TokioTaskSpawner), + started_at: std::time::Instant::now(), + silence: Mutex::new(None), + terminal: Mutex::new(QaRecordingTerminalState::default()), + }; + limit_progress.arm(); + limit_progress + .publish(crate::ports::RecordingEvent::LimitReached) + .unwrap(); + tokio::task::yield_now().await; + assert_eq!(qa.stops.load(Ordering::Acquire), 2); + let started_at = std::time::Instant::now(); let no_speech = QaRecordingProgress { session_id: SessionId::new(), @@ -9439,6 +9518,35 @@ mod tests { backend.shutdown().await.unwrap(); } + #[tokio::test] + async fn recording_limit_requests_the_normal_dictation_stop_path() { + let (backend, _) = backend(); + backend.start().await.unwrap(); + let session_id = backend.start_dictation().await.unwrap(); + let mut events = backend.subscribe(); + + backend + .engine_progress_sink() + .publish(session_id, EngineProgress::RecordingLimitReached) + .unwrap(); + + let request = std::iter::from_fn(|| events.try_recv().ok()).find_map(|event| match event { + crate::events::BackendEvent { + session_id: Some(id), + kind: BackendEventKind::RecordingControlRequested(request), + .. + } if id == session_id => Some(request), + _ => None, + }); + assert_eq!( + request.map(|request| request.action), + Some(crate::events::RecordingControlAction::Stop) + ); + + backend.cancel_dictation(Some(session_id)).await.unwrap(); + backend.shutdown().await.unwrap(); + } + #[tokio::test] async fn stop_time_context_update_failure_cancels_and_resets_the_session() { use crate::testing::FixtureEngineAction; diff --git a/openless-all/app/crates/openless-core/src/cloud_providers.rs b/openless-all/app/crates/openless-core/src/cloud_providers.rs index 29c31f088..1f29141b7 100644 --- a/openless-all/app/crates/openless-core/src/cloud_providers.rs +++ b/openless-all/app/crates/openless-core/src/cloud_providers.rs @@ -36,8 +36,8 @@ use crate::errors::{BackendError, BackendErrorCode}; use crate::ports::{ ActiveRecording, AudioConsumer, AudioRecorder, DictationEngine, EngineFailure, EngineFailureStage, EngineProgress, EngineProgressSink, EngineResult, EngineStage, - PolishOutput, RecordingProgressSink, TextPolisher, TextStreamChunk, TextStreamSink, - TranscriptOutput, TranscriptionEngine, TranscriptionSession, + PolishOutput, PreparedTranscription, RecordingProgressSink, TextPolisher, TextStreamChunk, + TextStreamSink, TranscriptOutput, TranscriptionEngine, TranscriptionSession, }; use crate::provider_rules::{ default_asr_endpoint, default_asr_model, default_llm_endpoint, default_llm_model, @@ -114,6 +114,68 @@ impl SharedCloudTranscriptionEngine { } } +const ASR_SNAPSHOT_ACCOUNTS: &[&str] = &[ + ASR_MODEL_ACCOUNT, + ASR_API_KEY_ACCOUNT, + ASR_ENDPOINT_ACCOUNT, + ASR_ADVANCED_CONFIG_ACCOUNT, + ASR_VOCABULARY_ID_ACCOUNT, + crate::credentials::VOLCENGINE_SERVICE_ACCOUNT, + VOLCENGINE_AUTH_MODE_ACCOUNT, + VOLCENGINE_APP_KEY_ACCOUNT, + VOLCENGINE_ACCESS_KEY_ACCOUNT, + VOLCENGINE_API_KEY_ACCOUNT, + VOLCENGINE_RESOURCE_ID_ACCOUNT, + XFYUN_APP_ID_ACCOUNT, + XFYUN_API_KEY_ACCOUNT, + TENCENT_CLOUD_APP_ID_ACCOUNT, + TENCENT_CLOUD_SECRET_ID_ACCOUNT, + TENCENT_CLOUD_SECRET_KEY_ACCOUNT, +]; + +#[derive(Clone)] +struct CloudTranscriptionPreparation { + context: Arc, + provider_type: String, + effective_provider: String, + kind: crate::provider_rules::ActiveAsrProviderKind, + model: String, + api_key: String, + endpoint: String, + advanced_config: crate::provider_rules::AdvancedAsrConfig, + credentials: HashMap<&'static str, Option>, +} + +impl CloudTranscriptionPreparation { + fn value(&self, account: &str) -> Option { + self.credentials.get(account).cloned().flatten() + } +} + +struct PreparedCloudTranscription { + preparation: CloudTranscriptionPreparation, + task_spawner: Arc, +} + +impl PreparedTranscription for PreparedCloudTranscription { + fn start( + &self, + partials: Arc, + ) -> BoxFuture<'static, Result, BackendError>> { + let preparation = self.preparation.clone(); + let task_spawner = Arc::clone(&self.task_spawner); + Box::pin(async move { + let (kind, asr_call_label) = + build_cloud_transcription_session(&preparation, task_spawner, partials).await?; + Ok(Arc::new(CloudTranscriptionSession { + kind, + asr_call_label, + finished: AtomicBool::new(false), + }) as Arc) + }) + } +} + #[derive(Clone)] enum CloudTranscriptionSessionKind { Volcengine(Arc), @@ -135,27 +197,32 @@ struct CloudTranscriptionSession { } impl TranscriptionEngine for SharedCloudTranscriptionEngine { + fn prepare( + self: Arc, + _session_id: SessionId, + context: Arc, + ) -> BoxFuture<'static, Result, BackendError>> { + let credentials = Arc::clone(&self.credentials); + Box::pin(async move { + let preparation = + prepare_cloud_transcription(credentials.as_ref(), Arc::clone(&context)).await?; + Ok(Arc::new(PreparedCloudTranscription { + preparation, + task_spawner: Arc::clone(&self.task_spawner), + }) as Arc) + }) + } + fn start( &self, - _session_id: SessionId, + session_id: SessionId, context: Arc, partials: Arc, ) -> BoxFuture<'static, Result, BackendError>> { - let credentials = Arc::clone(&self.credentials); - let task_spawner = Arc::clone(&self.task_spawner); + let engine = Arc::new(self.clone()); Box::pin(async move { - let (kind, asr_call_label) = build_cloud_transcription_session( - credentials.as_ref(), - &context, - task_spawner, - partials, - ) - .await?; - Ok(Arc::new(CloudTranscriptionSession { - kind, - asr_call_label, - finished: AtomicBool::new(false), - }) as Arc) + let prepared = engine.prepare(session_id, context).await?; + prepared.start(partials).await }) } } @@ -299,24 +366,19 @@ where } } -async fn build_cloud_transcription_session( +async fn prepare_cloud_transcription( credentials: &dyn CredentialStore, - context: &DictationContext, - task_spawner: Arc, - partials: Arc, -) -> Result<(CloudTranscriptionSessionKind, crate::AsrCallLabel), BackendError> { - use crate::asr::volcengine::VolcengineAuthMode; - use crate::provider_rules::{ActiveAsrProviderKind, BailianEndpointProtocol}; - - let channel_id = context.asr.provider_id.trim(); - let provider_type = context.asr.provider_type.trim(); + context: Arc, +) -> Result { + let channel_id = context.asr.provider_id.trim().to_string(); + let provider_type = context.asr.provider_type.trim().to_string(); if channel_id.is_empty() || provider_type.is_empty() { return Err(BackendError::new( BackendErrorCode::InvalidArgument, "ASR channel id and provider type must not be empty", )); } - if provider_descriptor(crate::ProviderKind::Asr, provider_type) + if provider_descriptor(crate::ProviderKind::Asr, &provider_type) .is_none_or(|descriptor| descriptor.validation_probe == ValidationProbe::Unsupported) { return Err(BackendError::new( @@ -324,52 +386,141 @@ async fn build_cloud_transcription_session( "shared cloud ASR provider is not supported", )); } - let stored_model = read_channel_credential( - credentials, - CredentialNamespace::Asr, - channel_id, - ASR_MODEL_ACCOUNT, - ) - .await?; + let mut values = HashMap::new(); + for account in ASR_SNAPSHOT_ACCOUNTS { + values.insert( + *account, + read_channel_credential(credentials, CredentialNamespace::Asr, &channel_id, account) + .await?, + ); + } + let value = |account: &str| values.get(account).cloned().flatten(); let model = context .asr .model .clone() - .or(stored_model) + .or_else(|| value(ASR_MODEL_ACCOUNT)) .unwrap_or_default(); - let effective = crate::provider_rules::resolve_effective_asr_provider(provider_type, &model) - .map_err(|message| BackendError::new(BackendErrorCode::InvalidArgument, message))?; - let api_key = read_channel_credential( - credentials, - CredentialNamespace::Asr, - channel_id, - ASR_API_KEY_ACCOUNT, - ) - .await? - .unwrap_or_default(); - let endpoint = read_channel_credential( - credentials, - CredentialNamespace::Asr, - channel_id, - ASR_ENDPOINT_ACCOUNT, - ) - .await? - .unwrap_or_default(); - let advanced_config_raw = read_channel_credential( - credentials, - CredentialNamespace::Asr, - channel_id, - ASR_ADVANCED_CONFIG_ACCOUNT, - ) - .await?; + let effective_provider = + crate::provider_rules::resolve_effective_asr_provider(&provider_type, &model) + .map_err(|message| BackendError::new(BackendErrorCode::InvalidArgument, message))?; let advanced_config = crate::provider_rules::advanced_asr_config_for( - provider_type, - advanced_config_raw.as_deref(), + &provider_type, + value(ASR_ADVANCED_CONFIG_ACCOUNT).as_deref(), ); + let preparation = CloudTranscriptionPreparation { + context, + provider_type, + kind: crate::provider_rules::active_asr_provider_kind(&effective_provider), + effective_provider, + api_key: value(ASR_API_KEY_ACCOUNT).unwrap_or_default(), + endpoint: value(ASR_ENDPOINT_ACCOUNT).unwrap_or_default(), + advanced_config, + model, + credentials: values, + }; + validate_cloud_transcription_preparation(&preparation)?; + Ok(preparation) +} - let (kind, label_model) = match crate::provider_rules::active_asr_provider_kind(&effective) { +fn validate_cloud_transcription_preparation( + preparation: &CloudTranscriptionPreparation, +) -> Result<(), BackendError> { + use crate::asr::volcengine::VolcengineAuthMode; + use crate::provider_rules::ActiveAsrProviderKind; + + match preparation.kind { + ActiveAsrProviderKind::Bailian + | ActiveAsrProviderKind::Qwen3Realtime + | ActiveAsrProviderKind::StepfunRealtime + | ActiveAsrProviderKind::Mimo + | ActiveAsrProviderKind::DashScopeMultimodal + | ActiveAsrProviderKind::ElevenLabs => { + require_configured(&preparation.api_key, "ASR API key")?; + } + ActiveAsrProviderKind::WhisperCompatible => { + if crate::provider_rules::api_key_required( + crate::ProviderKind::Asr, + &preparation.provider_type, + Some(&preparation.endpoint), + ) { + require_configured(&preparation.api_key, "ASR API key")?; + } + } + ActiveAsrProviderKind::Volcengine => { + let service = crate::asr::volcengine::VolcengineService::parse( + preparation + .value(crate::credentials::VOLCENGINE_SERVICE_ACCOUNT) + .as_deref() + .unwrap_or_default(), + ) + .map_err(|message| BackendError::new(BackendErrorCode::InvalidArgument, message))?; + let configured = preparation + .value(VOLCENGINE_AUTH_MODE_ACCOUNT) + .map(|value| VolcengineAuthMode::parse(&value)) + .unwrap_or(VolcengineAuthMode::AppIdToken); + let auth_mode = service.auth_mode(configured); + let app_id = preparation + .value(VOLCENGINE_APP_KEY_ACCOUNT) + .unwrap_or_default(); + let secret_account = match auth_mode { + VolcengineAuthMode::AppIdToken => VOLCENGINE_ACCESS_KEY_ACCOUNT, + VolcengineAuthMode::ApiKey => VOLCENGINE_API_KEY_ACCOUNT, + }; + let secret = preparation.value(secret_account).unwrap_or_default(); + if !auth_mode.auth_ok(&app_id, &secret) { + return Err(credential_missing("Volcengine credentials")); + } + } + ActiveAsrProviderKind::Xfyun => { + require_configured( + &preparation.value(XFYUN_APP_ID_ACCOUNT).unwrap_or_default(), + "Xfyun application id", + )?; + require_configured( + &preparation.value(XFYUN_API_KEY_ACCOUNT).unwrap_or_default(), + "Xfyun API key", + )?; + } + ActiveAsrProviderKind::TencentCloud => { + require_configured( + &preparation + .value(TENCENT_CLOUD_APP_ID_ACCOUNT) + .unwrap_or_default(), + "Tencent Cloud AppID", + )?; + require_configured( + &preparation + .value(TENCENT_CLOUD_SECRET_ID_ACCOUNT) + .unwrap_or_default(), + "Tencent Cloud SecretID", + )?; + require_configured( + &preparation + .value(TENCENT_CLOUD_SECRET_KEY_ACCOUNT) + .unwrap_or_default(), + "Tencent Cloud SecretKey", + )?; + } + } + Ok(()) +} + +async fn build_cloud_transcription_session( + preparation: &CloudTranscriptionPreparation, + task_spawner: Arc, + partials: Arc, +) -> Result<(CloudTranscriptionSessionKind, crate::AsrCallLabel), BackendError> { + use crate::asr::volcengine::VolcengineAuthMode; + use crate::provider_rules::{ActiveAsrProviderKind, BailianEndpointProtocol}; + + let provider_type = preparation.provider_type.as_str(); + let model = preparation.model.clone(); + let api_key = preparation.api_key.clone(); + let endpoint = preparation.endpoint.clone(); + let context = &preparation.context; + let (kind, label_model) = match preparation.kind { ActiveAsrProviderKind::Bailian => { - require_configured(&api_key, "ASR API key")?; let stored_endpoint = non_blank_owned(endpoint) .unwrap_or_else(|| crate::asr::bailian::DEFAULT_ENDPOINT.to_string()); let endpoint = if provider_type == crate::asr::bailian::PROVIDER_ID { @@ -383,14 +534,9 @@ async fn build_cloud_transcription_session( }; let effective_model = non_blank_owned(model) .unwrap_or_else(|| crate::asr::bailian::DEFAULT_MODEL.to_string()); - let vocabulary_id = read_channel_credential( - credentials, - CredentialNamespace::Asr, - channel_id, - ASR_VOCABULARY_ID_ACCOUNT, - ) - .await? - .and_then(non_blank_owned); + let vocabulary_id = preparation + .value(ASR_VOCABULARY_ID_ACCOUNT) + .and_then(non_blank_owned); let provider = Arc::new(BailianRealtimeASR::with_task_spawner( BailianCredentials { api_key, @@ -408,7 +554,6 @@ async fn build_cloud_transcription_session( ) } ActiveAsrProviderKind::Qwen3Realtime => { - require_configured(&api_key, "ASR API key")?; let stored_endpoint = non_blank_owned(endpoint) .unwrap_or_else(|| crate::asr::qwen_realtime::DEFAULT_ENDPOINT.to_string()); let endpoint = if provider_type == crate::asr::bailian::PROVIDER_ID { @@ -438,7 +583,6 @@ async fn build_cloud_transcription_session( ) } ActiveAsrProviderKind::StepfunRealtime => { - require_configured(&api_key, "ASR API key")?; let effective_model = non_blank_owned(model) .unwrap_or_else(|| crate::asr::stepfun_realtime::DEFAULT_MODEL.to_string()); let provider = Arc::new(StepfunRealtimeASR::with_task_spawner( @@ -458,7 +602,6 @@ async fn build_cloud_transcription_session( ) } ActiveAsrProviderKind::Mimo => { - require_configured(&api_key, "ASR API key")?; let effective_model = non_blank_owned(model) .unwrap_or_else(|| default_asr_model(provider_type).unwrap().to_string()); let endpoint = non_blank_owned(endpoint) @@ -474,7 +617,6 @@ async fn build_cloud_transcription_session( ) } ActiveAsrProviderKind::DashScopeMultimodal => { - require_configured(&api_key, "ASR API key")?; let model = non_blank_owned(model) .unwrap_or_else(|| crate::asr::dashscope_multimodal::DEFAULT_MODEL.to_string()); let stored_endpoint = non_blank_owned(endpoint) @@ -502,7 +644,6 @@ async fn build_cloud_transcription_session( ) } ActiveAsrProviderKind::ElevenLabs => { - require_configured(&api_key, "ASR API key")?; let effective_model = non_blank_owned(model) .unwrap_or_else(|| crate::asr::elevenlabs::DEFAULT_MODEL.to_string()); ( @@ -516,13 +657,6 @@ async fn build_cloud_transcription_session( ) } ActiveAsrProviderKind::WhisperCompatible => { - if crate::provider_rules::api_key_required( - crate::ProviderKind::Asr, - provider_type, - Some(&endpoint), - ) { - require_configured(&api_key, "ASR API key")?; - } let default_endpoint = default_asr_endpoint(provider_type).unwrap_or(""); let default_model = default_asr_model(provider_type).unwrap_or("whisper-1"); let effective_model = @@ -532,10 +666,13 @@ async fn build_cloud_transcription_session( non_blank_owned(endpoint).unwrap_or_else(|| default_endpoint.to_string()), effective_model.clone(), context.asr.prompt.clone(), - crate::provider_rules::batch_asr_chunk_limit_ms(provider_type, advanced_config), + crate::provider_rules::batch_asr_chunk_limit_ms( + provider_type, + preparation.advanced_config, + ), crate::provider_rules::whisper_supports_verbose_json( provider_type, - advanced_config, + preparation.advanced_config, ), ) .with_request_format(crate::provider_rules::whisper_request_format(provider_type)); @@ -552,7 +689,7 @@ async fn build_cloud_transcription_session( }); provider = provider .with_language(language) - .with_enable_itn(advanced_config.enable_itn); + .with_enable_itn(preparation.advanced_config.enable_itn); } ( CloudTranscriptionSessionKind::Whisper(Arc::new(provider)), @@ -560,57 +697,27 @@ async fn build_cloud_transcription_session( ) } ActiveAsrProviderKind::Volcengine => { - let service = read_channel_credential( - credentials, - CredentialNamespace::Asr, - channel_id, - crate::credentials::VOLCENGINE_SERVICE_ACCOUNT, - ) - .await?; let service = crate::asr::volcengine::VolcengineService::parse( - service.as_deref().unwrap_or_default(), + preparation + .value(crate::credentials::VOLCENGINE_SERVICE_ACCOUNT) + .as_deref() + .unwrap_or_default(), ) .map_err(|message| BackendError::new(BackendErrorCode::InvalidArgument, message))?; - let auth_mode = read_channel_credential( - credentials, - CredentialNamespace::Asr, - channel_id, - VOLCENGINE_AUTH_MODE_ACCOUNT, - ) - .await? - .map(|value| VolcengineAuthMode::parse(&value)) - .unwrap_or(VolcengineAuthMode::AppIdToken); + let auth_mode = preparation + .value(VOLCENGINE_AUTH_MODE_ACCOUNT) + .map(|value| VolcengineAuthMode::parse(&value)) + .unwrap_or(VolcengineAuthMode::AppIdToken); let auth_mode = service.auth_mode(auth_mode); - let app_id = read_channel_credential( - credentials, - CredentialNamespace::Asr, - channel_id, - VOLCENGINE_APP_KEY_ACCOUNT, - ) - .await? - .unwrap_or_default(); + let app_id = preparation + .value(VOLCENGINE_APP_KEY_ACCOUNT) + .unwrap_or_default(); let secret_account = match auth_mode { VolcengineAuthMode::AppIdToken => VOLCENGINE_ACCESS_KEY_ACCOUNT, VolcengineAuthMode::ApiKey => VOLCENGINE_API_KEY_ACCOUNT, }; - let access_token = read_channel_credential( - credentials, - CredentialNamespace::Asr, - channel_id, - secret_account, - ) - .await? - .unwrap_or_default(); - if !auth_mode.auth_ok(&app_id, &access_token) { - return Err(credential_missing("Volcengine credentials")); - } - let resource_id = read_channel_credential( - credentials, - CredentialNamespace::Asr, - channel_id, - VOLCENGINE_RESOURCE_ID_ACCOUNT, - ) - .await?; + let access_token = preparation.value(secret_account).unwrap_or_default(); + let resource_id = preparation.value(VOLCENGINE_RESOURCE_ID_ACCOUNT); let credentials = VolcengineCredentials { service, auth_mode, @@ -640,24 +747,8 @@ async fn build_cloud_transcription_session( (CloudTranscriptionSessionKind::Volcengine(provider), label) } ActiveAsrProviderKind::Xfyun => { - let app_id = read_channel_credential( - credentials, - CredentialNamespace::Asr, - channel_id, - XFYUN_APP_ID_ACCOUNT, - ) - .await? - .unwrap_or_default(); - let api_key = read_channel_credential( - credentials, - CredentialNamespace::Asr, - channel_id, - XFYUN_API_KEY_ACCOUNT, - ) - .await? - .unwrap_or_default(); - require_configured(&app_id, "Xfyun application id")?; - require_configured(&api_key, "Xfyun API key")?; + let app_id = preparation.value(XFYUN_APP_ID_ACCOUNT).unwrap_or_default(); + let api_key = preparation.value(XFYUN_API_KEY_ACCOUNT).unwrap_or_default(); let provider = Arc::new(XfyunStreamingASR::with_task_spawner( XfyunCredentials { app_id, api_key }, Arc::clone(&task_spawner), @@ -667,42 +758,17 @@ async fn build_cloud_transcription_session( (CloudTranscriptionSessionKind::Xfyun(provider), None) } ActiveAsrProviderKind::TencentCloud => { - let app_id = read_channel_credential( - credentials, - CredentialNamespace::Asr, - channel_id, - TENCENT_CLOUD_APP_ID_ACCOUNT, - ) - .await? - .unwrap_or_default(); - let secret_id = read_channel_credential( - credentials, - CredentialNamespace::Asr, - channel_id, - TENCENT_CLOUD_SECRET_ID_ACCOUNT, - ) - .await? - .unwrap_or_default(); - let secret_key = read_channel_credential( - credentials, - CredentialNamespace::Asr, - channel_id, - TENCENT_CLOUD_SECRET_KEY_ACCOUNT, - ) - .await? - .unwrap_or_default(); - require_configured(&app_id, "Tencent Cloud AppID")?; - require_configured(&secret_id, "Tencent Cloud SecretID")?; - require_configured(&secret_key, "Tencent Cloud SecretKey")?; - let model = read_channel_credential( - credentials, - CredentialNamespace::Asr, - channel_id, - ASR_MODEL_ACCOUNT, - ) - .await? - .filter(|value| !value.trim().is_empty()) - .unwrap_or_else(|| crate::asr::tencent_cloud::DEFAULT_MODEL.to_string()); + let app_id = preparation + .value(TENCENT_CLOUD_APP_ID_ACCOUNT) + .unwrap_or_default(); + let secret_id = preparation + .value(TENCENT_CLOUD_SECRET_ID_ACCOUNT) + .unwrap_or_default(); + let secret_key = preparation + .value(TENCENT_CLOUD_SECRET_KEY_ACCOUNT) + .unwrap_or_default(); + let model = non_blank_owned(preparation.model.clone()) + .unwrap_or_else(|| crate::asr::tencent_cloud::DEFAULT_MODEL.to_string()); let provider = Arc::new(TencentCloudStreamingASR::with_task_spawner( TencentCloudCredentials { app_id, @@ -723,7 +789,10 @@ async fn build_cloud_transcription_session( ) } }; - Ok((kind, crate::AsrCallLabel::new(effective, label_model))) + Ok(( + kind, + crate::AsrCallLabel::new(preparation.effective_provider.clone(), label_model), + )) } async fn read_channel_credential( @@ -2657,6 +2726,95 @@ mod tests { session.cancel().await.unwrap(); } + #[tokio::test] + async fn cloud_asr_preparation_freezes_channel_credentials_before_network_start() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let endpoint = format!("http://{}/v1", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.unwrap(); + let mut request = Vec::new(); + let mut chunk = [0_u8; 4096]; + let header_end; + loop { + let count = socket.read(&mut chunk).await.unwrap(); + assert!(count > 0, "client closed before sending its request"); + request.extend_from_slice(&chunk[..count]); + if let Some(end) = request.windows(4).position(|window| window == b"\r\n\r\n") { + header_end = end + 4; + break; + } + } + let headers = String::from_utf8_lossy(&request[..header_end]); + let content_length = headers + .lines() + .find_map(|line| line.strip_prefix("Content-Length: ")) + .and_then(|value| value.trim().parse::().ok()) + .unwrap_or(0); + while request.len() < header_end + content_length { + let count = socket.read(&mut chunk).await.unwrap(); + assert!(count > 0, "client closed before sending its body"); + request.extend_from_slice(&chunk[..count]); + } + let body = br#"{"choices":[{"message":{"content":"frozen"}}]}"#; + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + body.len() + ); + socket.write_all(response.as_bytes()).await.unwrap(); + socket.write_all(body).await.unwrap(); + request + }); + + let store = Arc::new(InMemoryCredentialStore::default()); + write_channel_secret( + store.as_ref(), + CredentialNamespace::Asr, + "frozen-channel", + ASR_API_KEY_ACCOUNT, + "frozen-key", + ) + .await; + write_channel_secret( + store.as_ref(), + CredentialNamespace::Asr, + "frozen-channel", + ASR_ENDPOINT_ACCOUNT, + &endpoint, + ) + .await; + let engine = Arc::new(SharedCloudTranscriptionEngine::new(store.clone())); + let context = Arc::new(DictationContext { + asr: ProviderInvocation::new("frozen-channel", "xiaomi-mimo-asr"), + ..DictationContext::default() + }); + let prepared = Arc::clone(&engine) + .prepare(SessionId::new(), context) + .await + .unwrap(); + write_channel_secret( + store.as_ref(), + CredentialNamespace::Asr, + "frozen-channel", + ASR_API_KEY_ACCOUNT, + "changed-key", + ) + .await; + + let session = prepared + .start(Arc::new(IgnoreTextStreamSink)) + .await + .unwrap(); + session.consume_pcm_chunk(&[0, 0]); + assert_eq!(session.finish().await.unwrap().text, "frozen"); + let request = String::from_utf8(server.await.unwrap()) + .unwrap() + .to_ascii_lowercase(); + assert!(request.contains("authorization: bearer frozen-key")); + assert!(!request.contains("changed-key")); + } + #[tokio::test] async fn cloud_polisher_reports_missing_credentials_without_opening_the_network() { let credentials: Arc = Arc::new(InMemoryCredentialStore::default()); diff --git a/openless-all/app/crates/openless-core/src/cloud_sync.rs b/openless-all/app/crates/openless-core/src/cloud_sync.rs index edde60f9b..b6d9eb264 100644 --- a/openless-all/app/crates/openless-core/src/cloud_sync.rs +++ b/openless-all/app/crates/openless-core/src/cloud_sync.rs @@ -533,6 +533,7 @@ fn capture_preferences( show_capsule: Some(preferences.show_capsule), audio_cue_on_record: Some(preferences.audio_cue_on_record), mute_during_recording: Some(preferences.mute_during_recording), + stable_transcription_enabled: Some(preferences.stable_transcription_enabled), silence_auto_stop_enabled: Some(preferences.silence_auto_stop_enabled), silence_auto_stop_seconds: Some(preferences.silence_auto_stop_seconds), show_overview_activity_heatmap: Some(preferences.show_overview_activity_heatmap), @@ -559,6 +560,7 @@ fn apply_preferences(preferences: &mut UserPreferences, incoming: &SyncPreferenc show_capsule, audio_cue_on_record, mute_during_recording, + stable_transcription_enabled, silence_auto_stop_enabled, silence_auto_stop_seconds, show_overview_activity_heatmap, diff --git a/openless-all/app/crates/openless-core/src/cloud_sync_types.rs b/openless-all/app/crates/openless-core/src/cloud_sync_types.rs index e06b0fc4b..55f7718df 100644 --- a/openless-all/app/crates/openless-core/src/cloud_sync_types.rs +++ b/openless-all/app/crates/openless-core/src/cloud_sync_types.rs @@ -143,6 +143,8 @@ pub struct SyncPreferences { #[serde(skip_serializing_if = "Option::is_none")] pub mute_during_recording: Option, #[serde(skip_serializing_if = "Option::is_none")] + pub stable_transcription_enabled: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub silence_auto_stop_enabled: Option, #[serde(skip_serializing_if = "Option::is_none")] pub silence_auto_stop_seconds: Option, diff --git a/openless-all/app/crates/openless-core/src/config.rs b/openless-all/app/crates/openless-core/src/config.rs index 6283f43dc..adf083ea8 100644 --- a/openless-all/app/crates/openless-core/src/config.rs +++ b/openless-all/app/crates/openless-core/src/config.rs @@ -119,6 +119,20 @@ impl TaskSpawner for TokioTaskSpawner { pub struct UnsupportedDictationEngine; impl DictationEngine for UnsupportedDictationEngine { + fn prepare_transcription( + self: Arc, + _session_id: crate::types::SessionId, + _context: Arc, + ) -> BoxFuture<'static, Result, BackendError>> + { + Box::pin(async { + Err(BackendError::new( + crate::errors::BackendErrorCode::Unsupported, + "dictation engine is not configured", + )) + }) + } + fn start( &self, _session_id: crate::types::SessionId, diff --git a/openless-all/app/crates/openless-core/src/dictation_context.rs b/openless-all/app/crates/openless-core/src/dictation_context.rs index 6ea37661e..19dc58be6 100644 --- a/openless-all/app/crates/openless-core/src/dictation_context.rs +++ b/openless-all/app/crates/openless-core/src/dictation_context.rs @@ -134,6 +134,8 @@ pub struct DictationInsertionContext { pub struct RecordingPlan { pub microphone_device_name: Option, pub mute_during_recording: bool, + /// Buffer the whole recording and start ASR only after the recorder stops. + pub transcribe_after_stop: bool, /// 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, @@ -247,6 +249,7 @@ impl DictationContext { recording: RecordingPlan { microphone_device_name: non_blank(&preferences.microphone_device_name), mute_during_recording: preferences.mute_during_recording, + transcribe_after_stop: preferences.stable_transcription_enabled, archive_enabled: true, archive_successful_recording: preferences.record_audio_for_debug, retention_days: preferences.history_retention_days, @@ -479,6 +482,7 @@ mod tests { active_asr_provider: "local-qwen3".to_string(), active_llm_provider: "openai".to_string(), local_asr_active_model: "qwen3-asr-1.7b".to_string(), + stable_transcription_enabled: true, history_max_entries: Some(100), audio_recording_max_entries: Some(7), working_languages: vec!["简体中文".to_string(), "English".to_string()], @@ -505,11 +509,13 @@ mod tests { preferences.microphone_device_name = "changed".to_string(); preferences.active_asr_provider = "changed".to_string(); + preferences.stable_transcription_enabled = false; assert_eq!( context.recording.microphone_device_name.as_deref(), Some("USB microphone") ); assert_eq!(context.asr.provider_id, "local-qwen3"); + assert!(context.recording.transcribe_after_stop); assert_eq!(context.asr.model.as_deref(), Some("qwen3-asr-1.7b")); // Audio archives have their own user-visible limit. History retention // may be much larger and must not silently override the recording cap. diff --git a/openless-all/app/crates/openless-core/src/dictation_engine.rs b/openless-all/app/crates/openless-core/src/dictation_engine.rs index 1df40a722..4e081370e 100644 --- a/openless-all/app/crates/openless-core/src/dictation_engine.rs +++ b/openless-all/app/crates/openless-core/src/dictation_engine.rs @@ -15,11 +15,18 @@ use crate::errors::{BackendError, BackendErrorCode}; use crate::ports::{ ActiveRecording, AudioCapture, AudioConsumer, AudioRecorder, CapturedPcm, DictationEngine, EngineFailure, EngineFailureStage, EngineProgress, EngineProgressSink, EngineResult, - EngineStage, RecordingProgressSink, TextPolisher, TextStreamChunk, TextStreamSink, - TranscriptionEngine, TranscriptionSession, VoiceCapture, + EngineStage, PreparedTranscription, RecordingProgressSink, TextPolisher, TextStreamChunk, + TextStreamSink, TranscriptionEngine, TranscriptionSession, VoiceCapture, }; use crate::types::{PolishDelta, SessionId, TranscriptDelta}; +// Keep one MiB for PCM callbacks that arrive while the host handles the stop request. +const MAX_BUFFERED_TRANSCRIPTION_PCM_BYTES: usize = 128 * 1024 * 1024; +const BUFFERED_TRANSCRIPTION_STOP_HEADROOM_BYTES: usize = 1024 * 1024; +const BUFFERED_TRANSCRIPTION_STOP_THRESHOLD_BYTES: usize = + MAX_BUFFERED_TRANSCRIPTION_PCM_BYTES - BUFFERED_TRANSCRIPTION_STOP_HEADROOM_BYTES; +const BUFFERED_TRANSCRIPTION_FORWARD_CHUNK_BYTES: usize = 64 * 1024; + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum PolishFailurePolicy { Fail, @@ -50,6 +57,8 @@ struct PipelineSession { struct PipelineResources { recording: Option>, transcription: Option>, + buffered: Option>, + prepared: Option>, } impl PipelineSession { @@ -134,57 +143,77 @@ impl DictationEngine for PipelineDictationEngine { } } - let transcript_partials: Arc = - Arc::new(TranscriptProgressForwarder { - session_id, - progress: Arc::clone(&progress), - }); - let transcription = match transcription_engine - .start(session_id, Arc::clone(&context), transcript_partials) + let prepared = match transcription_engine + .prepare(session_id, Arc::clone(&context)) .await { - Ok(transcription) => transcription, + Ok(prepared) => prepared, Err(error) => { remove_session(&sessions, session_id, &session); - return Err(error); + return Err(if session.cancelled.load(Ordering::Acquire) { + cancelled_error("dictation cancelled while preparing the provider") + } else { + error + }); } }; - - { + if session.cancelled.load(Ordering::Acquire) { + remove_session(&sessions, session_id, &session); + return Err(cancelled_error( + "dictation cancelled while preparing the provider", + )); + } + let stable = context.recording.transcribe_after_stop; + let transcript_partials: Arc = if stable { + Arc::new(DiscardTextStream) + } else { + Arc::new(TranscriptProgressForwarder { + session_id, + progress: Arc::clone(&progress), + }) + }; + let recording_progress: Arc = + Arc::new(RecordingProgressForwarder { + session_id, + session: Arc::downgrade(&session), + progress: Arc::clone(&progress), + }); + let buffered = Arc::new(BufferedTranscriptionSession::new( + prepared, + transcript_partials, + Arc::clone(&recording_progress), + )); + let registered = { let mut resources = session .resources .lock() .expect("pipeline resource lock poisoned"); - if !session.cancelled.load(Ordering::Acquire) { - resources.transcription = Some(Arc::clone(&transcription)); + if session.cancelled.load(Ordering::Acquire) { + false + } else { + let transcription: Arc = buffered.clone(); + resources.transcription = Some(transcription); + resources.buffered = Some(Arc::clone(&buffered)); + resources.prepared = Some(buffered.prepared()); + true } - } - if session.cancelled.load(Ordering::Acquire) { - let cancel_result = - cancel_transcription_once(&session, Arc::clone(&transcription)).await; + }; + if !registered { + let _ = buffered.cancel().await; remove_session(&sessions, session_id, &session); - cancel_result?; return Err(cancelled_error( - "dictation cancelled while ASR was starting", + "dictation cancelled before the recorder started", )); } - let audio_consumer: Arc = Arc::new(SessionAudioConsumer { - session: Arc::clone(&transcription), - }); - let recording_progress: Arc = - Arc::new(RecordingProgressForwarder { - session_id, - session: Arc::clone(&session), - progress, - }); + let audio_consumer: Arc = buffered.clone(); let recording = match recorder .start(session_id, context, audio_consumer, recording_progress) .await { Ok(recording) => recording, Err(error) => { - let _ = cancel_transcription_once(&session, transcription).await; + let _ = cancel_transcription_once(&session, Arc::clone(&buffered)).await; remove_session(&sessions, session_id, &session); return Err(error); } @@ -202,7 +231,8 @@ impl DictationEngine for PipelineDictationEngine { } if let Some(recording) = recording { let stop_result = recording.stop().await; - let cancel_result = cancel_transcription_once(&session, transcription).await; + let cancel_result = + cancel_transcription_once(&session, Arc::clone(&buffered)).await; remove_session(&sessions, session_id, &session); stop_result?; cancel_result?; @@ -210,6 +240,36 @@ impl DictationEngine for PipelineDictationEngine { "dictation cancelled while the recorder was starting", )); } + if !stable { + if let Err(error) = buffered.attach().await { + let recording = session + .resources + .lock() + .expect("pipeline resource lock poisoned") + .recording + .take(); + if let Some(recording) = recording { + let _ = recording.stop().await; + } + let _ = cancel_transcription_once(&session, Arc::clone(&buffered)).await; + remove_session(&sessions, session_id, &session); + return Err(error); + } + } + if session.cancelled.load(Ordering::Acquire) { + let recording = session + .resources + .lock() + .expect("pipeline resource lock poisoned") + .recording + .take(); + if let Some(recording) = recording { + let _ = recording.stop().await; + } + let _ = cancel_transcription_once(&session, Arc::clone(&buffered)).await; + remove_session(&sessions, session_id, &session); + return Err(cancelled_error("dictation cancelled while starting")); + } Ok(()) }) } @@ -223,6 +283,14 @@ impl DictationEngine for PipelineDictationEngine { self.transcription.start(session_id, context, partials) } + fn prepare_transcription( + self: Arc, + session_id: SessionId, + context: Arc, + ) -> BoxFuture<'static, Result, BackendError>> { + Arc::clone(&self.transcription).prepare(session_id, context) + } + fn start_voice_capture( &self, session_id: SessionId, @@ -236,41 +304,81 @@ impl DictationEngine for PipelineDictationEngine { Box::pin(async move { if cancel.is_cancelled() { return Err(cancelled_error( - "voice capture cancelled before ASR startup", + "voice capture cancelled before recorder startup", )); } - let transcription = transcription_engine - .start(session_id, Arc::clone(&context), partials) + let prepared = transcription_engine + .prepare(session_id, Arc::clone(&context)) .await?; if cancel.is_cancelled() { - let _ = transcription.cancel().await; return Err(cancelled_error( - "voice capture cancelled while ASR was starting", + "voice capture cancelled while preparing the provider", )); } - let consumer: Arc = Arc::new(SessionAudioConsumer { - session: Arc::clone(&transcription), - }); - match recorder + let stable = context.recording.transcribe_after_stop; + let partials = if stable { + Arc::new(DiscardTextStream) as Arc + } else { + partials + }; + let buffered = Arc::new(BufferedTranscriptionSession::new( + prepared, + partials, + Arc::clone(&progress), + )); + let consumer: Arc = buffered.clone(); + let recording = recorder .start(session_id, context, consumer, progress) - .await - { - Ok(recording) if cancel.is_cancelled() => { - let _ = recording.stop().await; - let _ = transcription.cancel().await; - Err(cancelled_error( - "voice capture cancelled while recorder was starting", - )) - } - Ok(recording) => Ok(VoiceCapture { - recording, - transcription, - }), - Err(error) => { - let _ = transcription.cancel().await; - Err(error) + .await?; + let transcription: Arc = buffered.clone(); + if cancel.is_cancelled() { + let (stop_result, cancel_result) = + futures_util::future::join(recording.stop(), transcription.cancel()).await; + stop_result?; + cancel_result?; + return Err(cancelled_error( + "voice capture cancelled while recorder was starting", + )); + } + if !stable { + let attaching = buffered.attach(); + tokio::pin!(attaching); + tokio::select! { + result = &mut attaching => { + if let Err(error) = result { + let (stop_result, _) = futures_util::future::join( + recording.stop(), + transcription.cancel(), + ).await; + let _ = stop_result; + return Err(error); + } + } + _ = cancel.cancelled() => { + let (stop_result, cancel_result, _) = tokio::join!( + recording.stop(), + transcription.cancel(), + &mut attaching, + ); + stop_result?; + cancel_result?; + return Err(cancelled_error( + "voice capture cancelled while ASR was starting", + )); + } } } + if cancel.is_cancelled() { + let (stop_result, cancel_result) = + futures_util::future::join(recording.stop(), transcription.cancel()).await; + stop_result?; + cancel_result?; + return Err(cancelled_error("voice capture cancelled while starting")); + } + Ok(VoiceCapture { + recording, + transcription, + }) }) } @@ -309,7 +417,6 @@ impl DictationEngine for PipelineDictationEngine { progress: Arc, ) -> BoxFuture<'static, Result> { let sessions = Arc::clone(&self.sessions); - let transcription_engine = Arc::clone(&self.transcription); let polisher = Arc::clone(&self.polisher); let policy = self.polish_failure_policy; Box::pin(async move { @@ -327,12 +434,16 @@ impl DictationEngine for PipelineDictationEngine { } let context = session.context(); - let (recording, transcription) = { + let (recording, transcription, prepared) = { let mut resources = session .resources .lock() .expect("pipeline resource lock poisoned"); - (resources.recording.take(), resources.transcription.clone()) + ( + resources.recording.take(), + resources.transcription.clone(), + resources.prepared.clone(), + ) }; let recording = recording.ok_or_else(|| { BackendError::new( @@ -374,6 +485,7 @@ impl DictationEngine for PipelineDictationEngine { let asr_started = std::time::Instant::now(); let mut asr_call_label = transcription.asr_call_label(); let transcription_result = transcription.finish().await; + asr_call_label = transcription.asr_call_label().or(asr_call_label); for notification in transcription.take_progress_notifications() { publish_progress( &session, @@ -391,6 +503,12 @@ impl DictationEngine for PipelineDictationEngine { } Err(first_error) => { let cancelled = session.cancelled.load(Ordering::Acquire); + let prepared = prepared.ok_or_else(|| { + BackendError::new( + BackendErrorCode::Internal, + "transcription provider preparation is unavailable", + ) + })?; let retry_pcm = if !cancelled && first_error.retryable { match archive.as_ref().filter(|archive| archive.is_available()) { Some(archive) => { @@ -404,10 +522,9 @@ impl DictationEngine for PipelineDictationEngine { let _ = cancel_transcription_once(&session, Arc::clone(&transcription)).await; match retry_pcm { Some(pcm) => match retry_transcription( - transcription_engine, + prepared, Arc::clone(&session), session_id, - Arc::clone(&context), Arc::clone(&progress), pcm, ) @@ -633,18 +750,27 @@ impl DictationEngine for PipelineDictationEngine { return Ok(()); } - let (recording, transcription) = { + let (recording, transcription, buffered) = { let mut resources = session .resources .lock() .expect("pipeline resource lock poisoned"); - (resources.recording.take(), resources.transcription.clone()) + ( + resources.recording.take(), + resources.transcription.clone(), + resources.buffered.take(), + ) }; let mut first_error = None; if let Some(recording) = recording { retain_first_error(&mut first_error, recording.stop().await); } - if let Some(transcription) = transcription { + if let Some(buffered) = buffered { + retain_first_error( + &mut first_error, + buffered.cancel_without_waiting_for_start().await, + ); + } else if let Some(transcription) = transcription { retain_first_error( &mut first_error, cancel_transcription_once(&session, transcription).await, @@ -697,10 +823,9 @@ fn remove_session( } async fn retry_transcription( - engine: Arc, + prepared: Arc, session: Arc, session_id: SessionId, - context: Arc, progress: Arc, pcm: Vec, ) -> Result< @@ -721,10 +846,7 @@ async fn retry_transcription( session_id, progress: Arc::clone(&progress), }); - let transcription = match engine - .start(session_id, Arc::clone(&context), partials) - .await - { + let transcription = match prepared.start(partials).await { Ok(transcription) => transcription, Err(_) if session.cancelled.load(Ordering::Acquire) => { return Err(( @@ -755,6 +877,7 @@ async fn retry_transcription( .transcription_finished .store(false, Ordering::Release); resources.transcription = Some(Arc::clone(&transcription)); + resources.buffered = None; true } }; @@ -823,10 +946,13 @@ async fn cancellable_backoff( } } -async fn cancel_transcription_once( +async fn cancel_transcription_once( session: &Arc, - transcription: Arc, -) -> Result<(), BackendError> { + transcription: Arc, +) -> Result<(), BackendError> +where + T: TranscriptionSession + ?Sized, +{ if session.transcription_finished.load(Ordering::Acquire) || session.transcription_cancelled.swap(true, Ordering::AcqRel) { @@ -880,19 +1006,386 @@ fn retain_first_error(first_error: &mut Option, result: Result<(), } } -struct SessionAudioConsumer { - session: Arc, +pub(crate) fn buffered_transcription_session( + prepared: Arc, + context: Arc, + partials: Arc, + progress: Arc, +) -> Arc { + let partials = if context.recording.transcribe_after_stop { + Arc::new(DiscardTextStream) as Arc + } else { + partials + }; + let buffered = Arc::new(BufferedTranscriptionSession::new( + prepared, partials, progress, + )); + if !context.recording.transcribe_after_stop { + buffered.attach_in_background(); + } + buffered +} + +struct BufferedTranscriptionSession { + inner: Arc, +} + +struct BufferedTranscriptionInner { + prepared: Arc, + partials: Arc, + progress: Arc, + limit_notified: AtomicBool, + limit_threshold_bytes: usize, + state: Mutex, +} + +enum BufferedTranscriptionState { + Buffering(Vec), + Attaching { + pcm: Vec, + waiter: Arc, + }, + Direct(Arc), + Failed(BackendError), + Cancelled, +} + +impl BufferedTranscriptionSession { + fn new( + prepared: Arc, + partials: Arc, + progress: Arc, + ) -> Self { + Self::new_with_limit( + prepared, + partials, + progress, + BUFFERED_TRANSCRIPTION_STOP_THRESHOLD_BYTES, + ) + } + + fn new_with_limit( + prepared: Arc, + partials: Arc, + progress: Arc, + limit_threshold_bytes: usize, + ) -> Self { + Self { + inner: Arc::new(BufferedTranscriptionInner { + prepared, + partials, + progress, + limit_notified: AtomicBool::new(false), + limit_threshold_bytes, + state: Mutex::new(BufferedTranscriptionState::Buffering(Vec::new())), + }), + } + } + + fn attach(&self) -> BoxFuture<'static, Result, BackendError>> { + let waiter = Arc::new(tokio::sync::Notify::new()); + let start = { + let mut state = self + .inner + .state + .lock() + .expect("buffered transcription lock poisoned"); + match &mut *state { + BufferedTranscriptionState::Buffering(pcm) => { + let pcm = std::mem::take(pcm); + *state = BufferedTranscriptionState::Attaching { + pcm: Vec::new(), + waiter: Arc::clone(&waiter), + }; + Some(pcm) + } + BufferedTranscriptionState::Direct(session) => { + let session = Arc::clone(session); + return Box::pin(async move { Ok(session) }); + } + BufferedTranscriptionState::Failed(error) => { + let error = error.clone(); + return Box::pin(async move { Err(error) }); + } + BufferedTranscriptionState::Cancelled => { + return Box::pin(async { + Err(cancelled_error("transcription buffer was cancelled")) + }); + } + BufferedTranscriptionState::Attaching { .. } => { + let waiter = match &*state { + BufferedTranscriptionState::Attaching { waiter, .. } => Arc::clone(waiter), + _ => unreachable!("buffer state changed while it was locked"), + }; + let inner = Arc::clone(&self.inner); + return Box::pin(async move { + waiter.notified().await; + match &*inner + .state + .lock() + .expect("buffered transcription lock poisoned") + { + BufferedTranscriptionState::Direct(session) => Ok(Arc::clone(session)), + BufferedTranscriptionState::Failed(error) => Err(error.clone()), + BufferedTranscriptionState::Cancelled => { + Err(cancelled_error("transcription buffer was cancelled")) + } + BufferedTranscriptionState::Buffering(_) + | BufferedTranscriptionState::Attaching { .. } => { + Err(BackendError::new( + BackendErrorCode::Internal, + "transcription attachment did not settle", + )) + } + } + }); + } + } + }; + let inner = Arc::clone(&self.inner); + Box::pin(async move { + attach_buffered_transcription(inner, start.expect("buffer attach must start"), waiter) + .await + }) + } + + fn attach_in_background(&self) { + let attaching = self.attach(); + tokio::spawn(async move { + if let Err(error) = attaching.await { + log::warn!("provider-only transcription startup failed: {error}"); + } + }); + } + + fn prepared(&self) -> Arc { + Arc::clone(&self.inner.prepared) + } + + fn downstream(&self) -> Option> { + match &*self + .inner + .state + .lock() + .expect("buffered transcription lock poisoned") + { + BufferedTranscriptionState::Direct(session) => Some(Arc::clone(session)), + _ => None, + } + } +} + +fn attach_buffered_transcription( + inner: Arc, + pcm: Vec, + waiter: Arc, +) -> BoxFuture<'static, Result, BackendError>> { + Box::pin(async move { + let _notify = NotifyOnDrop { waiter }; + let downstream = match inner.prepared.start(Arc::clone(&inner.partials)).await { + Ok(session) => session, + Err(error) => { + let mut state = inner + .state + .lock() + .expect("buffered transcription lock poisoned"); + if matches!(*state, BufferedTranscriptionState::Cancelled) { + return Err(cancelled_error( + "transcription buffer was cancelled while ASR was starting", + )); + } + *state = BufferedTranscriptionState::Failed(error.clone()); + return Err(error); + } + }; + + let mut first_chunk = Some(pcm); + + loop { + let chunk = if let Some(chunk) = first_chunk.take().filter(|chunk| !chunk.is_empty()) { + if matches!( + &*inner + .state + .lock() + .expect("buffered transcription lock poisoned"), + BufferedTranscriptionState::Cancelled + ) { + Err(cancelled_error( + "transcription buffer was cancelled while ASR was starting", + )) + } else { + Ok(chunk) + } + } else { + let mut state = inner + .state + .lock() + .expect("buffered transcription lock poisoned"); + match &mut *state { + BufferedTranscriptionState::Attaching { pcm, .. } if pcm.is_empty() => { + *state = BufferedTranscriptionState::Direct(Arc::clone(&downstream)); + return Ok(downstream); + } + BufferedTranscriptionState::Attaching { pcm, .. } => Ok(std::mem::take(pcm)), + BufferedTranscriptionState::Cancelled => Err(cancelled_error( + "transcription buffer was cancelled while ASR was starting", + )), + BufferedTranscriptionState::Failed(error) => Err(error.clone()), + BufferedTranscriptionState::Direct(session) => return Ok(Arc::clone(session)), + BufferedTranscriptionState::Buffering(_) => unreachable!( + "buffered transcription cannot return to buffering while attaching" + ), + } + }; + match chunk { + Ok(chunk) => { + for chunk in chunk.chunks(BUFFERED_TRANSCRIPTION_FORWARD_CHUNK_BYTES) { + downstream.consume_pcm_chunk(chunk); + } + } + Err(error) => { + let _ = downstream.cancel().await; + return Err(error); + } + } + } + }) +} + +struct NotifyOnDrop { + waiter: Arc, } -impl AudioConsumer for SessionAudioConsumer { +impl Drop for NotifyOnDrop { + fn drop(&mut self) { + self.waiter.notify_waiters(); + self.waiter.notify_one(); + } +} + +impl AudioConsumer for BufferedTranscriptionSession { fn consume_pcm_chunk(&self, pcm: &[u8]) { - self.session.consume_pcm_chunk(pcm); + let (downstream, buffer_limit_reached) = { + let mut state = self + .inner + .state + .lock() + .expect("buffered transcription lock poisoned"); + match &mut *state { + BufferedTranscriptionState::Buffering(buffer) + | BufferedTranscriptionState::Attaching { pcm: buffer, .. } => { + buffer.extend_from_slice(pcm); + (None, buffer.len() >= self.inner.limit_threshold_bytes) + } + BufferedTranscriptionState::Direct(session) => (Some(Arc::clone(session)), false), + BufferedTranscriptionState::Failed(_) | BufferedTranscriptionState::Cancelled => { + (None, false) + } + } + }; + if let Some(downstream) = downstream { + downstream.consume_pcm_chunk(pcm); + } + if buffer_limit_reached + && self + .inner + .limit_notified + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .is_ok() + { + // ponytail: one stop request only; if the host cannot honor it, PCM + // may exceed 128 MiB until the caller stops. Never truncate the tail; + // add a hard cap only with a product-level overflow policy. + if let Err(error) = self + .inner + .progress + .publish(crate::ports::RecordingEvent::LimitReached) + { + log::warn!("failed to request recording stop at PCM limit: {error}"); + } + } + } +} + +impl TranscriptionSession for BufferedTranscriptionSession { + fn asr_call_label(&self) -> Option { + self.downstream() + .and_then(|session| session.asr_call_label()) + } + + fn take_progress_notifications(&self) -> Vec { + self.downstream() + .map_or_else(Vec::new, |session| session.take_progress_notifications()) + } + + fn finish(&self) -> BoxFuture<'static, Result> { + let attaching = self.attach(); + Box::pin(async move { + let downstream = attaching.await?; + downstream.finish().await + }) + } + + fn cancel(&self) -> BoxFuture<'static, Result<(), BackendError>> { + let (downstream, waiter) = self.take_cancel_state(); + Box::pin(async move { + if let Some(waiter) = waiter { + waiter.notified().await; + } + match downstream { + Some(session) => session.cancel().await, + None => Ok(()), + } + }) + } +} + +impl BufferedTranscriptionSession { + fn take_cancel_state( + &self, + ) -> ( + Option>, + Option>, + ) { + let mut state = self + .inner + .state + .lock() + .expect("buffered transcription lock poisoned"); + let waiter = match &*state { + BufferedTranscriptionState::Attaching { waiter, .. } => Some(Arc::clone(waiter)), + _ => None, + }; + let downstream = match std::mem::replace(&mut *state, BufferedTranscriptionState::Cancelled) + { + BufferedTranscriptionState::Direct(session) => Some(session), + _ => None, + }; + (downstream, waiter) + } + + fn cancel_without_waiting_for_start(&self) -> BoxFuture<'static, Result<(), BackendError>> { + let (downstream, _) = self.take_cancel_state(); + Box::pin(async move { + match downstream { + Some(session) => session.cancel().await, + None => Ok(()), + } + }) + } +} + +struct DiscardTextStream; + +impl TextStreamSink for DiscardTextStream { + fn publish(&self, _chunk: TextStreamChunk) -> Result<(), BackendError> { + Ok(()) } } struct RecordingProgressForwarder { session_id: SessionId, - session: Arc, + session: std::sync::Weak, progress: Arc, } @@ -912,12 +1405,16 @@ impl RecordingProgressSink for RecordingProgressForwarder { crate::ports::RecordingEvent::Level { elapsed_ms, level } => { self.publish_level(elapsed_ms, level) } + crate::ports::RecordingEvent::LimitReached => self + .progress + .publish(self.session_id, EngineProgress::RecordingLimitReached), crate::ports::RecordingEvent::Fatal(error) => { - *self - .session - .recording_fault - .lock() - .expect("recording fault lock poisoned") = Some(error.clone()); + if let Some(session) = self.session.upgrade() { + *session + .recording_fault + .lock() + .expect("recording fault lock poisoned") = Some(error.clone()); + } self.progress .publish(self.session_id, EngineProgress::RecordingFault(error)) } @@ -985,6 +1482,46 @@ mod tests { } } + struct NoopRecordingProgress; + + impl RecordingProgressSink for NoopRecordingProgress { + fn publish_level(&self, _elapsed_ms: u64, _level: f32) -> Result<(), BackendError> { + Ok(()) + } + } + + #[derive(Default)] + struct LimitRecordingProgress { + limits: AtomicUsize, + } + + impl RecordingProgressSink for LimitRecordingProgress { + fn publish_level(&self, _elapsed_ms: u64, _level: f32) -> Result<(), BackendError> { + Ok(()) + } + + fn publish(&self, event: crate::ports::RecordingEvent) -> Result<(), BackendError> { + if matches!(event, crate::ports::RecordingEvent::LimitReached) { + self.limits.fetch_add(1, Ordering::SeqCst); + } + Ok(()) + } + } + + #[test] + fn recording_progress_forwarder_does_not_keep_pipeline_session_alive() { + let session = Arc::new(PipelineSession::new(raw_dictation_context())); + let weak = Arc::downgrade(&session); + let _forwarder = RecordingProgressForwarder { + session: Arc::downgrade(&session), + session_id: SessionId::new(), + progress: Arc::new(RecordingProgress::default()), + }; + + drop(session); + assert!(weak.upgrade().is_none()); + } + struct FixtureRecording { stops: Arc, archive: Arc, @@ -1055,6 +1592,35 @@ mod tests { } } + struct ExposedRecorder { + consumer: Arc>>>, + stops: Arc, + } + + impl AudioRecorder for ExposedRecorder { + fn start( + &self, + _session_id: SessionId, + _context: Arc, + consumer: Arc, + _progress: Arc, + ) -> BoxFuture<'static, Result, BackendError>> { + consumer.consume_pcm_chunk(&[1, 0]); + *self.consumer.lock().unwrap() = Some(consumer); + let stops = Arc::clone(&self.stops); + Box::pin(async move { + Ok(Box::new(FixtureRecording { + stops, + archive: Arc::new(FixtureArchive { + available: AtomicBool::new(false), + discards: Arc::new(AtomicUsize::new(0)), + pcm: Vec::new(), + }), + }) as Box) + }) + } + } + struct FixtureTranscriptionSession { pcm: Arc>>, cancels: Arc, @@ -1094,6 +1660,7 @@ mod tests { struct FixtureTranscriber { session: Arc, + starts: Arc, } impl TranscriptionEngine for FixtureTranscriber { @@ -1103,6 +1670,7 @@ mod tests { _context: Arc, partials: Arc, ) -> BoxFuture<'static, Result, BackendError>> { + self.starts.fetch_add(1, Ordering::AcqRel); let session = Arc::clone(&self.session); Box::pin(async move { partials.publish(TextStreamChunk { @@ -1114,6 +1682,75 @@ mod tests { } } + struct DelayedTranscriber { + entered: Arc, + release: Arc, + session: Arc, + } + + impl TranscriptionEngine for DelayedTranscriber { + fn start( + &self, + _session_id: SessionId, + _context: Arc, + _partials: Arc, + ) -> BoxFuture<'static, Result, BackendError>> { + let entered = Arc::clone(&self.entered); + let release = Arc::clone(&self.release); + let session = Arc::clone(&self.session); + Box::pin(async move { + entered.notify_one(); + release.notified().await; + Ok(session as Arc) + }) + } + } + + struct FailingTranscriber; + + impl TranscriptionEngine for FailingTranscriber { + fn start( + &self, + _session_id: SessionId, + _context: Arc, + _partials: Arc, + ) -> BoxFuture<'static, Result, BackendError>> { + Box::pin(async { + Err(BackendError::new( + BackendErrorCode::Provider, + "fixture ASR start failed", + )) + }) + } + } + + struct FailingPreparationTranscriber; + + impl TranscriptionEngine for FailingPreparationTranscriber { + fn prepare( + self: Arc, + _session_id: SessionId, + _context: Arc, + ) -> BoxFuture<'static, Result, BackendError>> + { + Box::pin(async { + Err(BackendError::new( + BackendErrorCode::Provider, + "fixture ASR preparation failed", + )) + }) + } + + fn start( + &self, + _session_id: SessionId, + _context: Arc, + _partials: Arc, + ) -> BoxFuture<'static, Result, BackendError>> { + Box::pin(async { unreachable!("preparation failure must precede provider start") }) + } + } + struct RetryTranscriber { outputs: Arc>>>, starts: Arc, @@ -1197,6 +1834,15 @@ mod tests { } } + fn noop_polisher() -> Arc { + Arc::new(FixturePolisher { + result: Ok(crate::ports::PolishOutput::text("unused")), + calls: Arc::new(AtomicUsize::new(0)), + cancels: Arc::new(AtomicUsize::new(0)), + contexts: Arc::new(Mutex::new(Vec::new())), + }) + } + struct FixtureParts { engine: PipelineDictationEngine, progress: Arc, @@ -1204,6 +1850,7 @@ mod tests { recorder_stops: Arc, archive_discards: Arc, transcription_cancels: Arc, + transcription_starts: Arc, polish_calls: Arc, polish_contexts: Arc>>>, } @@ -1223,6 +1870,7 @@ mod tests { pcm: vec![1, 0, 2, 0], }); let transcription_cancels = Arc::new(AtomicUsize::new(0)); + let transcription_starts = Arc::new(AtomicUsize::new(0)); let polish_calls = Arc::new(AtomicUsize::new(0)); let polish_contexts = Arc::new(Mutex::new(Vec::new())); let transcriber = Arc::new(FixtureTranscriber { @@ -1232,6 +1880,7 @@ mod tests { finish_entered, finish_release, }), + starts: Arc::clone(&transcription_starts), }); let engine = PipelineDictationEngine::new( Arc::new(FixtureRecorder { @@ -1254,6 +1903,7 @@ mod tests { recorder_stops, archive_discards, transcription_cancels, + transcription_starts, polish_calls, polish_contexts, } @@ -1306,6 +1956,147 @@ mod tests { }))); } + #[tokio::test] + async fn stable_pipeline_starts_asr_only_after_stop_and_emits_only_final_transcript() { + let fixture = fixture_engine( + false, + Ok(crate::ports::PolishOutput::text("unused")), + None, + None, + ); + let session_id = SessionId::new(); + let mut context = (*raw_dictation_context()).clone(); + context.recording.transcribe_after_stop = true; + + fixture + .engine + .start(session_id, Arc::new(context), fixture.progress.clone()) + .await + .unwrap(); + + assert_eq!(fixture.transcription_starts.load(Ordering::Acquire), 0); + assert!(fixture.pcm.lock().unwrap().is_empty()); + + let result = fixture + .engine + .finish(session_id, fixture.progress.clone()) + .await + .unwrap(); + + assert_eq!(result.raw_text, "raw text"); + assert_eq!(fixture.transcription_starts.load(Ordering::Acquire), 1); + assert_eq!(&*fixture.pcm.lock().unwrap(), &[1, 0, 2, 0]); + let transcript_events: Vec<_> = fixture + .progress + .events + .lock() + .unwrap() + .iter() + .filter_map(|event| match event { + EngineProgress::TranscriptDelta(delta) => Some(delta.clone()), + _ => None, + }) + .collect(); + assert_eq!( + transcript_events, + vec![TranscriptDelta { + text: "raw text".into(), + offset: 0, + is_final: true, + }] + ); + } + + #[tokio::test] + async fn stable_pipeline_cancel_discards_pcm_without_starting_asr() { + 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.transcribe_after_stop = true; + fixture + .engine + .start(session_id, Arc::new(context), fixture.progress.clone()) + .await + .unwrap(); + + fixture.engine.cancel(session_id).await.unwrap(); + + assert_eq!(fixture.transcription_starts.load(Ordering::Acquire), 0); + assert_eq!(fixture.transcription_cancels.load(Ordering::Acquire), 0); + assert_eq!(fixture.recorder_stops.load(Ordering::Acquire), 1); + } + + #[tokio::test] + async fn stable_voice_capture_defers_asr_until_the_shared_finish_handle() { + let fixture = fixture_engine( + false, + Ok(crate::ports::PolishOutput::text("unused")), + None, + None, + ); + let mut context = DictationContext::default(); + context.recording.transcribe_after_stop = true; + let capture = fixture + .engine + .start_voice_capture( + SessionId::new(), + Arc::new(context), + Arc::new(DiscardTextStream), + Arc::new(NoopRecordingProgress), + crate::CancellationToken::new(), + ) + .await + .unwrap(); + + assert_eq!(fixture.transcription_starts.load(Ordering::Acquire), 0); + capture.recording.stop().await.unwrap(); + let output = capture.transcription.finish().await.unwrap(); + + assert_eq!(output.text, "raw text"); + assert_eq!(fixture.transcription_starts.load(Ordering::Acquire), 1); + assert_eq!(&*fixture.pcm.lock().unwrap(), &[1, 0, 2, 0]); + } + + #[tokio::test] + async fn buffered_pcm_limit_requests_one_stop_and_preserves_the_tail() { + let pcm = Arc::new(Mutex::new(Vec::new())); + let starts = Arc::new(AtomicUsize::new(0)); + let progress = Arc::new(LimitRecordingProgress::default()); + let transcriber = Arc::new(FixtureTranscriber { + session: Arc::new(FixtureTranscriptionSession { + pcm: Arc::clone(&pcm), + cancels: Arc::new(AtomicUsize::new(0)), + finish_entered: None, + finish_release: None, + }), + starts: Arc::clone(&starts), + }); + let prepared = transcriber + .prepare(SessionId::new(), raw_dictation_context()) + .await + .unwrap(); + let buffered = BufferedTranscriptionSession::new_with_limit( + prepared, + Arc::new(DiscardTextStream), + progress.clone(), + 4, + ); + + buffered.consume_pcm_chunk(&[1, 0]); + buffered.consume_pcm_chunk(&[2, 0, 3, 0]); + buffered.consume_pcm_chunk(&[4, 0]); + + assert_eq!(progress.limits.load(Ordering::SeqCst), 1); + buffered.attach().await.unwrap().finish().await.unwrap(); + assert_eq!(&*pcm.lock().unwrap(), &[1, 0, 2, 0, 3, 0, 4, 0]); + assert_eq!(starts.load(Ordering::SeqCst), 1); + } + #[tokio::test] async fn pipeline_uses_the_updated_context_when_finalizing() { let fixture = fixture_engine( @@ -1432,7 +2223,7 @@ mod tests { } #[tokio::test] - async fn recorder_start_failure_cancels_transcription_exactly_once() { + async fn recorder_start_failure_never_starts_transcription() { let fixture = fixture_engine( false, Ok(crate::ports::PolishOutput::text("unused")), @@ -1456,10 +2247,171 @@ mod tests { .await .unwrap_err(); assert_eq!(error.code, BackendErrorCode::Platform); - assert_eq!(failing.transcription_cancels.load(Ordering::SeqCst), 1); + assert_eq!(failing.transcription_starts.load(Ordering::SeqCst), 0); + assert_eq!(failing.transcription_cancels.load(Ordering::SeqCst), 0); assert_eq!(fixture.transcription_cancels.load(Ordering::SeqCst), 0); } + #[tokio::test] + async fn realtime_pipeline_preserves_pcm_before_during_and_after_asr_start() { + let consumer = Arc::new(Mutex::new(None)); + let stops = Arc::new(AtomicUsize::new(0)); + let pcm = Arc::new(Mutex::new(Vec::new())); + let entered = Arc::new(tokio::sync::Notify::new()); + let release = Arc::new(tokio::sync::Notify::new()); + let engine = Arc::new(PipelineDictationEngine::new( + Arc::new(ExposedRecorder { + consumer: Arc::clone(&consumer), + stops: Arc::clone(&stops), + }), + Arc::new(DelayedTranscriber { + entered: Arc::clone(&entered), + release: Arc::clone(&release), + session: Arc::new(FixtureTranscriptionSession { + pcm: Arc::clone(&pcm), + cancels: Arc::new(AtomicUsize::new(0)), + finish_entered: None, + finish_release: None, + }), + }), + noop_polisher(), + )); + let progress = Arc::new(RecordingProgress::default()); + let session_id = SessionId::new(); + let starting = tokio::spawn({ + let engine = Arc::clone(&engine); + let progress = Arc::clone(&progress); + async move { + engine + .start(session_id, raw_dictation_context(), progress) + .await + } + }); + entered.notified().await; + + assert!(pcm.lock().unwrap().is_empty()); + consumer + .lock() + .unwrap() + .as_ref() + .unwrap() + .consume_pcm_chunk(&[2, 0]); + release.notify_one(); + starting.await.unwrap().unwrap(); + consumer + .lock() + .unwrap() + .as_ref() + .unwrap() + .consume_pcm_chunk(&[3, 0]); + + engine.finish(session_id, progress).await.unwrap(); + + assert_eq!(&*pcm.lock().unwrap(), &[1, 0, 2, 0, 3, 0]); + assert_eq!(stops.load(Ordering::Acquire), 1); + } + + #[tokio::test] + async fn realtime_asr_start_failure_stops_recorder_and_removes_session() { + let stops = Arc::new(AtomicUsize::new(0)); + let engine = PipelineDictationEngine::new( + Arc::new(ExposedRecorder { + consumer: Arc::new(Mutex::new(None)), + stops: Arc::clone(&stops), + }), + Arc::new(FailingTranscriber), + noop_polisher(), + ); + let error = engine + .start( + SessionId::new(), + raw_dictation_context(), + Arc::new(RecordingProgress::default()), + ) + .await + .unwrap_err(); + + assert_eq!(error.code, BackendErrorCode::Provider); + assert_eq!(stops.load(Ordering::Acquire), 1); + assert!(engine.sessions.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn provider_preparation_failure_does_not_start_the_recorder() { + let stops = Arc::new(AtomicUsize::new(0)); + let engine = PipelineDictationEngine::new( + Arc::new(ExposedRecorder { + consumer: Arc::new(Mutex::new(None)), + stops: Arc::clone(&stops), + }), + Arc::new(FailingPreparationTranscriber), + noop_polisher(), + ); + let error = engine + .start( + SessionId::new(), + raw_dictation_context(), + Arc::new(RecordingProgress::default()), + ) + .await + .unwrap_err(); + + assert_eq!(error.code, BackendErrorCode::Provider); + assert_eq!(stops.load(Ordering::Acquire), 0); + assert!(engine.sessions.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn cancel_while_realtime_asr_starts_stops_recorder_and_cancels_late_session() { + let consumer = Arc::new(Mutex::new(None)); + let stops = Arc::new(AtomicUsize::new(0)); + let pcm = Arc::new(Mutex::new(Vec::new())); + let cancels = Arc::new(AtomicUsize::new(0)); + let entered = Arc::new(tokio::sync::Notify::new()); + let release = Arc::new(tokio::sync::Notify::new()); + let engine = Arc::new(PipelineDictationEngine::new( + Arc::new(ExposedRecorder { + consumer: Arc::clone(&consumer), + stops: Arc::clone(&stops), + }), + Arc::new(DelayedTranscriber { + entered: Arc::clone(&entered), + release: Arc::clone(&release), + session: Arc::new(FixtureTranscriptionSession { + pcm: Arc::clone(&pcm), + cancels: Arc::clone(&cancels), + finish_entered: None, + finish_release: None, + }), + }), + noop_polisher(), + )); + let session_id = SessionId::new(); + let starting = tokio::spawn({ + let engine = Arc::clone(&engine); + async move { + engine + .start( + session_id, + raw_dictation_context(), + Arc::new(RecordingProgress::default()), + ) + .await + } + }); + entered.notified().await; + + engine.cancel(session_id).await.unwrap(); + assert_eq!(stops.load(Ordering::Acquire), 1); + release.notify_one(); + + let error = starting.await.unwrap().unwrap_err(); + assert_eq!(error.code, BackendErrorCode::Cancelled); + assert_eq!(cancels.load(Ordering::Acquire), 1); + assert!(pcm.lock().unwrap().is_empty()); + assert!(engine.sessions.lock().unwrap().is_empty()); + } + #[tokio::test] async fn provider_polish_failure_uses_raw_text_fallback() { let fixture = fixture_engine( @@ -1632,10 +2584,12 @@ mod tests { context.polish.mode = crate::types::PolishMode::Raw; context.polish.style_system_prompt = crate::style_packs::default_style_system_prompt_for_mode(crate::types::PolishMode::Raw); + context.recording.transcribe_after_stop = true; engine .start(session_id, Arc::new(context), progress.clone()) .await .unwrap(); + assert_eq!(starts.load(Ordering::Acquire), 0); let result = engine.finish(session_id, progress).await.unwrap(); @@ -1793,12 +2747,14 @@ mod tests { #[tokio::test] async fn cancellation_while_retry_asr_starts_cancels_the_late_resource_without_feeding_it() { - struct DelayedRetryStart { + struct RetryThenDelayedStart { + starts: Arc, entered: Arc, release: Arc, session: Arc, + pcm: Arc>>>, } - impl TranscriptionEngine for DelayedRetryStart { + impl TranscriptionEngine for RetryThenDelayedStart { fn start( &self, _: SessionId, @@ -1806,6 +2762,16 @@ mod tests { _: Arc, ) -> BoxFuture<'static, Result, BackendError>> { + if self.starts.fetch_add(1, Ordering::AcqRel) == 0 { + let pcm = Arc::clone(&self.pcm); + return Box::pin(async move { + Ok(Arc::new(RetryTranscriptionSession { + output: Err(BackendError::new(BackendErrorCode::Provider, "temporary") + .retryable(true)), + pcm, + }) as Arc) + }); + } let entered = self.entered.clone(); let release = self.release.clone(); let session = self.session.clone(); @@ -1816,19 +2782,6 @@ mod tests { }) } } - let (mut engine, progress, _) = retry_test_engine( - vec![Err(BackendError::new( - BackendErrorCode::Provider, - "temporary", - ) - .retryable(true))], - true, - ); - let session_id = SessionId::new(); - engine - .start(session_id, raw_dictation_context(), progress.clone()) - .await - .unwrap(); let entered = Arc::new(tokio::sync::Notify::new()); let release = Arc::new(tokio::sync::Notify::new()); let late = Arc::new(FixtureTranscriptionSession { @@ -1837,12 +2790,20 @@ mod tests { finish_entered: None, finish_release: None, }); - // The first session is already registered; only the retry start blocks. - engine.transcription = Arc::new(DelayedRetryStart { + let (mut engine, progress, _) = retry_test_engine(Vec::new(), true); + let pcm = Arc::new(Mutex::new(Vec::new())); + engine.transcription = Arc::new(RetryThenDelayedStart { + starts: Arc::new(AtomicUsize::new(0)), entered: entered.clone(), release: release.clone(), session: late.clone(), + pcm, }); + let session_id = SessionId::new(); + engine + .start(session_id, raw_dictation_context(), progress.clone()) + .await + .unwrap(); let engine = Arc::new(engine); let finishing = tokio::spawn({ let engine = engine.clone(); diff --git a/openless-all/app/crates/openless-core/src/lib.rs b/openless-all/app/crates/openless-core/src/lib.rs index 4d0d4bd69..bf3db2275 100644 --- a/openless-all/app/crates/openless-core/src/lib.rs +++ b/openless-all/app/crates/openless-core/src/lib.rs @@ -174,6 +174,7 @@ pub mod contract { FixtureTranscriptionEngine, LinuxCapabilityFixture, RecordingHostActions, RecordingRemoteInputRuntime, }; + pub use crate::PreparedTranscription; pub use crate::{ require_backend_contract_version, ActivityDay, AudioConsumer, AudioRecorder, BackendConfig, BackendDependencies, BackendError, BackendErrorCode, BackendEvent, BackendEventKind, @@ -279,6 +280,7 @@ pub use model_store::{ ModelTransport, ModelTransportRequest, ModelTransportResponse, ReqwestModelTransport, MODEL_PARTIAL_INDEX, MODEL_READY_SENTINEL, }; +pub use ports::PreparedTranscription; pub use ports::{ ActiveRecording, AudioConsumer, AudioRecorder, DictationEngine, DirectoryResourceResolver, EditObservationAdapter, EditObservationSink, EngineFailure, EngineFailureStage, EngineProgress, diff --git a/openless-all/app/crates/openless-core/src/ports.rs b/openless-all/app/crates/openless-core/src/ports.rs index f570c5156..0ba837b9d 100644 --- a/openless-all/app/crates/openless-core/src/ports.rs +++ b/openless-all/app/crates/openless-core/src/ports.rs @@ -194,6 +194,7 @@ pub enum EngineStage { #[derive(Debug, Clone, PartialEq)] pub enum EngineProgress { RecordingLevel { elapsed_ms: u64, level: f32 }, + RecordingLimitReached, RecordingFault(BackendError), Notification(crate::types::NotificationPayload), Stage(EngineStage), @@ -205,7 +206,7 @@ pub trait EngineProgressSink: Send + Sync { fn publish(&self, session_id: SessionId, progress: EngineProgress) -> Result<(), BackendError>; } -pub trait DictationEngine: Send + Sync { +pub trait DictationEngine: Send + Sync + 'static { fn start( &self, session_id: SessionId, @@ -255,9 +256,45 @@ pub trait DictationEngine: Send + Sync { }) } - /// Initialize ASR, then the microphone, honoring cancellation between the - /// two effects. A handle produced after cancellation must be stopped before - /// this future settles; Core keeps the voice resource hold for that lifetime. + /// Resolve the provider-facing side before the host starts recording. + /// Legacy implementations defer their existing provider-only start until + /// the returned prepared session is actually started. + fn prepare_transcription( + self: Arc, + session_id: SessionId, + context: Arc, + ) -> BoxFuture<'static, Result, BackendError>> { + Box::pin(async move { + Ok(Arc::new(DeferredDictationPrepared { + engine: self, + session_id, + context, + }) as Arc) + }) + } + + /// Provider-only capture variant used by hosts that own the microphone. + /// It uses the same buffered lifecycle as normal voice capture. + fn start_transcription_with_progress( + self: Arc, + session_id: SessionId, + context: Arc, + partials: Arc, + progress: Arc, + ) -> BoxFuture<'static, Result, BackendError>> { + let preparation = self.prepare_transcription(session_id, Arc::clone(&context)); + Box::pin(async move { + let prepared = preparation.await?; + Ok(crate::dictation_engine::buffered_transcription_session( + prepared, context, partials, progress, + )) + }) + } + + /// Start the microphone before ASR and buffer PCM while the provider connects. + /// Stable sessions defer provider startup until `TranscriptionSession::finish`. + /// A handle produced after cancellation must be stopped before this future + /// settles; Core keeps the voice resource hold for that lifetime. fn start_voice_capture( &self, _session_id: SessionId, @@ -349,6 +386,7 @@ pub trait AudioConsumer: Send + Sync { #[derive(Debug, Clone)] pub enum RecordingEvent { Level { elapsed_ms: u64, level: f32 }, + LimitReached, Fatal(BackendError), } @@ -358,6 +396,10 @@ pub trait RecordingProgressSink: Send + Sync { fn publish(&self, event: RecordingEvent) -> Result<(), BackendError> { match event { RecordingEvent::Level { elapsed_ms, level } => self.publish_level(elapsed_ms, level), + RecordingEvent::LimitReached => Err(BackendError::new( + BackendErrorCode::Unsupported, + "recording limit stop is not supported", + )), RecordingEvent::Fatal(error) => Err(error), } } @@ -462,7 +504,31 @@ pub trait TranscriptionSession: AudioConsumer { fn cancel(&self) -> BoxFuture<'static, Result<(), BackendError>>; } -pub trait TranscriptionEngine: Send + Sync { +/// Provider/channel configuration fixed before recording starts. Creating this +/// object must not open a network connection; [`Self::start`] is the delayed +/// provider startup point. +pub trait PreparedTranscription: Send + Sync { + fn start( + &self, + partials: Arc, + ) -> BoxFuture<'static, Result, BackendError>>; +} + +pub trait TranscriptionEngine: Send + Sync + 'static { + fn prepare( + self: Arc, + session_id: SessionId, + context: Arc, + ) -> BoxFuture<'static, Result, BackendError>> { + Box::pin(async move { + Ok(Arc::new(DeferredPreparedTranscription { + engine: self, + session_id, + context, + }) as Arc) + }) + } + fn start( &self, session_id: SessionId, @@ -471,6 +537,38 @@ pub trait TranscriptionEngine: Send + Sync { ) -> BoxFuture<'static, Result, BackendError>>; } +struct DeferredPreparedTranscription { + engine: Arc, + session_id: SessionId, + context: Arc, +} + +impl PreparedTranscription for DeferredPreparedTranscription { + fn start( + &self, + partials: Arc, + ) -> BoxFuture<'static, Result, BackendError>> { + self.engine + .start(self.session_id, Arc::clone(&self.context), partials) + } +} + +struct DeferredDictationPrepared { + engine: Arc, + session_id: SessionId, + context: Arc, +} + +impl PreparedTranscription for DeferredDictationPrepared { + fn start( + &self, + partials: Arc, + ) -> BoxFuture<'static, Result, BackendError>> { + self.engine + .start_transcription(self.session_id, Arc::clone(&self.context), partials) + } +} + pub trait TextPolisher: Send + Sync { fn polish( &self, diff --git a/openless-all/app/crates/openless-core/src/provider_registry.rs b/openless-all/app/crates/openless-core/src/provider_registry.rs index e5b8dac1e..347fe2e76 100644 --- a/openless-all/app/crates/openless-core/src/provider_registry.rs +++ b/openless-all/app/crates/openless-core/src/provider_registry.rs @@ -63,6 +63,19 @@ impl TranscriptionRouter { } impl TranscriptionEngine for TranscriptionRouter { + fn prepare( + self: Arc, + session_id: SessionId, + context: Arc, + ) -> BoxFuture<'static, Result, BackendError>> + { + let provider = match self.resolve(&context.asr.provider_type) { + Ok(provider) => provider, + Err(error) => return Box::pin(async move { Err(error) }), + }; + provider.prepare(session_id, context) + } + fn start( &self, session_id: SessionId, @@ -325,6 +338,19 @@ impl DictationEngine for DictationEngineRouter { engine.start_transcription(session_id, context, partials) } + fn prepare_transcription( + self: Arc, + session_id: SessionId, + context: Arc, + ) -> BoxFuture<'static, Result, BackendError>> + { + let engine = match self.resolve(&context) { + Ok(engine) => engine, + Err(error) => return Box::pin(async move { Err(error) }), + }; + engine.prepare_transcription(session_id, context) + } + fn start_voice_capture( &self, session_id: SessionId, @@ -755,6 +781,35 @@ mod tests { assert_eq!(context.llm.provider_id, "llm-channel"); } + #[tokio::test] + async fn prepared_asr_route_keeps_the_original_provider_after_replacement() { + let router = Arc::new(TranscriptionRouter::default()); + router + .register( + "asr-protocol", + Arc::new(TaggedTranscriptionEngine("original")), + ) + .unwrap(); + let mut context = + context_with_providers("asr-channel", "llm", "omni", PipelineMode::Traditional); + context.asr.provider_type = "asr-protocol".to_string(); + let context = Arc::new(context); + let prepared = Arc::clone(&router) + .prepare(SessionId::new(), context) + .await + .unwrap(); + + router + .register( + "asr-protocol", + Arc::new(TaggedTranscriptionEngine("replacement")), + ) + .unwrap(); + let session = prepared.start(Arc::new(NoopTextSink)).await.unwrap(); + + assert_eq!(session.finish().await.unwrap().text, "original"); + } + #[tokio::test] async fn engine_router_pins_the_selected_omni_adapter_for_the_session() { let traditional = Arc::new(TaggedDictationEngine::new("traditional")); diff --git a/openless-all/app/crates/openless-core/src/shared_types.rs b/openless-all/app/crates/openless-core/src/shared_types.rs index c2f9fed15..87c334826 100644 --- a/openless-all/app/crates/openless-core/src/shared_types.rs +++ b/openless-all/app/crates/openless-core/src/shared_types.rs @@ -339,6 +339,9 @@ pub struct UserPreferences { /// 录音期间临时静音系统输出，停止/取消/出错后恢复原静音状态。 #[serde(default)] pub mute_during_recording: bool, + /// 录音结束后再连接当前 ASR 并提交整段 PCM。默认关闭。 + #[serde(default)] + pub stable_transcription_enabled: bool, /// 按下录音热键进入 recording 状态时，播放一段即时合成的提示音，提醒「已开始录音」。 /// 默认开启；可在「录音与输入」设置里关闭。提示音由 capsule 窗口用 Web Audio API 合成， /// 不依赖 show_capsule —— 胶囊隐藏时仍会响。 @@ -781,6 +784,8 @@ struct UserPreferencesWire { capsule_style: CapsuleStyle, #[serde(default)] mute_during_recording: bool, + #[serde(default)] + stable_transcription_enabled: bool, #[serde(default = "default_true")] audio_cue_on_record: bool, #[serde(default)] @@ -1015,6 +1020,7 @@ impl Default for UserPreferencesWire { show_capsule: prefs.show_capsule, capsule_style: prefs.capsule_style, mute_during_recording: prefs.mute_during_recording, + stable_transcription_enabled: prefs.stable_transcription_enabled, audio_cue_on_record: prefs.audio_cue_on_record, silence_auto_stop_enabled: prefs.silence_auto_stop_enabled, silence_auto_stop_seconds: prefs.silence_auto_stop_seconds, @@ -1173,6 +1179,7 @@ impl<'de> Deserialize<'de> for UserPreferences { show_capsule: wire.show_capsule, capsule_style: wire.capsule_style, mute_during_recording: wire.mute_during_recording, + stable_transcription_enabled: wire.stable_transcription_enabled, audio_cue_on_record: wire.audio_cue_on_record, silence_auto_stop_enabled: wire.silence_auto_stop_enabled, silence_auto_stop_seconds: wire.silence_auto_stop_seconds, @@ -1531,6 +1538,7 @@ impl Default for UserPreferences { show_capsule: true, capsule_style: CapsuleStyle::Siri, mute_during_recording: false, + stable_transcription_enabled: false, audio_cue_on_record: true, silence_auto_stop_enabled: false, silence_auto_stop_seconds: default_silence_auto_stop_seconds(), @@ -2934,6 +2942,21 @@ mod tests { assert!(!restored.audio_cue_on_record); } + #[test] + fn stable_transcription_defaults_off_and_round_trips_when_enabled() { + let legacy: UserPreferences = serde_json::from_str("{}").unwrap(); + assert!(!legacy.stable_transcription_enabled); + + let enabled = UserPreferences { + stable_transcription_enabled: true, + ..Default::default() + }; + let json = serde_json::to_string(&enabled).unwrap(); + assert!(json.contains("\"stableTranscriptionEnabled\":true")); + let restored: UserPreferences = serde_json::from_str(&json).unwrap(); + assert!(restored.stable_transcription_enabled); + } + #[test] fn action_hotkeys_default_to_enabled() { // issue #576：默认仍开启（Some 默认键），对老用户零行为变化。 diff --git a/openless-all/app/crates/openless-core/tests/auxiliary_voice_lifecycle.rs b/openless-all/app/crates/openless-core/tests/auxiliary_voice_lifecycle.rs index 24f04683a..b60957fe2 100644 --- a/openless-all/app/crates/openless-core/tests/auxiliary_voice_lifecycle.rs +++ b/openless-all/app/crates/openless-core/tests/auxiliary_voice_lifecycle.rs @@ -135,6 +135,23 @@ struct SlowAsr { inner: testing::FixtureTranscriptionEngine, } +struct CountingAsr { + starts: Arc, + inner: testing::FixtureTranscriptionEngine, +} + +impl TranscriptionEngine for CountingAsr { + fn start( + &self, + id: SessionId, + context: Arc, + sink: Arc, + ) -> BoxFuture<'static, Result, BackendError>> { + self.starts.fetch_add(1, Ordering::SeqCst); + self.inner.start(id, context, sink) + } +} + // Model only the native archive boundary: once an archive was requested, a // filesystem sharing violation would make its final deletion fail. PCM remains // available in memory and must not depend on this optional disk side effect. @@ -438,6 +455,105 @@ async fn qa_and_selection_voice_never_request_disk_archives() { } } +#[tokio::test] +async fn stable_mode_is_shared_by_dictation_qa_selection_and_less_computer() { + for entry in ["dictation", "qa", "selection", "less"] { + let starts = Arc::new(AtomicUsize::new(0)); + let (backend, path) = backend( + Arc::new(testing::FixtureAudioRecorder::new( + vec![vec![0; 320]], + Vec::new(), + )), + Arc::new(CountingAsr { + starts: Arc::clone(&starts), + inner: testing::FixtureTranscriptionEngine::successful("instruction", 10), + }), + Arc::new(QaRuntime::default()), + ); + let mut preferences = backend.get_preferences(); + preferences.stable_transcription_enabled = true; + backend + .update_settings( + preferences, + SettingsUpdateOptions::STRICT, + &NoopSettingsRuntime, + ) + .unwrap(); + backend.start().await.unwrap(); + + match entry { + "dictation" => { + backend + .start_dictation_with_options(DictationStartOptions { + insert_text: false, + ..DictationStartOptions::default() + }) + .await + .unwrap(); + assert_eq!(starts.load(Ordering::SeqCst), 0, "{entry}"); + backend.stop_dictation().await.unwrap(); + } + "qa" => { + backend.services().qa.toggle_recording().await.unwrap(); + let id = backend + .services() + .qa + .snapshot() + .await + .unwrap() + .session_id + .unwrap(); + let capture = backend + .start_qa_voice_capture( + id, + DictationStartOptions::default(), + Arc::new(Progress), + ) + .await + .unwrap(); + assert_eq!(starts.load(Ordering::SeqCst), 0, "{entry}"); + capture.finish().await.unwrap(); + backend.services().qa.dismiss().await.unwrap(); + } + "selection" => { + let id = backend + .services() + .selection_voice + .begin(SelectionCapture { + text: "selection".into(), + source_app: None, + }) + .await + .unwrap(); + let capture = backend + .start_selection_voice_capture(id, Arc::new(Control)) + .await + .unwrap(); + assert_eq!(starts.load(Ordering::SeqCst), 0, "{entry}"); + capture.finish().await.unwrap(); + backend + .services() + .selection_voice + .cancel(Some(id)) + .await + .unwrap(); + } + "less" => { + let capture = backend + .start_less_computer_voice(SessionId::new(), Arc::new(Control)) + .await + .unwrap(); + assert_eq!(starts.load(Ordering::SeqCst), 0, "{entry}"); + let _ = capture.finish().await; + } + _ => unreachable!(), + } + assert_eq!(starts.load(Ordering::SeqCst), 1, "{entry}"); + drop(backend); + std::fs::remove_dir_all(path).unwrap(); + } +} + #[tokio::test] async fn cli_cancel_covers_less_capture_without_expanding_to_qa() { let recorder = Arc::new(testing::FixtureAudioRecorder::default()); @@ -552,7 +668,7 @@ async fn cancelled_auxiliary_capture_keeps_gate_until_native_stop_finishes() { } #[tokio::test] -async fn cancellation_during_cold_asr_never_starts_the_microphone() { +async fn cancellation_during_cold_asr_stops_the_already_started_microphone() { for less in [true, false] { let starts = Arc::new(AtomicUsize::new(0)); let entered = Arc::new(Semaphore::new(0)); @@ -614,7 +730,7 @@ async fn cancellation_during_cold_asr_never_starts_the_microphone() { } gate.add_permits(1); assert!(starting.await.unwrap().is_err()); - assert_eq!(starts.load(Ordering::SeqCst), 0); + assert_eq!(starts.load(Ordering::SeqCst), 1); if less { let phases = std::iter::from_fn(|| events.try_recv().ok()) .filter_map(|event| match event.kind { diff --git a/openless-all/app/crates/openless-core/tests/cloud_sync_contract.rs b/openless-all/app/crates/openless-core/tests/cloud_sync_contract.rs index 7b5f9bd5d..2dad33788 100644 --- a/openless-all/app/crates/openless-core/tests/cloud_sync_contract.rs +++ b/openless-all/app/crates/openless-core/tests/cloud_sync_contract.rs @@ -244,6 +244,7 @@ async fn portable_data_round_trips_without_uploading_credentials_or_device_setti preferences.remote_input_pin = "never-sync-device-pin".into(); preferences.active_asr_provider = "source-only-asr".into(); preferences.theme_mode = openless_core::shared_types::ThemeMode::Dark; + preferences.stable_transcription_enabled = true; source .backend .repositories() @@ -268,6 +269,7 @@ async fn portable_data_round_trips_without_uploading_credentials_or_device_setti assert_eq!(saved.counts.corrections, 1); let uploaded = server.state.lock().unwrap().payload.clone().unwrap(); let serialized = uploaded.to_string(); + assert_eq!(uploaded["preferences"]["stableTranscriptionEnabled"], true); for private in [ "cloud-fixture-token", "never-sync-asr-secret", @@ -330,6 +332,7 @@ async fn portable_data_round_trips_without_uploading_credentials_or_device_setti Some("/private/target-workspace") ); assert_eq!(preferences.remote_input_pin, "target-pin"); + assert!(preferences.stable_transcription_enabled); let secret = target .credentials .read( @@ -348,6 +351,27 @@ async fn portable_data_round_trips_without_uploading_credentials_or_device_setti reopened.preferences.get().active_style_pack_id, "custom.cloud" ); + + let legacy_target = Device::new(&server).await; + let mut local = legacy_target.backend.get_preferences(); + local.stable_transcription_enabled = true; + legacy_target + .backend + .repositories() + .preferences + .set(local) + .unwrap(); + server.state.lock().unwrap().payload.as_mut().unwrap()["preferences"] + .as_object_mut() + .unwrap() + .remove("stableTranscriptionEnabled"); + legacy_target.backend.cloud_sync_restore().await.unwrap(); + assert!( + legacy_target + .backend + .get_preferences() + .stable_transcription_enabled + ); } #[tokio::test] diff --git a/openless-all/app/linux-egui/src/main.rs b/openless-all/app/linux-egui/src/main.rs index eee6f63c8..dafefe53f 100644 --- a/openless-all/app/linux-egui/src/main.rs +++ b/openless-all/app/linux-egui/src/main.rs @@ -1575,6 +1575,11 @@ mod linux_app { ui.heading("环境与设置"); ui.strong("现有功能设置"); if let Some(preferences) = self.preferences.as_mut() { + ui.checkbox( + &mut preferences.stable_transcription_enabled, + "稳定模式（先录音后识别）", + ); + ui.small("录音期间不连接 ASR，停止后提交整段音频；结果更晚，但录音不受建连延迟和网络抖动影响。"); ui.checkbox(&mut preferences.streaming_insert, "流式插入"); ui.small("将转写逐步发送到原输入目标，实际结果以听写与历史反馈为准。"); ui.checkbox(&mut preferences.coding_agent_enabled, "启用 Less Computer"); diff --git a/openless-all/app/src/i18n/de.ts b/openless-all/app/src/i18n/de.ts index 4e04956f3..ea3628ad5 100644 --- a/openless-all/app/src/i18n/de.ts +++ b/openless-all/app/src/i18n/de.ts @@ -1119,6 +1119,9 @@ export const de: typeof zhCN = { capsuleStyleLabel: 'Kapselstil', capsuleStyleSiri: 'Schimmernder Siri-Stil', capsuleStyleClassic: 'OpenLess-Standardstil', + stableTranscriptionLabel: 'Stabiler Modus (erst aufnehmen, dann transkribieren)', + stableTranscriptionDesc: + 'Wenn aktiviert, verbindet sich ASR erst nach dem Aufnahmestopp und erhält die gesamte Aufnahme. Das Ergebnis erscheint später, aber Verbindungsaufbau und Netzwerkschwankungen während der Aufnahme stören die Aufzeichnung nicht.', muteDuringRecordingLabel: 'Während der Aufnahme stummschalten', muteDuringRecordingDesc: 'Schaltet die Systemausgabe während der Spracheingabe vorübergehend stumm, um Lautsprecherechos zu vermeiden.', diff --git a/openless-all/app/src/i18n/en.ts b/openless-all/app/src/i18n/en.ts index cd55ef226..bf2d03b75 100644 --- a/openless-all/app/src/i18n/en.ts +++ b/openless-all/app/src/i18n/en.ts @@ -1095,6 +1095,9 @@ export const en: typeof zhCN = { capsuleStyleLabel: 'Capsule style', capsuleStyleSiri: 'Shimmer Siri style', capsuleStyleClassic: 'OpenLess default style', + stableTranscriptionLabel: 'Stable mode (record, then transcribe)', + stableTranscriptionDesc: + 'When enabled, ASR connects only after recording stops and receives the complete audio. Results arrive later, but connection delays and network instability during recording cannot interrupt capture.', muteDuringRecordingLabel: 'Mute while recording', muteDuringRecordingDesc: 'Temporarily mute system output during voice input to avoid speaker echo.', diff --git a/openless-all/app/src/i18n/es.ts b/openless-all/app/src/i18n/es.ts index ec43ba5cf..0c71ec6f2 100644 --- a/openless-all/app/src/i18n/es.ts +++ b/openless-all/app/src/i18n/es.ts @@ -1114,6 +1114,9 @@ export const es: typeof zhCN = { capsuleStyleLabel: 'Estilo de cápsula', capsuleStyleSiri: 'Estilo luminoso Siri', capsuleStyleClassic: 'Estilo predeterminado de OpenLess', + stableTranscriptionLabel: 'Modo estable (grabar y luego transcribir)', + stableTranscriptionDesc: + 'Al activarlo, ASR se conecta solo después de detener la grabación y recibe el audio completo. El resultado tarda más, pero la captura no se ve afectada por retrasos de conexión ni por inestabilidad de red durante la grabación.', muteDuringRecordingLabel: 'Silenciar durante la grabación', muteDuringRecordingDesc: 'Silencia temporalmente el sonido del sistema durante la entrada de voz para evitar el eco de los altavoces.', diff --git a/openless-all/app/src/i18n/fr.ts b/openless-all/app/src/i18n/fr.ts index 7a1850fe8..692b1e9d2 100644 --- a/openless-all/app/src/i18n/fr.ts +++ b/openless-all/app/src/i18n/fr.ts @@ -1126,6 +1126,9 @@ export const fr: typeof zhCN = { capsuleStyleLabel: 'Style de capsule', capsuleStyleSiri: 'Style lumineux Siri', capsuleStyleClassic: 'Style par défaut d’OpenLess', + stableTranscriptionLabel: 'Mode stable (enregistrer puis transcrire)', + stableTranscriptionDesc: + 'Une fois activé, l’ASR se connecte après l’arrêt et reçoit l’enregistrement complet. Le résultat arrive plus tard, mais la capture n’est pas perturbée par le délai de connexion ni par les variations du réseau pendant l’enregistrement.', muteDuringRecordingLabel: 'Couper le son pendant l’enregistrement', muteDuringRecordingDesc: 'Coupe temporairement le son système pendant la saisie vocale pour éviter l’écho des haut-parleurs.', diff --git a/openless-all/app/src/i18n/ja.ts b/openless-all/app/src/i18n/ja.ts index 74f7ae579..7e5440262 100644 --- a/openless-all/app/src/i18n/ja.ts +++ b/openless-all/app/src/i18n/ja.ts @@ -1083,6 +1083,9 @@ export const ja: typeof zhCN = { capsuleStyleLabel: 'カプセルスタイル', capsuleStyleSiri: '光条 Siri スタイル', capsuleStyleClassic: 'Openless デフォルトスタイル', + stableTranscriptionLabel: '安定モード（録音後に文字起こし）', + stableTranscriptionDesc: + '有効にすると、録音中は ASR に接続せず、停止後に音声全体を送信します。結果は遅くなりますが、接続遅延や録音中のネットワーク変動に録音が影響されません。', muteDuringRecordingLabel: '録音中はミュート', muteDuringRecordingDesc: '録音中にシステム出力を一時的にミュートし、スピーカーのエコーを防ぎます。', diff --git a/openless-all/app/src/i18n/ko.ts b/openless-all/app/src/i18n/ko.ts index b8b7cf6a7..6dc745a55 100644 --- a/openless-all/app/src/i18n/ko.ts +++ b/openless-all/app/src/i18n/ko.ts @@ -1075,6 +1075,9 @@ export const ko: typeof zhCN = { capsuleStyleLabel: '캡슐 스타일', capsuleStyleSiri: '시리 광선 스타일', capsuleStyleClassic: 'Openless 기본 스타일', + stableTranscriptionLabel: '안정 모드 (녹음 후 전사)', + stableTranscriptionDesc: + '켜면 녹음 중에는 ASR에 연결하지 않고 중지한 뒤 전체 오디오를 전송합니다. 결과는 늦게 표시되지만 연결 지연이나 녹음 중 네트워크 불안정이 녹음에 영향을 주지 않습니다.', muteDuringRecordingLabel: '녹음 중 음소거', muteDuringRecordingDesc: '녹음 중 시스템 출력을 일시적으로 음소거하여 스피커 에코를 방지합니다.', diff --git a/openless-all/app/src/i18n/zh-CN.ts b/openless-all/app/src/i18n/zh-CN.ts index 2753f4b19..f503982eb 100644 --- a/openless-all/app/src/i18n/zh-CN.ts +++ b/openless-all/app/src/i18n/zh-CN.ts @@ -1047,6 +1047,9 @@ export const zhCN = { capsuleStyleLabel: '胶囊样式', capsuleStyleSiri: '流光 Siri 风格', capsuleStyleClassic: 'Openless 默认风格', + stableTranscriptionLabel: '稳定模式（先录音后识别）', + stableTranscriptionDesc: + '开启后，录音期间不连接 ASR，停止后才提交整段音频。结果出现更晚，但录音不受建连延迟和录音期间网络抖动影响。', muteDuringRecordingLabel: '录音时静音', muteDuringRecordingDesc: '录音期间临时静音系统输出，避免扬声器回音。', audioCueLabel: '录音提示音', diff --git a/openless-all/app/src/i18n/zh-TW.ts b/openless-all/app/src/i18n/zh-TW.ts index dedc72735..e2b3cff71 100644 --- a/openless-all/app/src/i18n/zh-TW.ts +++ b/openless-all/app/src/i18n/zh-TW.ts @@ -1060,6 +1060,9 @@ export const zhTW: typeof zhCN = { capsuleStyleLabel: '膠囊樣式', capsuleStyleSiri: '流光 Siri 風格', capsuleStyleClassic: 'Openless 預設風格', + stableTranscriptionLabel: '穩定模式（先錄音後辨識）', + stableTranscriptionDesc: + '開啟後，錄音期間不連接 ASR，停止後才提交整段音訊。結果會較晚出現，但錄音不受連線延遲和錄音期間的網路波動影響。', muteDuringRecordingLabel: '錄音時靜音', muteDuringRecordingDesc: '錄音期間臨時靜音系統輸出，避免揚聲器回音。', audioCueLabel: '錄音提示音', diff --git a/openless-all/app/src/lib/ipc/mock-data.ts b/openless-all/app/src/lib/ipc/mock-data.ts index eefc40e39..0298d0e01 100644 --- a/openless-all/app/src/lib/ipc/mock-data.ts +++ b/openless-all/app/src/lib/ipc/mock-data.ts @@ -48,6 +48,7 @@ export let mockSettings: UserPreferences = { showCapsule: true, capsuleStyle: 'siri', muteDuringRecording: false, + stableTranscriptionEnabled: false, audioCueOnRecord: true, silenceAutoStopEnabled: false, silenceAutoStopSeconds: 3, diff --git a/openless-all/app/src/lib/types.ts b/openless-all/app/src/lib/types.ts index 701706543..25ecc2cd2 100644 --- a/openless-all/app/src/lib/types.ts +++ b/openless-all/app/src/lib/types.ts @@ -345,6 +345,8 @@ export interface UserPreferences { capsuleStyle: CapsuleStyle; /** 录音期间临时静音系统输出，停止/取消/出错后恢复原静音状态。 */ muteDuringRecording: boolean; + /** 先完整录音，停止后再连接当前 ASR 并提交整段音频。默认关闭。 */ + stableTranscriptionEnabled: boolean; /** 按下录音热键进入 recording 状态时，播放一段合成提示音提醒「已开始录音」。 * 默认开启；在 capsule 窗口用 Web Audio API 合成，不依赖 showCapsule。 */ audioCueOnRecord: boolean; diff --git a/openless-all/app/src/pages/settings/RecordingInputSection.tsx b/openless-all/app/src/pages/settings/RecordingInputSection.tsx index 373d8e288..6a7f94abc 100644 --- a/openless-all/app/src/pages/settings/RecordingInputSection.tsx +++ b/openless-all/app/src/pages/settings/RecordingInputSection.tsx @@ -448,6 +448,15 @@ export function RecordingInputSection() { )} + + savePrefs({ ...prefs, stableTranscriptionEnabled: next })} + /> +