diff --git a/crates/transcription/src/local_engine.rs b/crates/transcription/src/local_engine.rs index f13805d..dac5f9c 100644 --- a/crates/transcription/src/local_engine.rs +++ b/crates/transcription/src/local_engine.rs @@ -61,6 +61,32 @@ impl Transcriber for SpeechModelAdapter { }) .collect()) } + + /// SAFETY: `transcribe-rs` owns the Parakeet decoder behind a `&mut + /// self` call that does not surface a cancellation hook. The best we + /// can do without forking transcribe-rs is short-circuit BEFORE the + /// decode call when the live session has already requested abort — + /// the same boundary the `Drop for InferenceTask` cancellation + /// route arrives through. Once the decode is in flight it runs to + /// completion, but the live session's `drain_inference` timeout + /// still drops our receiver, so the wedged orphan thread exits the + /// instant it tries to send its result. The engine `Mutex` is held + /// only across THIS call, so a future task will not deadlock — it + /// simply queues until the orphan releases. Documenting the + /// uncancellable middle so future audits don't get surprised. + fn transcribe_sync_with_abort( + &mut self, + samples: &[f32], + options: &TranscriptionOptions, + abort_flag: Arc, + ) -> Result> { + if abort_flag.load(std::sync::atomic::Ordering::Relaxed) { + return Err(Error::TranscriptionFailed( + "transcription aborted before decoder dispatch".to_string(), + )); + } + self.transcribe_sync(samples, options) + } } /// Owns the currently-loaded speech backend and serialises inference @@ -159,9 +185,10 @@ impl LocalEngine { /// Cancellable variant of `transcribe_sync`. Pipes `abort_flag` /// through to the backend so the live-session drain timeout can - /// break a wedged whisper-rs decode out of its loop. Backends that - /// can't honour the flag (Parakeet) fall through to their plain - /// `transcribe_sync` via the trait's default implementation. + /// break a wedged whisper-rs decode out of its loop. Every backend + /// MUST implement the trait method (no default impl) — Parakeet's + /// adapter, for example, honours the flag at the pre-decode + /// boundary even though the in-flight decode itself is opaque. pub fn transcribe_sync_with_abort( &self, audio: &AudioSamples, @@ -244,6 +271,8 @@ pub fn load_whisper(model_path: &Path) -> Result> { #[cfg(test)] mod tests { use super::*; + use std::sync::atomic::Ordering; + use transcribe_rs::{ModelCapabilities, TranscribeError, TranscribeOptions, TranscriptionResult}; #[test] fn engine_reports_not_available_before_loading() { @@ -252,4 +281,86 @@ mod tests { assert!(engine.loaded_model_id().is_none()); assert!(engine.capabilities().is_none()); } + + /// Minimal fake `SpeechModel` for the Parakeet adapter tests. Records + /// whether `transcribe_raw` was called so we can prove the pre-decode + /// abort short-circuit actually fires. + struct FakeSpeechModel { + call_count: Arc, + } + + impl transcribe_rs::SpeechModel for FakeSpeechModel { + fn capabilities(&self) -> ModelCapabilities { + ModelCapabilities { + name: "fake", + engine_id: "fake", + sample_rate: 16_000, + languages: &[], + supports_timestamps: false, + supports_translation: false, + supports_streaming: false, + } + } + + fn transcribe_raw( + &mut self, + _samples: &[f32], + _options: &TranscribeOptions, + ) -> std::result::Result { + self.call_count + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + Ok(TranscriptionResult { + text: String::new(), + segments: Some(Vec::new()), + }) + } + } + + #[test] + fn speech_model_adapter_short_circuits_when_abort_set_pre_dispatch() { + // Lifecycle-2 regression: without an explicit + // `transcribe_sync_with_abort` impl, SpeechModelAdapter used to + // inherit a default that silently dropped the abort flag and + // ran the decoder anyway. With the trait method made required, + // the adapter now checks the flag at the safest available + // boundary (pre-decode) and short-circuits. + let call_count = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let model = Box::new(FakeSpeechModel { + call_count: call_count.clone(), + }); + let mut adapter = SpeechModelAdapter(model); + + let abort = Arc::new(AtomicBool::new(true)); + let options = TranscriptionOptions::default(); + let res = adapter.transcribe_sync_with_abort(&[0.0_f32; 16], &options, abort); + assert!(res.is_err(), "pre-set abort flag must short-circuit"); + assert_eq!( + call_count.load(Ordering::Relaxed), + 0, + "underlying decoder must NOT be called when abort was set before dispatch" + ); + } + + #[test] + fn speech_model_adapter_dispatches_decoder_when_abort_clear() { + // Companion to the short-circuit test: when the abort flag is + // clear at dispatch time, the adapter must still call the + // underlying decoder. Without this we'd have killed + // transcription entirely. + let call_count = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let model = Box::new(FakeSpeechModel { + call_count: call_count.clone(), + }); + let mut adapter = SpeechModelAdapter(model); + + let abort = Arc::new(AtomicBool::new(false)); + let options = TranscriptionOptions::default(); + let res = adapter.transcribe_sync_with_abort(&[0.0_f32; 16], &options, abort); + assert!(res.is_ok(), "clear abort flag must allow dispatch"); + assert_eq!( + call_count.load(Ordering::Relaxed), + 1, + "decoder must run exactly once when abort flag is clear" + ); + } } diff --git a/crates/transcription/src/transcriber.rs b/crates/transcription/src/transcriber.rs index 63aebd6..5482262 100644 --- a/crates/transcription/src/transcriber.rs +++ b/crates/transcription/src/transcriber.rs @@ -51,26 +51,38 @@ pub trait Transcriber: Send { /// Variant of `transcribe_sync` that accepts an external abort flag. /// - /// Backends that can react to mid-inference cancellation (whisper-rs - /// via `set_abort_callback_safe`) override this to poll the flag - /// during decoding so a wedged inference can be unstuck by the live - /// session's `drain_inference` timeout. The default implementation - /// ignores the flag and dispatches to `transcribe_sync` so - /// non-cancellable backends (Parakeet via transcribe-rs) keep their - /// current behaviour. + /// REQUIRED so every backend opts in to a cancellation story at + /// compile time. Without this requirement a default impl that + /// simply forwarded to `transcribe_sync` would silently re-introduce + /// the wedge for any non-whisper backend: the live session's + /// `drain_inference` timeout would set the flag, but the backend + /// would ignore it, the orphan inference thread would keep holding + /// the engine `Mutex`, and the next start/stop would deadlock on it. + /// + /// Implementer guidance: + /// * Backends that can react to mid-inference cancellation + /// (whisper-rs via `set_abort_callback_safe`) wire the flag into + /// their decoder's abort hook so the wedged inference can be + /// unstuck by the live session's drain timeout. + /// * Backends that genuinely cannot honour the flag mid-call + /// (synchronous external API calls, transcribe-rs adapters that + /// own opaque decoder state) MUST still implement this method — + /// even if the implementation is to check the flag at the safest + /// available boundary and otherwise dispatch to `transcribe_sync`. + /// Document the uncancellability with a `// SAFETY:` comment so a + /// future audit can find it. fn transcribe_sync_with_abort( &mut self, samples: &[f32], options: &TranscriptionOptions, - _abort_flag: Arc, - ) -> Result> { - self.transcribe_sync(samples, options) - } + abort_flag: Arc, + ) -> Result>; } #[cfg(test)] mod tests { use super::*; + use std::sync::atomic::Ordering; #[test] fn transcriber_trait_is_object_safe() { @@ -79,4 +91,63 @@ mod tests { // method) this declaration fails to build. No runtime work. let _: Option> = None; } + + /// Fake backend that records whether the abort flag was observed + /// at dispatch time. Proves the compile-time-required + /// `transcribe_sync_with_abort` actually receives the flag instead + /// of silently falling back to a default impl that drops it on the + /// floor (the Lifecycle-2 wedge). + struct FlagSnoopingBackend { + observed_abort: bool, + } + + impl Transcriber for FlagSnoopingBackend { + fn capabilities(&self) -> TranscriberCapabilities { + TranscriberCapabilities { + sample_rate: 16_000, + channels: 1, + supports_initial_prompt: false, + } + } + + fn transcribe_sync( + &mut self, + _samples: &[f32], + _options: &TranscriptionOptions, + ) -> Result> { + Ok(Vec::new()) + } + + fn transcribe_sync_with_abort( + &mut self, + _samples: &[f32], + _options: &TranscriptionOptions, + abort_flag: Arc, + ) -> Result> { + self.observed_abort = abort_flag.load(Ordering::Relaxed); + Ok(Vec::new()) + } + } + + #[test] + fn transcribe_sync_with_abort_is_required_and_flag_is_observed() { + // Lifecycle-2 regression: removing the trait's default impl + // forces every backend to receive the abort flag at the call + // site. Without an explicit method on this fake backend the + // file wouldn't compile; with it, asserting the snoop is the + // runtime witness that the flag is actually piped through. + let mut backend = FlagSnoopingBackend { + observed_abort: false, + }; + let flag = Arc::new(AtomicBool::new(true)); + let options = TranscriptionOptions::default(); + let segs = backend + .transcribe_sync_with_abort(&[], &options, flag.clone()) + .expect("fake backend never errors"); + assert!(segs.is_empty()); + assert!( + backend.observed_abort, + "trait must hand the abort flag to the backend, not drop it via a default impl" + ); + } }