agent: code-atomiser-fix — require Transcriber::transcribe_sync_with_abort (Lifecycle-2)
The trait's default implementation of transcribe_sync_with_abort fell
through to plain transcribe_sync, silently dropping the abort flag for
any backend that did not explicitly override it. That re-introduced the
5725836 wedge for SpeechModelAdapter (Parakeet today, any future
cloud STT or transformer adapter tomorrow): the live session's
drain_inference timeout would set the flag, the backend would ignore
it, the orphan inference thread would keep the engine Mutex held, and
the next start/stop would deadlock on it.
Removing the default impl makes the method required, so any backend
that does not implement it fails to compile. Compile-time enforcement
of cancellation completeness.
SpeechModelAdapter now implements the method explicitly with a
pre-decode short-circuit: if the abort flag is already set before we
dispatch into transcribe-rs (which owns the Parakeet decoder behind an
opaque call that has no cancellation hook), return an error
immediately. The in-flight decode itself is still uncancellable, but
that is documented with a SAFETY comment and bounded by the live
session's drain timeout dropping the receiver — the orphan exits the
instant it tries to send.
Regression tests:
- transcriber::tests::transcribe_sync_with_abort_is_required_and_flag_is_observed —
fake backend records the abort flag at dispatch time; proves the
trait method is required (file would not compile without it) and the
flag is actually piped through.
- local_engine::tests::speech_model_adapter_short_circuits_when_abort_set_pre_dispatch —
SpeechModelAdapter must NOT call the underlying decoder when the
abort flag was set before dispatch.
- local_engine::tests::speech_model_adapter_dispatches_decoder_when_abort_clear —
companion: clear flag must still dispatch (we did not kill
transcription entirely).
cargo test -p lumotia-transcription --lib: 64 passed, 0 failed.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -61,6 +61,32 @@ impl Transcriber for SpeechModelAdapter {
|
|||||||
})
|
})
|
||||||
.collect())
|
.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<AtomicBool>,
|
||||||
|
) -> Result<Vec<Segment>> {
|
||||||
|
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
|
/// Owns the currently-loaded speech backend and serialises inference
|
||||||
@@ -159,9 +185,10 @@ impl LocalEngine {
|
|||||||
|
|
||||||
/// Cancellable variant of `transcribe_sync`. Pipes `abort_flag`
|
/// Cancellable variant of `transcribe_sync`. Pipes `abort_flag`
|
||||||
/// through to the backend so the live-session drain timeout can
|
/// through to the backend so the live-session drain timeout can
|
||||||
/// break a wedged whisper-rs decode out of its loop. Backends that
|
/// break a wedged whisper-rs decode out of its loop. Every backend
|
||||||
/// can't honour the flag (Parakeet) fall through to their plain
|
/// MUST implement the trait method (no default impl) — Parakeet's
|
||||||
/// `transcribe_sync` via the trait's default implementation.
|
/// 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(
|
pub fn transcribe_sync_with_abort(
|
||||||
&self,
|
&self,
|
||||||
audio: &AudioSamples,
|
audio: &AudioSamples,
|
||||||
@@ -244,6 +271,8 @@ pub fn load_whisper(model_path: &Path) -> Result<Box<dyn Transcriber + Send>> {
|
|||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
use std::sync::atomic::Ordering;
|
||||||
|
use transcribe_rs::{ModelCapabilities, TranscribeError, TranscribeOptions, TranscriptionResult};
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn engine_reports_not_available_before_loading() {
|
fn engine_reports_not_available_before_loading() {
|
||||||
@@ -252,4 +281,86 @@ mod tests {
|
|||||||
assert!(engine.loaded_model_id().is_none());
|
assert!(engine.loaded_model_id().is_none());
|
||||||
assert!(engine.capabilities().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<std::sync::atomic::AtomicUsize>,
|
||||||
|
}
|
||||||
|
|
||||||
|
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<TranscriptionResult, TranscribeError> {
|
||||||
|
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"
|
||||||
|
);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -51,26 +51,38 @@ pub trait Transcriber: Send {
|
|||||||
|
|
||||||
/// Variant of `transcribe_sync` that accepts an external abort flag.
|
/// Variant of `transcribe_sync` that accepts an external abort flag.
|
||||||
///
|
///
|
||||||
/// Backends that can react to mid-inference cancellation (whisper-rs
|
/// REQUIRED so every backend opts in to a cancellation story at
|
||||||
/// via `set_abort_callback_safe`) override this to poll the flag
|
/// compile time. Without this requirement a default impl that
|
||||||
/// during decoding so a wedged inference can be unstuck by the live
|
/// simply forwarded to `transcribe_sync` would silently re-introduce
|
||||||
/// session's `drain_inference` timeout. The default implementation
|
/// the wedge for any non-whisper backend: the live session's
|
||||||
/// ignores the flag and dispatches to `transcribe_sync` so
|
/// `drain_inference` timeout would set the flag, but the backend
|
||||||
/// non-cancellable backends (Parakeet via transcribe-rs) keep their
|
/// would ignore it, the orphan inference thread would keep holding
|
||||||
/// current behaviour.
|
/// 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(
|
fn transcribe_sync_with_abort(
|
||||||
&mut self,
|
&mut self,
|
||||||
samples: &[f32],
|
samples: &[f32],
|
||||||
options: &TranscriptionOptions,
|
options: &TranscriptionOptions,
|
||||||
_abort_flag: Arc<AtomicBool>,
|
abort_flag: Arc<AtomicBool>,
|
||||||
) -> Result<Vec<Segment>> {
|
) -> Result<Vec<Segment>>;
|
||||||
self.transcribe_sync(samples, options)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
use std::sync::atomic::Ordering;
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn transcriber_trait_is_object_safe() {
|
fn transcriber_trait_is_object_safe() {
|
||||||
@@ -79,4 +91,63 @@ mod tests {
|
|||||||
// method) this declaration fails to build. No runtime work.
|
// method) this declaration fails to build. No runtime work.
|
||||||
let _: Option<Box<dyn Transcriber + Send>> = None;
|
let _: Option<Box<dyn Transcriber + Send>> = 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<Vec<Segment>> {
|
||||||
|
Ok(Vec::new())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn transcribe_sync_with_abort(
|
||||||
|
&mut self,
|
||||||
|
_samples: &[f32],
|
||||||
|
_options: &TranscriptionOptions,
|
||||||
|
abort_flag: Arc<AtomicBool>,
|
||||||
|
) -> Result<Vec<Segment>> {
|
||||||
|
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"
|
||||||
|
);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user