mozu-audio 0.2.0

//! Offline long-audio ASR over bounded canonical sample sources.
pub use crate::audio::WavSource;
use crate::audio::{AudioSampleSource, AudioSource};
use crate::result::{AsrResult, ProfiledRecognizer};
use crate::{
    Cancellation, Error, HaasEngine, InputLimits, Result,
    recording::SAMPLE_RATE,
    vad::{SileroVad, SpeechBoundaries, SpeechBoundary, VoiceDetector},
    window::{
        AsrConfig, ENCODER_STEP, TimedToken, Transcript, WindowRecognizer, align_tokens,
        recognize_window,
    },
};
use serde::Serialize;
use std::{collections::BTreeSet, ops::Range, path::Path};
#[derive(Clone, Copy, Debug, Serialize)]
pub struct BatchProgress {
    pub processed_samples: u64,
    pub total_samples: u64,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum BatchUpdateKind {
    Partial,
    Final,
}

/// A callback-borrowed transcript. Partial text and tokens may be revised later.
/// audio_range is the actual inference window, or None for the final assembly.
pub struct BatchUpdate<'a> {
    pub kind: BatchUpdateKind,
    pub progress: BatchProgress,
    pub audio_range: Option<Range<u64>>,
    pub text: &'a str,
    pub tokens: &'a [TimedToken],
}
impl BatchUpdate<'_> {
    /// Take an owned snapshot without retaining the callback's borrowed storage.
    pub fn transcript(&self) -> Transcript {
        Transcript {
            text: self.text.to_owned(),
            tokens: self.tokens.to_vec(),
            audio_samples: self.progress.processed_samples,
        }
    }
}

enum BatchText<'a> {
    Ready(&'a str),
    FromTokens,
}
struct BatchState<'a> {
    kind: BatchUpdateKind,
    progress: BatchProgress,
    audio_range: Option<Range<u64>>,
    text: BatchText<'a>,
    tokens: &'a [TimedToken],
}
trait BatchObserver {
    fn progress(&mut self, progress: BatchProgress) -> Result<()>;
    fn transcript(
        &mut self,
        recognizer: &dyn WindowRecognizer,
        state: BatchState<'_>,
    ) -> Result<()>;
}
struct ProgressOnly<F>(F);
impl<F: FnMut(BatchProgress) -> Result<()>> BatchObserver for ProgressOnly<F> {
    fn progress(&mut self, progress: BatchProgress) -> Result<()> {
        (self.0)(progress)
    }
    fn transcript(&mut self, _: &dyn WindowRecognizer, _: BatchState<'_>) -> Result<()> {
        Ok(())
    }
}
struct Detailed<F>(F);
impl<F: FnMut(BatchUpdate<'_>) -> Result<()>> BatchObserver for Detailed<F> {
    fn progress(&mut self, _: BatchProgress) -> Result<()> {
        Ok(())
    }
    fn transcript(
        &mut self,
        recognizer: &dyn WindowRecognizer,
        state: BatchState<'_>,
    ) -> Result<()> {
        let rendered;
        let text = match state.text {
            BatchText::Ready(text) => text,
            BatchText::FromTokens => {
                rendered = recognizer.render(state.tokens)?;
                &rendered
            }
        };
        (self.0)(BatchUpdate {
            kind: state.kind,
            progress: state.progress,
            audio_range: state.audio_range,
            text,
            tokens: state.tokens,
        })
    }
}
#[derive(Serialize)]
pub struct BatchResult {
    pub transcript: Transcript,
    /// Peak buffered raw audio, excluding model tensors and decoder state.
    pub peak_buffered_samples: usize,
    pub elapsed_seconds: f64,
}
/// Shared ASR result plus batch I/O/VAD/callback elapsed time and buffer usage.
#[derive(Serialize)]
pub struct BatchRecognition {
    pub result: AsrResult,
    pub peak_buffered_samples: usize,
    pub elapsed_seconds: f64,
}
impl BatchRecognition {
    fn into_legacy(self) -> BatchResult {
        BatchResult {
            transcript: self.result.into_transcript(),
            peak_buffered_samples: self.peak_buffered_samples,
            elapsed_seconds: self.elapsed_seconds,
        }
    }
}

/// Find contiguous ranges at quiet points without reading the entire source.
/// Search within two seconds of each target using a 100 ms energy window.
/// Targets reserve the search width and half-window so every range fits `limits`.
/// Reference algorithm: third-intelligence/Haas-ASR longform.py.
pub fn split_at_quiet_points<S: AudioSampleSource>(
    mut source: S,
    limits: InputLimits,
    cancellation: &Cancellation,
) -> Result<Vec<Range<u64>>> {
    quiet_ranges(&mut source, limits, cancellation).map(|split| split.ranges)
}

/// Transcribe a canonical WAV using contiguous quiet-point ranges without VAD.
pub fn transcribe_quiet_wav<F>(
    engine: &HaasEngine,
    path: &Path,
    cancellation: &Cancellation,
    progress: F,
) -> Result<BatchResult>
where
    F: FnMut(BatchProgress) -> Result<()>,
{
    transcribe_quiet_wav_result(engine, path, cancellation, progress)
        .map(BatchRecognition::into_legacy)
}

/// Decode a system-supported file and transcribe quiet-point ranges without VAD.
pub fn transcribe_quiet_file<F>(
    engine: &HaasEngine,
    path: &Path,
    cancellation: &Cancellation,
    progress: F,
) -> Result<BatchResult>
where
    F: FnMut(BatchProgress) -> Result<()>,
{
    transcribe_quiet_file_result(engine, path, cancellation, progress)
        .map(BatchRecognition::into_legacy)
}

/// Transcribe a canonical source from its beginning using quiet-point ranges.
pub fn transcribe_quiet_source<S: AudioSampleSource, F>(
    engine: &HaasEngine,
    source: S,
    cancellation: &Cancellation,
    progress: F,
) -> Result<BatchResult>
where
    F: FnMut(BatchProgress) -> Result<()>,
{
    transcribe_quiet_source_result(engine, source, cancellation, progress)
        .map(BatchRecognition::into_legacy)
}

pub fn transcribe_quiet_wav_result<F>(
    engine: &HaasEngine,
    path: &Path,
    cancellation: &Cancellation,
    progress: F,
) -> Result<BatchRecognition>
where
    F: FnMut(BatchProgress) -> Result<()>,
{
    let start = std::time::Instant::now();
    cancellation.check()?;
    let source = WavSource::open(path)?;
    transcribe_quiet_loaded(engine, source, cancellation, progress, start)
}

pub fn transcribe_quiet_file_result<F>(
    engine: &HaasEngine,
    path: &Path,
    cancellation: &Cancellation,
    progress: F,
) -> Result<BatchRecognition>
where
    F: FnMut(BatchProgress) -> Result<()>,
{
    let start = std::time::Instant::now();
    cancellation.check()?;
    let source = AudioSource::open(path, cancellation)?;
    transcribe_quiet_loaded(engine, source, cancellation, progress, start)
}

/// Return shared metadata and absolute token times for quiet-point transcription.
pub fn transcribe_quiet_source_result<S: AudioSampleSource, F>(
    engine: &HaasEngine,
    source: S,
    cancellation: &Cancellation,
    progress: F,
) -> Result<BatchRecognition>
where
    F: FnMut(BatchProgress) -> Result<()>,
{
    transcribe_quiet_loaded(
        engine,
        source,
        cancellation,
        progress,
        std::time::Instant::now(),
    )
}

fn transcribe_quiet_loaded<S: AudioSampleSource, F>(
    engine: &HaasEngine,
    source: S,
    cancellation: &Cancellation,
    progress: F,
    start: std::time::Instant,
) -> Result<BatchRecognition>
where
    F: FnMut(BatchProgress) -> Result<()>,
{
    transcribe_quiet_observed(
        engine,
        source,
        cancellation,
        &mut ProgressOnly(progress),
        start,
    )
}
fn transcribe_quiet_observed<S: AudioSampleSource>(
    engine: &dyn WindowRecognizer,
    source: S,
    cancellation: &Cancellation,
    observer: &mut impl BatchObserver,
    start: std::time::Instant,
) -> Result<BatchRecognition> {
    let recognizer = ProfiledRecognizer::new(engine);
    let batch = QuietBatch {
        recognizer: &recognizer,
        source,
        cancellation,
    }
    .run_observed(observer, start)?;
    complete_batch(engine, &recognizer, batch, cancellation, observer)
}

const QUIET_SEARCH_SAMPLES: u64 = 32000;
const QUIET_WINDOW_SAMPLES: usize = 1600;

/// Contextual vocabulary and detailed updates over the same quiet-point policy.
pub fn transcribe_quiet_source_with_vocabulary<S, F>(
    engine: &HaasEngine,
    source: S,
    vocabulary: std::sync::Arc<crate::vocabulary::CompiledVocabulary>,
    cancellation: &Cancellation,
    updates: F,
) -> Result<BatchRecognition>
where
    S: AudioSampleSource,
    F: FnMut(BatchUpdate<'_>) -> Result<()>,
{
    let recognizer = crate::vocabulary::VocabularyRecognizer::new(engine, vocabulary)?;
    transcribe_quiet_observed(
        &recognizer,
        source,
        cancellation,
        &mut Detailed(updates),
        std::time::Instant::now(),
    )
}
pub fn transcribe_source_with_vocabulary<S, F>(
    engine: &HaasEngine,
    vad: &SileroVad,
    source: S,
    config: AsrConfig,
    vocabulary: std::sync::Arc<crate::vocabulary::CompiledVocabulary>,
    cancellation: &Cancellation,
    updates: F,
) -> Result<BatchRecognition>
where
    S: AudioSampleSource,
    F: FnMut(BatchUpdate<'_>) -> Result<()>,
{
    config.validate_for(engine.input_limits())?;
    let recognizer = crate::vocabulary::VocabularyRecognizer::new(engine, vocabulary)?;
    transcribe_observed(
        &recognizer,
        vad,
        source,
        config,
        cancellation,
        &mut Detailed(updates),
        std::time::Instant::now(),
    )
}

/// Report each recognized window and one final assembled transcript.
/// Callback errors and cancellation terminate the operation explicitly.
pub fn transcribe_quiet_source_with_updates<S, F>(
    engine: &HaasEngine,
    source: S,
    cancellation: &Cancellation,
    updates: F,
) -> Result<BatchRecognition>
where
    S: AudioSampleSource,
    F: FnMut(BatchUpdate<'_>) -> Result<()>,
{
    transcribe_quiet_observed(
        engine,
        source,
        cancellation,
        &mut Detailed(updates),
        std::time::Instant::now(),
    )
}
pub fn transcribe_quiet_wav_with_updates<F>(
    engine: &HaasEngine,
    path: &Path,
    cancellation: &Cancellation,
    updates: F,
) -> Result<BatchRecognition>
where
    F: FnMut(BatchUpdate<'_>) -> Result<()>,
{
    let start = std::time::Instant::now();
    cancellation.check()?;
    transcribe_quiet_observed(
        engine,
        WavSource::open(path)?,
        cancellation,
        &mut Detailed(updates),
        start,
    )
}
pub fn transcribe_quiet_file_with_updates<F>(
    engine: &HaasEngine,
    path: &Path,
    cancellation: &Cancellation,
    updates: F,
) -> Result<BatchRecognition>
where
    F: FnMut(BatchUpdate<'_>) -> Result<()>,
{
    let start = std::time::Instant::now();
    cancellation.check()?;
    transcribe_quiet_observed(
        engine,
        AudioSource::open(path, cancellation)?,
        cancellation,
        &mut Detailed(updates),
        start,
    )
}
pub fn transcribe_source_with_updates<S, F>(
    engine: &HaasEngine,
    vad: &SileroVad,
    source: S,
    config: AsrConfig,
    cancellation: &Cancellation,
    updates: F,
) -> Result<BatchRecognition>
where
    S: AudioSampleSource,
    F: FnMut(BatchUpdate<'_>) -> Result<()>,
{
    config.validate_for(engine.input_limits())?;
    cancellation.check()?;
    transcribe_observed(
        engine,
        vad,
        source,
        config,
        cancellation,
        &mut Detailed(updates),
        std::time::Instant::now(),
    )
}
pub fn transcribe_wav_with_updates<F>(
    engine: &HaasEngine,
    vad: &SileroVad,
    path: &Path,
    config: AsrConfig,
    cancellation: &Cancellation,
    updates: F,
) -> Result<BatchRecognition>
where
    F: FnMut(BatchUpdate<'_>) -> Result<()>,
{
    let start = std::time::Instant::now();
    config.validate_for(engine.input_limits())?;
    cancellation.check()?;
    transcribe_observed(
        engine,
        vad,
        WavSource::open(path)?,
        config,
        cancellation,
        &mut Detailed(updates),
        start,
    )
}
pub fn transcribe_file_with_updates<F>(
    engine: &HaasEngine,
    vad: &SileroVad,
    path: &Path,
    config: AsrConfig,
    cancellation: &Cancellation,
    updates: F,
) -> Result<BatchRecognition>
where
    F: FnMut(BatchUpdate<'_>) -> Result<()>,
{
    let start = std::time::Instant::now();
    config.validate_for(engine.input_limits())?;
    cancellation.check()?;
    transcribe_observed(
        engine,
        vad,
        AudioSource::open(path, cancellation)?,
        config,
        cancellation,
        &mut Detailed(updates),
        start,
    )
}

struct QuietRanges {
    ranges: Vec<Range<u64>>,
    peak_buffered_samples: usize,
}

fn read_complete_window<S: AudioSampleSource>(
    source: &mut S,
    offset: u64,
    count: usize,
    cancellation: &Cancellation,
) -> Result<Vec<f32>> {
    cancellation.check()?;
    let audio = source.read_window(offset, count)?;
    if audio.len() != count || audio.iter().any(|sample| !sample.is_finite()) {
        return Err(Error::Invalid(format!(
            "Invalid quiet-point source window: source={:?}, offset={offset}, requested={count}, received={}, non_finite={}",
            source.origin(),
            audio.len(),
            audio.iter().any(|sample| !sample.is_finite())
        )));
    }
    cancellation.check()?;
    Ok(audio)
}

fn quiet_ranges<S: AudioSampleSource>(
    source: &mut S,
    limits: InputLimits,
    cancellation: &Cancellation,
) -> Result<QuietRanges> {
    cancellation.check()?;
    let maximum = limits.maximum_samples() as u64;
    let reserve = QUIET_SEARCH_SAMPLES + QUIET_WINDOW_SAMPLES as u64 / 2;
    let step = maximum.checked_sub(reserve).filter(|step| *step >= QUIET_WINDOW_SAMPLES as u64)
        .ok_or_else(|| Error::Invalid(format!(
            "Model input too short for quiet-point search: maximum={maximum}, reserve={reserve}, energy_window={QUIET_WINDOW_SAMPLES}"
        )))?;
    let total = source.total_samples();
    let mut ranges = Vec::new();
    let mut start = 0;
    let mut peak = 0;
    while total - start > step {
        cancellation.check()?;
        let target = start + step;
        let low = (start + step / 2).max(target.saturating_sub(QUIET_SEARCH_SAMPLES));
        let high =
            (total - QUIET_WINDOW_SAMPLES as u64).min(target.saturating_add(QUIET_SEARCH_SAMPLES));
        let cut = if high > low {
            let count = (high + QUIET_WINDOW_SAMPLES as u64 - low) as usize;
            let audio = read_complete_window(source, low, count, cancellation)?;
            peak = peak.max(audio.len());
            low + quietest_window(&audio, cancellation)? as u64 + QUIET_WINDOW_SAMPLES as u64 / 2
        } else {
            target
        };
        if cut <= start || cut > total || cut - start > maximum {
            return Err(Error::Invalid(format!(
                "Quiet-point range exceeds model limits: start={start}, end={cut}, total={total}, maximum={maximum}"
            )));
        }
        ranges.push(start..cut);
        start = cut;
    }
    if start < total {
        ranges.push(start..total);
    }
    cancellation.check()?;
    Ok(QuietRanges {
        ranges,
        peak_buffered_samples: peak,
    })
}

fn quietest_window(audio: &[f32], cancellation: &Cancellation) -> Result<usize> {
    let mut energy = Vec::with_capacity(audio.len() + 1);
    energy.push(0.0_f64);
    let mut accumulated = 0.0_f64;
    for (index, value) in audio.iter().enumerate() {
        if index.is_multiple_of(4096) {
            cancellation.check()?;
        }
        accumulated += f64::from(*value) * f64::from(*value);
        energy.push(accumulated);
    }
    let mut best = 0;
    let mut minimum = energy[QUIET_WINDOW_SAMPLES] - energy[0];
    for offset in 1..energy.len() - QUIET_WINDOW_SAMPLES {
        let value = energy[offset + QUIET_WINDOW_SAMPLES] - energy[offset];
        if value < minimum {
            minimum = value;
            best = offset;
        }
    }
    cancellation.check()?;
    Ok(best)
}

struct QuietBatch<'a, S: AudioSampleSource> {
    recognizer: &'a dyn WindowRecognizer,
    source: S,
    cancellation: &'a Cancellation,
}

impl<S: AudioSampleSource> QuietBatch<'_, S> {
    #[cfg(test)]
    fn run<F>(self, progress: F, start: std::time::Instant) -> Result<BatchResult>
    where
        F: FnMut(BatchProgress) -> Result<()>,
    {
        self.run_observed(&mut ProgressOnly(progress), start)
    }
    fn run_observed(
        mut self,
        observer: &mut impl BatchObserver,
        start: std::time::Instant,
    ) -> Result<BatchResult> {
        let total = self.source.total_samples();
        let split = quiet_ranges(
            &mut self.source,
            self.recognizer.input_limits(),
            self.cancellation,
        )?;
        let mut peak = split.peak_buffered_samples;
        let mut text = String::new();
        let mut tokens = Vec::new();
        for range in split.ranges {
            let audio = read_complete_window(
                &mut self.source,
                range.start,
                (range.end - range.start) as usize,
                self.cancellation,
            )?;
            peak = peak.max(audio.len());
            let mut decoded = recognize_window(self.recognizer, &audio, self.cancellation)?;
            text.push_str(self.recognizer.render(&decoded)?.trim());
            for token in &mut decoded {
                token.start_sample += range.start;
            }
            tokens.extend(decoded);
            self.cancellation.check()?;
            let progress = BatchProgress {
                processed_samples: range.end,
                total_samples: total,
            };
            observer.progress(progress)?;
            observer.transcript(
                self.recognizer,
                BatchState {
                    kind: BatchUpdateKind::Partial,
                    progress,
                    audio_range: Some(range),
                    text: BatchText::Ready(&text),
                    tokens: &tokens,
                },
            )?;
            self.cancellation.check()?;
        }
        if total == 0 {
            observer.progress(BatchProgress {
                processed_samples: 0,
                total_samples: 0,
            })?;
        }
        self.cancellation.check()?;
        Ok(BatchResult {
            transcript: Transcript {
                text,
                tokens,
                audio_samples: total,
            },
            peak_buffered_samples: peak,
            elapsed_seconds: start.elapsed().as_secs_f64(),
        })
    }
}

/// Decode files within the model window intact. Longer files use VAD boundaries
/// and independent overlapping windows bounded by the loaded model. Left plus right
/// context sets the overlap; center and partial intervals apply to live sessions.
/// Calls progress after successfully recognized input. Callback errors and
/// cancellation fail the job rather than returning a partial transcript.
pub fn transcribe_wav<F>(
    engine: &HaasEngine,
    vad: &SileroVad,
    path: &Path,
    config: AsrConfig,
    cancellation: &Cancellation,
    progress: F,
) -> Result<BatchResult>
where
    F: FnMut(BatchProgress) -> Result<()>,
{
    transcribe_wav_result(engine, vad, path, config, cancellation, progress)
        .map(BatchRecognition::into_legacy)
}
pub fn transcribe_file<F>(
    engine: &HaasEngine,
    vad: &SileroVad,
    path: &Path,
    config: AsrConfig,
    cancellation: &Cancellation,
    progress: F,
) -> Result<BatchResult>
where
    F: FnMut(BatchProgress) -> Result<()>,
{
    transcribe_file_result(engine, vad, path, config, cancellation, progress)
        .map(BatchRecognition::into_legacy)
}
pub fn transcribe_source<S: AudioSampleSource, F>(
    engine: &HaasEngine,
    vad: &SileroVad,
    source: S,
    config: AsrConfig,
    cancellation: &Cancellation,
    progress: F,
) -> Result<BatchResult>
where
    F: FnMut(BatchProgress) -> Result<()>,
{
    transcribe_source_result(engine, vad, source, config, cancellation, progress)
        .map(BatchRecognition::into_legacy)
}
pub fn transcribe_wav_result<F>(
    engine: &HaasEngine,
    vad: &SileroVad,
    path: &Path,
    config: AsrConfig,
    cancellation: &Cancellation,
    progress: F,
) -> Result<BatchRecognition>
where
    F: FnMut(BatchProgress) -> Result<()>,
{
    let start = std::time::Instant::now();
    config.validate_for(engine.input_limits())?;
    cancellation.check()?;
    let source = WavSource::open(path)?;
    transcribe_loaded(engine, vad, source, config, cancellation, progress, start)
}

/// Convert a system-supported audio file in bounded blocks, then transcribe it.
/// Converted temporary storage is released on success, failure or cancellation.
pub fn transcribe_file_result<F>(
    engine: &HaasEngine,
    vad: &SileroVad,
    path: &Path,
    config: AsrConfig,
    cancellation: &Cancellation,
    progress: F,
) -> Result<BatchRecognition>
where
    F: FnMut(BatchProgress) -> Result<()>,
{
    let start = std::time::Instant::now();
    config.validate_for(engine.input_limits())?;
    cancellation.check()?;
    let source = AudioSource::open(path, cancellation)?;
    transcribe_loaded(engine, vad, source, config, cancellation, progress, start)
}

/// Transcribe a canonical source from its beginning without another conversion pass.
/// Pass a mutable borrow to retain its storage for subsequent jobs.
pub fn transcribe_source_result<S: AudioSampleSource, F>(
    engine: &HaasEngine,
    vad: &SileroVad,
    source: S,
    config: AsrConfig,
    cancellation: &Cancellation,
    progress: F,
) -> Result<BatchRecognition>
where
    F: FnMut(BatchProgress) -> Result<()>,
{
    config.validate_for(engine.input_limits())?;
    cancellation.check()?;
    transcribe_loaded(
        engine,
        vad,
        source,
        config,
        cancellation,
        progress,
        std::time::Instant::now(),
    )
}
fn transcribe_loaded<S: AudioSampleSource, F>(
    engine: &HaasEngine,
    vad: &SileroVad,
    source: S,
    config: AsrConfig,
    cancellation: &Cancellation,
    progress: F,
    start: std::time::Instant,
) -> Result<BatchRecognition>
where
    F: FnMut(BatchProgress) -> Result<()>,
{
    transcribe_observed(
        engine,
        vad,
        source,
        config,
        cancellation,
        &mut ProgressOnly(progress),
        start,
    )
}
fn transcribe_observed<S: AudioSampleSource>(
    engine: &dyn WindowRecognizer,
    vad: &SileroVad,
    source: S,
    config: AsrConfig,
    cancellation: &Cancellation,
    observer: &mut impl BatchObserver,
    start: std::time::Instant,
) -> Result<BatchRecognition> {
    let recognizer = ProfiledRecognizer::new(engine);
    let legacy = if source.total_samples() <= engine.input_limits().maximum_samples() as u64 {
        transcribe_single_window_observed(&recognizer, source, cancellation, observer, start)?
    } else {
        OfflineBatch {
            recognizer: &recognizer,
            source,
            config,
            cancellation,
        }
        .run_observed(vad.detector().as_mut(), observer, start)?
    };
    complete_batch(engine, &recognizer, legacy, cancellation, observer)
}

fn complete_batch(
    engine: &dyn WindowRecognizer,
    recognizer: &ProfiledRecognizer<'_>,
    batch: BatchResult,
    cancellation: &Cancellation,
    observer: &mut impl BatchObserver,
) -> Result<BatchRecognition> {
    let result = AsrResult::build(engine, &batch.transcript, recognizer.metrics())?;
    cancellation.check()?;
    observer.transcript(
        engine,
        BatchState {
            kind: BatchUpdateKind::Final,
            progress: BatchProgress {
                processed_samples: batch.transcript.audio_samples,
                total_samples: batch.transcript.audio_samples,
            },
            audio_range: None,
            text: BatchText::Ready(&batch.transcript.text),
            tokens: &batch.transcript.tokens,
        },
    )?;
    cancellation.check()?;
    Ok(BatchRecognition {
        result,
        peak_buffered_samples: batch.peak_buffered_samples,
        elapsed_seconds: batch.elapsed_seconds,
    })
}

struct OfflineBatch<'a, S: AudioSampleSource> {
    recognizer: &'a dyn WindowRecognizer,
    source: S,
    config: AsrConfig,
    cancellation: &'a Cancellation,
}

impl<S: AudioSampleSource> OfflineBatch<'_, S> {
    #[cfg(test)]
    fn run<F>(
        self,
        detector: &mut dyn VoiceDetector,
        progress: F,
        start: std::time::Instant,
    ) -> Result<BatchResult>
    where
        F: FnMut(BatchProgress) -> Result<()>,
    {
        self.run_observed(detector, &mut ProgressOnly(progress), start)
    }
    fn run_observed(
        mut self,
        detector: &mut dyn VoiceDetector,
        observer: &mut impl BatchObserver,
        start: std::time::Instant,
    ) -> Result<BatchResult> {
        self.config.validate_for(self.recognizer.input_limits())?;
        let total = self.source.total_samples();
        let maximum = self.recognizer.input_limits().maximum_samples() as u64;
        let spans = self.utterances(detector)?;
        let overlap = (self.config.left_samples + self.config.right_samples) as u64;
        let stride = (maximum - overlap) / ENCODER_STEP as u64 * ENCODER_STEP as u64;
        if stride == 0 {
            return Err(Error::Invalid(format!(
                "Offline windows cannot advance: maximum={maximum}, overlap={overlap}"
            )));
        }
        let mut tokens = Vec::new();
        let mut peak = 4096;
        let mut covered = 0;
        let mut head_tokens = BTreeSet::new();
        for &(begin, end) in &spans {
            let mut window_start = begin;
            let mut previous_end = begin;
            loop {
                self.cancellation.check()?;
                let window_end = (window_start + maximum).min(end);
                let audio = self
                    .source
                    .read_window(window_start, (window_end - window_start) as usize)?;
                peak = peak.max(audio.len());
                let mut decoded = recognize_window(self.recognizer, &audio, self.cancellation)?;
                for token in &mut decoded {
                    token.start_sample += window_start;
                }
                head_tokens.extend(
                    decoded
                        .iter()
                        .filter(|t| t.start_sample == window_start)
                        .map(|t| (t.token_id, t.start_sample)),
                );
                let splice = reconcile_windows(&tokens, &decoded, overlap);
                tokens.truncate(splice.retained_tokens);
                tokens.extend(splice.suffix);
                covered = covered.max(window_end);
                self.cancellation.check()?;
                let progress = BatchProgress {
                    processed_samples: covered,
                    total_samples: total,
                };
                observer.progress(progress)?;
                observer.transcript(
                    self.recognizer,
                    BatchState {
                        kind: BatchUpdateKind::Partial,
                        progress,
                        audio_range: Some(window_start..window_end),
                        text: BatchText::FromTokens,
                        tokens: &tokens,
                    },
                )?;
                self.cancellation.check()?;
                if window_end == end {
                    break;
                }
                if window_end <= previous_end {
                    return Err(Error::Invalid(format!(
                        "Offline window failed to advance: start={window_start}, end={window_end}, previous_end={previous_end}"
                    )));
                }
                previous_end = window_end;
                let next = window_start + stride;
                // End-align on the segment's encoder grid without exceeding the model limit.
                let final_start = begin
                    + (end - begin - maximum).div_ceil(ENCODER_STEP as u64) * ENCODER_STEP as u64;
                window_start = next.min(final_start);
            }
        }
        let (tokens, repair_peak) = self.repair_gaps(&tokens, &spans, &head_tokens)?;
        peak = peak.max(repair_peak);
        self.cancellation.check()?;
        let transcript = Transcript {
            text: self.recognizer.render(&tokens)?,
            tokens,
            audio_samples: total,
        };
        if covered != total {
            observer.progress(BatchProgress {
                processed_samples: total,
                total_samples: total,
            })?;
        }
        self.cancellation.check()?;
        Ok(BatchResult {
            transcript,
            peak_buffered_samples: peak,
            elapsed_seconds: start.elapsed().as_secs_f64(),
        })
    }

    /// Probe interior gaps only. Opening/tail recovery is model-specific and is not inferred here.
    fn repair_gaps(
        &mut self,
        initial: &[TimedToken],
        utterances: &[(u64, u64)],
        head_tokens: &BTreeSet<(usize, u64)>,
    ) -> Result<(Vec<TimedToken>, usize)> {
        if initial.len() < 2 {
            return Ok((initial.to_vec(), 0));
        }
        let threshold = self.speech_threshold()?;
        let budget =
            32usize.max((self.source.total_samples() * 12 / (SAMPLE_RATE as u64 * 60)) as usize);
        let mut probed = BTreeSet::new();
        let mut working = initial.to_vec();
        let mut peak = 0;
        for _ in 0..3 {
            let mut revised = Vec::with_capacity(working.len());
            let mut recovered = 0;
            for pair in working.windows(2) {
                self.cancellation.check()?;
                let left = &pair[0];
                let right = &pair[1];
                revised.push(left.clone());
                if !reliable_gap_anchors(left, right, utterances, head_tokens) {
                    continue;
                }
                let start = left.start_sample + left.duration_samples.max(ENCODER_STEP as u64);
                let end = right.start_sample;
                if end.saturating_sub(start) < 24000
                    || probed.len() >= budget
                    || probed.contains(&start)
                {
                    continue;
                }
                if !self.gap_has_speech(start, end, threshold)? {
                    continue;
                }
                probed.insert(start);
                let (candidate, probe_peak) = self.recover_gap(left, right, start, end)?;
                peak = peak.max(probe_peak);
                recovered += candidate.len();
                revised.extend(candidate);
            }
            revised.push(working.last().unwrap().clone());
            working = revised;
            if recovered == 0 {
                break;
            }
        }
        Ok((working, peak))
    }

    fn speech_threshold(&mut self) -> Result<f64> {
        let mut rms = Vec::new();
        let total = self.source.total_samples();
        let frame = ENCODER_STEP as u64;
        let mut start = 0;
        while start + frame <= total {
            self.cancellation.check()?;
            let audio = self.source.read_window(start, ENCODER_STEP)?;
            let level = frame_rms(&audio);
            if level > 0.0 {
                rms.push(level);
            }
            start += frame;
        }
        if rms.is_empty() {
            return Ok(0.008);
        }
        rms.sort_by(f64::total_cmp);
        Ok((rms[rms.len() * 3 / 4] * 0.3).clamp(0.0005, 0.008))
    }

    fn gap_has_speech(&mut self, begin: u64, end: u64, threshold: f64) -> Result<bool> {
        let mut start = begin;
        let mut speech_samples = 0;
        while start + ENCODER_STEP as u64 <= end {
            self.cancellation.check()?;
            let audio = self.source.read_window(start, ENCODER_STEP)?;
            if frame_rms(&audio) > threshold {
                speech_samples += ENCODER_STEP;
            }
            if speech_samples >= SAMPLE_RATE as usize / 2 {
                return Ok(true);
            }
            start += ENCODER_STEP as u64;
        }
        Ok(false)
    }

    fn recover_gap(
        &mut self,
        left: &TimedToken,
        right: &TimedToken,
        begin: u64,
        end: u64,
    ) -> Result<(Vec<TimedToken>, usize)> {
        let total = self.source.total_samples();
        let pad = 12800;
        let maximum = self.recognizer.input_limits().maximum_samples() as u64;
        let placements = [
            (begin, maximum),
            (
                (begin + (end - begin) / 2).saturating_sub(maximum / 2),
                maximum,
            ),
            (begin.saturating_sub(pad), end - begin + pad * 2),
        ];
        let mut visited = BTreeSet::new();
        let mut peak = 0;
        for (position, length) in placements {
            self.cancellation.check()?;
            let count = length.min(maximum).min(total);
            let start = position.min(total - count) / ENCODER_STEP as u64 * ENCODER_STEP as u64;
            if !visited.insert((start, count)) {
                continue;
            }
            let audio = self.source.read_window(start, count as usize)?;
            peak = peak.max(audio.len());
            let mut decoded = recognize_window(self.recognizer, &audio, self.cancellation)?;
            for token in &mut decoded {
                token.start_sample += start;
            }
            let candidate = gap_candidate(&decoded, left, right, begin, end);
            let mut confidence: Vec<_> = candidate.iter().map(|t| t.confidence).collect();
            confidence.sort_by(f32::total_cmp);
            if !confidence.is_empty()
                && confidence[confidence.len() / 2] >= 0.7
                && self
                    .recognizer
                    .render(&candidate)?
                    .chars()
                    .any(char::is_alphanumeric)
            {
                return Ok((candidate, peak));
            }
        }
        Ok((Vec::new(), peak))
    }

    fn utterances(&mut self, detector: &mut dyn VoiceDetector) -> Result<Vec<(u64, u64)>> {
        let mut spans = Vec::new();
        let mut received = 0;
        let mut begin = 0;
        let mut has_audio = false;
        let mut boundaries = SpeechBoundaries::new(self.config.vad_min_silence_samples);
        loop {
            self.cancellation.check()?;
            if received == self.source.total_samples() {
                break;
            }
            let count = (self.source.total_samples() - received).min(4096) as usize;
            let audio = self.source.read_window(received, count)?;
            if audio.len() != count {
                return Err(Error::Invalid(format!(
                    "Audio source returned incomplete batch input: source={:?}, offset={received}, requested={count}, received={}",
                    self.source.origin(),
                    audio.len()
                )));
            }
            if audio.iter().any(|v| !v.is_finite()) {
                return Err(Error::Invalid(format!(
                    "Non-finite offline audio: source={:?}, start={received}",
                    self.source.origin()
                )));
            }
            received += audio.len() as u64;
            let probability = detector.probability(&audio)?;
            let silent = audio.iter().all(|v| *v == 0.0);
            if silent && !has_audio {
                begin = received;
                continue;
            }
            has_audio = true;
            if let Some(boundary) = boundaries.observe(probability, silent) {
                spans.push((begin, received));
                begin = if matches!(boundary, SpeechBoundary::DigitalSilence) {
                    received
                } else {
                    received
                        .saturating_sub(self.config.left_samples as u64)
                        .max(begin)
                };
                has_audio = false;
            }
        }
        if received != self.source.total_samples() {
            return Err(Error::Invalid(format!(
                "WAV ended before declared length: source={:?}, received={received}, declared={}",
                self.source.origin(),
                self.source.total_samples()
            )));
        }
        if has_audio {
            spans.push((begin, received));
        }
        Ok(spans)
    }
}

/// Haas can emit a spoken prefix at frame zero before its acoustic position.
/// Such clocks and separate utterances cannot anchor an interior missing-speech probe.
fn reliable_gap_anchors(
    left: &TimedToken,
    right: &TimedToken,
    utterances: &[(u64, u64)],
    heads: &BTreeSet<(usize, u64)>,
) -> bool {
    !heads.contains(&(left.token_id, left.start_sample))
        && !heads.contains(&(right.token_id, right.start_sample))
        && utterances
            .iter()
            .any(|&(begin, end)| begin <= left.start_sample && right.start_sample < end)
}

fn frame_rms(audio: &[f32]) -> f64 {
    (audio.iter().map(|v| (*v as f64).powi(2)).sum::<f64>() / audio.len() as f64).sqrt()
}

/// Reject re-heard neighbours while retaining genuine repetitions beyond the edge tolerance.
fn gap_candidate(
    tokens: &[TimedToken],
    left: &TimedToken,
    right: &TimedToken,
    begin: u64,
    end: u64,
) -> Vec<TimedToken> {
    let mut candidate: Vec<_> = tokens
        .iter()
        .filter(|t| {
            t.start_sample > begin && t.start_sample < end.saturating_sub(ENCODER_STEP as u64)
        })
        .cloned()
        .collect();
    let echoes = |token: &TimedToken, neighbour: &TimedToken| {
        let distance = token.start_sample.abs_diff(neighbour.start_sample);
        distance <= 2 * ENCODER_STEP as u64
            || token.token_id == neighbour.token_id && distance <= 5 * ENCODER_STEP as u64
    };
    let first = candidate
        .iter()
        .position(|t| !echoes(t, left))
        .unwrap_or(candidate.len());
    candidate.drain(..first);
    while candidate.last().is_some_and(|t| echoes(t, right)) {
        candidate.pop();
    }
    candidate
}

struct TokenSplice {
    retained_tokens: usize,
    suffix: Vec<TimedToken>,
}

/// Reconcile independent hypotheses, retaining the longer token run between shared anchors.
/// Japanese continuation pieces are spliceable without space or word-initial guards.
fn reconcile_windows(left: &[TimedToken], right: &[TimedToken], overlap: u64) -> TokenSplice {
    let Some(last) = left.last() else {
        return TokenSplice {
            retained_tokens: 0,
            suffix: right.to_vec(),
        };
    };
    let Some(first) = right.first() else {
        return TokenSplice {
            retained_tokens: left.len(),
            suffix: Vec::new(),
        };
    };
    let left_end = last.start_sample + ENCODER_STEP as u64;
    if left_end <= first.start_sample {
        return TokenSplice {
            retained_tokens: left.len(),
            suffix: right.to_vec(),
        };
    }
    let li = left.partition_point(|t| {
        t.start_sample + ENCODER_STEP as u64 <= first.start_sample.saturating_sub(overlap)
    });
    let rn = right.partition_point(|t| t.start_sample < left_end.saturating_add(overlap));
    let a = &left[li..];
    let b = &right[..rn];
    // Prefer acoustically aligned repetitions over a longer, uniformly shifted match.
    let matches: Vec<_> = align_tokens(a, b, overlap / 2, |distance| {
        (overlap / 2 + 1 - distance, -(distance as i64))
    })
    .into_iter()
    .map(|(i, j)| (li + i, j))
    .collect();
    let mut result = Vec::new();
    let retained_tokens;
    if let Some(&(first_left, _)) = matches.first() {
        retained_tokens = first_left;
        for (index, &(l, r)) in matches.iter().enumerate() {
            result.push(left[l].clone());
            if let Some(&(next_l, next_r)) = matches.get(index + 1) {
                let lg = &left[l + 1..next_l];
                let rg = &right[r + 1..next_r];
                result.extend_from_slice(if rg.len() > lg.len() { rg } else { lg });
            }
        }
        result.extend_from_slice(&right[matches.last().unwrap().1 + 1..]);
    } else {
        let cutoff = first.start_sample + (left_end - first.start_sample) / 2;
        retained_tokens = left.partition_point(|t| t.start_sample < cutoff);
        result.extend_from_slice(&right[right.partition_point(|t| t.start_sample < cutoff)..]);
    }
    // Preserve decoder token order; timing jitter must never reorder Japanese pieces.
    let mut previous = left[..retained_tokens]
        .last()
        .map(|t| t.start_sample)
        .unwrap_or(0);
    for token in &mut result {
        token.start_sample = token.start_sample.max(previous);
        previous = token.start_sample;
    }
    TokenSplice {
        retained_tokens,
        suffix: result,
    }
}

/// Keep audio that fits the model window intact; only longer files need segmentation.
#[cfg(test)]
fn transcribe_single_window<S: AudioSampleSource, F>(
    recognizer: &dyn WindowRecognizer,
    source: S,
    cancellation: &Cancellation,
    progress: F,
    start: std::time::Instant,
) -> Result<BatchResult>
where
    F: FnMut(BatchProgress) -> Result<()>,
{
    transcribe_single_window_observed(
        recognizer,
        source,
        cancellation,
        &mut ProgressOnly(progress),
        start,
    )
}
fn transcribe_single_window_observed<S: AudioSampleSource>(
    recognizer: &dyn WindowRecognizer,
    source: S,
    cancellation: &Cancellation,
    observer: &mut impl BatchObserver,
    start: std::time::Instant,
) -> Result<BatchResult> {
    let mut reader = source;
    let total_samples = reader.total_samples();
    let maximum = recognizer.input_limits().maximum_samples();
    if total_samples > maximum as u64 {
        return Err(Error::Invalid(format!(
            "Single-window file exceeds model input: samples={total_samples}, maximum={maximum}"
        )));
    }
    let audio = if total_samples == 0 {
        Vec::new()
    } else {
        reader.read_window(0, total_samples as usize)?
    };
    if audio.len() as u64 != total_samples {
        return Err(Error::Invalid(format!(
            "WAV ended before declared length: received={}, declared={total_samples}",
            audio.len()
        )));
    }
    let tokens = recognize_window(recognizer, &audio, cancellation)?;
    let transcript = Transcript {
        text: recognizer.render(&tokens)?,
        tokens,
        audio_samples: total_samples,
    };
    cancellation.check()?;
    let progress = BatchProgress {
        processed_samples: total_samples,
        total_samples,
    };
    observer.progress(progress)?;
    if total_samples > 0 {
        observer.transcript(
            recognizer,
            BatchState {
                kind: BatchUpdateKind::Partial,
                progress,
                audio_range: Some(0..total_samples),
                text: BatchText::Ready(&transcript.text),
                tokens: &transcript.tokens,
            },
        )?;
    }
    cancellation.check()?;
    Ok(BatchResult {
        transcript,
        peak_buffered_samples: audio.len(),
        elapsed_seconds: start.elapsed().as_secs_f64(),
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::recording::MAX_SAMPLES;
    use crate::window::test_support::{FULL_WINDOW, Markers, SHORT_WINDOW};
    use std::path::PathBuf;

    #[test]
    fn detailed_quiet_callbacks_report_real_ranges_and_borrowed_transcript_prefixes() {
        let recognizer = Markers::new();
        let cancellation = Cancellation::new();
        let audio: Vec<_> = (0..640000)
            .map(|index| if index % 16000 == 0 { 7.0 } else { 0.1 })
            .collect();
        let mut snapshots = Vec::new();
        let mut ranges = Vec::new();
        let mut observer = Detailed(|update: BatchUpdate<'_>| {
            assert_eq!(update.kind, BatchUpdateKind::Partial);
            snapshots.push(update.transcript());
            ranges.push(update.audio_range.unwrap());
            Ok(())
        });
        let result = QuietBatch {
            recognizer: &recognizer,
            source: AudioSource::from_samples(audio).unwrap(),
            cancellation: &cancellation,
        }
        .run_observed(&mut observer, std::time::Instant::now())
        .unwrap();
        assert!(snapshots.len() > 1);
        assert_eq!(ranges[0].start, 0);
        assert_eq!(ranges.last().unwrap().end, 640000);
        assert!(ranges.windows(2).all(|pair| pair[0].end == pair[1].start));
        assert_eq!(snapshots.last().unwrap().text, result.transcript.text);
        assert_eq!(snapshots.last().unwrap().tokens, result.transcript.tokens);
        assert!(
            snapshots
                .windows(2)
                .all(|pair| pair[1].tokens.starts_with(&pair[0].tokens))
        );
    }

    #[test]
    fn detailed_callback_failure_and_cancellation_stop_before_the_next_window() {
        let recognizer = Markers::new();
        let cancellation = Cancellation::new();
        let mut observer = Detailed(|_: BatchUpdate<'_>| {
            Err(Error::Invalid("callback rejected transcript".into()))
        });
        let error = QuietBatch {
            recognizer: &recognizer,
            source: AudioSource::from_samples(vec![0.1; 640000]).unwrap(),
            cancellation: &cancellation,
        }
        .run_observed(&mut observer, std::time::Instant::now());
        assert!(
            matches!(error, Err(Error::Invalid(message)) if message=="callback rejected transcript")
        );
        assert_eq!(recognizer.calls.get(), 1);
        let recognizer = Markers::new();
        let cancellation = Cancellation::new();
        let mut observer = Detailed(|_: BatchUpdate<'_>| {
            cancellation.cancel();
            Ok(())
        });
        assert!(matches!(
            QuietBatch {
                recognizer: &recognizer,
                source: AudioSource::from_samples(vec![0.1; 640000]).unwrap(),
                cancellation: &cancellation
            }
            .run_observed(&mut observer, std::time::Instant::now()),
            Err(Error::Cancelled)
        ));
        assert_eq!(recognizer.calls.get(), 1);
    }

    #[test]
    fn quiet_ranges_cover_audio_and_respect_both_model_limits() {
        for limits in [FULL_WINDOW, SHORT_WINDOW] {
            for samples in [0, 1, 10240, 240000, 491520, 640000, 1528000] {
                let source = AudioSource::from_samples(vec![0.0; samples]).unwrap();
                let ranges = split_at_quiet_points(source, limits, &Cancellation::new()).unwrap();
                if samples == 0 {
                    assert!(ranges.is_empty());
                } else {
                    assert_eq!(ranges.first().unwrap().start, 0);
                    assert_eq!(ranges.last().unwrap().end, samples as u64);
                    assert!(ranges.windows(2).all(|pair| pair[0].end == pair[1].start));
                    assert!(ranges.iter().all(|range| range.start < range.end
                        && range.end - range.start <= limits.maximum_samples() as u64));
                }
            }
        }
    }

    #[test]
    fn quiet_search_keeps_the_last_candidate_inside_the_model_limit() {
        for limits in [FULL_WINDOW, SHORT_WINDOW] {
            let audio: Vec<f32> = (0..640000)
                .map(|i| 0.5 - 0.45 * i as f32 / 639999.0)
                .collect();
            let source = AudioSource::from_samples(audio).unwrap();
            let ranges = split_at_quiet_points(source, limits, &Cancellation::new()).unwrap();
            assert_eq!(ranges[0], 0..limits.maximum_samples() as u64);
        }
    }

    #[test]
    fn quiet_search_uses_the_center_of_the_first_lowest_energy_window() {
        for (limits, silence_start, expected) in [
            (SHORT_WINDOW, 208000, 208800),
            (FULL_WINDOW, 456000, 456800),
        ] {
            let mut audio = vec![0.1; 720000];
            audio[silence_start..silence_start + 8000].fill(0.0);
            let source = AudioSource::from_samples(audio).unwrap();
            let ranges = split_at_quiet_points(source, limits, &Cancellation::new()).unwrap();
            assert_eq!(ranges[0].end, expected);
        }
    }

    #[test]
    fn quiet_batch_preserves_all_tokens_and_absolute_times_without_overlap() {
        for limits in [FULL_WINDOW, SHORT_WINDOW] {
            let cancellation = Cancellation::new();
            let length = 640000;
            let ranges = split_at_quiet_points(
                AudioSource::from_samples(vec![0.1; length]).unwrap(),
                limits,
                &cancellation,
            )
            .unwrap();
            let mut positions = BTreeSet::from([0, length - 1]);
            for range in &ranges[..ranges.len() - 1] {
                for position in [range.end - 1, range.end, range.end + 1] {
                    positions.insert(position as usize);
                }
            }
            let mut audio = vec![0.1; length];
            for &position in &positions {
                audio[position] = 7.0;
            }
            let expected: Vec<_> = positions.iter().map(|&p| p as u64).collect();
            let recognizer = Markers::with_limits(limits);
            let mut updates = Vec::new();
            let result = QuietBatch {
                recognizer: &recognizer,
                source: AudioSource::from_samples(audio).unwrap(),
                cancellation: &cancellation,
            }
            .run(
                |progress| {
                    updates.push(progress);
                    Ok(())
                },
                std::time::Instant::now(),
            )
            .unwrap();
            assert_eq!(
                result
                    .transcript
                    .tokens
                    .iter()
                    .map(|t| t.start_sample)
                    .collect::<Vec<_>>(),
                expected
            );
            assert!(result.peak_buffered_samples <= limits.maximum_samples());
            assert_eq!(result.transcript.audio_samples, length as u64);
            assert_eq!(updates.len(), recognizer.calls.get());
            assert!(
                updates
                    .windows(2)
                    .all(|pair| pair[0].processed_samples < pair[1].processed_samples)
            );
            assert_eq!(updates.last().unwrap().processed_samples, length as u64);
        }
    }

    #[test]
    fn quiet_batch_pads_short_audio_and_counts_only_original_coverage() {
        let recognizer = Markers::new();
        let profiled = ProfiledRecognizer::new(&recognizer);
        let cancellation = Cancellation::new();
        let result = QuietBatch {
            recognizer: &profiled,
            source: AudioSource::from_samples(vec![7.0]).unwrap(),
            cancellation: &cancellation,
        }
        .run(|_| Ok(()), std::time::Instant::now())
        .unwrap();
        assert_eq!(result.transcript.audio_samples, 1);
        assert_eq!(result.transcript.tokens.len(), 1);
        assert_eq!(profiled.metrics().calls, 1);
        assert_eq!(profiled.metrics().audio_samples, 1);
        assert_eq!(profiled.metrics().padded_audio_samples, 10240);
    }

    #[test]
    fn quiet_batch_propagates_progress_errors_and_final_callback_cancellation() {
        for length in [0, 1, 640000] {
            let recognizer = Markers::new();
            let cancellation = Cancellation::new();
            let job = QuietBatch {
                recognizer: &recognizer,
                source: AudioSource::from_samples(vec![0.1; length]).unwrap(),
                cancellation: &cancellation,
            };
            let result = job.run(
                |_| {
                    cancellation.cancel();
                    Ok(())
                },
                std::time::Instant::now(),
            );
            assert!(matches!(result, Err(Error::Cancelled)));
            assert!(recognizer.calls.get() <= 1);
            let cancellation = Cancellation::new();
            let job = QuietBatch {
                recognizer: &recognizer,
                source: AudioSource::from_samples(vec![0.1; length]).unwrap(),
                cancellation: &cancellation,
            };
            assert!(
                matches!(job.run(|_| Err(Error::Invalid("progress failed".into())), std::time::Instant::now()), Err(Error::Invalid(message)) if message == "progress failed")
            );
        }
    }

    struct IncompleteSource;
    impl AudioSampleSource for IncompleteSource {
        fn origin(&self) -> Option<&Path> {
            None
        }
        fn total_samples(&self) -> u64 {
            640000
        }
        fn read_window(&mut self, _: u64, count: usize) -> Result<Vec<f32>> {
            Ok(vec![0.0; count - 1])
        }
        fn read_block(&mut self, _: usize) -> Result<Vec<f32>> {
            Err(Error::Invalid("Sequential reads are not supported".into()))
        }
    }

    #[test]
    fn quiet_ranges_reject_incomplete_sources_and_cancel_before_reading() {
        let cancellation = Cancellation::new();
        assert!(
            matches!(split_at_quiet_points(IncompleteSource, FULL_WINDOW, &cancellation), Err(Error::Invalid(message)) if message.contains("requested=") && message.contains("received=") && message.contains("offset="))
        );
        cancellation.cancel();
        assert!(matches!(
            split_at_quiet_points(IncompleteSource, FULL_WINDOW, &cancellation),
            Err(Error::Cancelled)
        ));
        assert!(
            split_at_quiet_points(
                IncompleteSource,
                InputLimits::new(1, 10240).unwrap(),
                &Cancellation::new()
            )
            .is_err()
        );
    }

    struct RevisedWindow;
    impl WindowRecognizer for RevisedWindow {
        fn input_limits(&self) -> crate::InputLimits {
            FULL_WINDOW
        }
        fn infer(&self, audio: &[f32], c: &Cancellation) -> Result<Vec<TimedToken>> {
            c.check()?;
            let start = audio[0] as u64;
            let full = [
                (1, 400000),
                (2, 440000),
                (3, 450000),
                (4, 460000),
                (5, 480000),
                (6, 590000),
            ];
            Ok(full
                .into_iter()
                .filter(|(id, t)| {
                    *t >= start
                        && *t < start + audio.len() as u64
                        && !(start == 0 && [2, 3].contains(id))
                })
                .map(|(token_id, t)| TimedToken {
                    token_id,
                    start_sample: t - start,
                    confidence: 0.95,
                    duration_samples: ENCODER_STEP as u64,
                })
                .collect())
        }
        fn render(&self, tokens: &[TimedToken]) -> Result<String> {
            Ok(tokens
                .iter()
                .map(|t| t.token_id.to_string())
                .collect::<Vec<_>>()
                .join(" "))
        }
    }
    struct GapSensitive {
        posterior: f32,
        limits: crate::InputLimits,
    }
    impl WindowRecognizer for GapSensitive {
        fn input_limits(&self) -> crate::InputLimits {
            self.limits
        }
        fn infer(&self, audio: &[f32], c: &Cancellation) -> Result<Vec<TimedToken>> {
            c.check()?;
            let start = audio[0] as u64;
            Ok([(1, 160000), (2, 200000), (3, 240000), (4, 500000)]
                .into_iter()
                .filter(|(id, time)| {
                    *time >= start
                        && *time < start + audio.len() as u64
                        && !(*id == 2 && audio.len() >= self.limits.maximum_samples() / 2)
                })
                .map(|(token_id, time)| TimedToken {
                    token_id,
                    start_sample: time - start,
                    confidence: self.posterior,
                    duration_samples: ENCODER_STEP as u64,
                })
                .collect())
        }
        fn render(&self, tokens: &[TimedToken]) -> Result<String> {
            Ok(tokens
                .iter()
                .map(|t| t.token_id.to_string())
                .collect::<Vec<_>>()
                .join(" "))
        }
    }
    #[test]
    fn seam_repair_recovers_only_confident_in_gap_speech() {
        let audio: Vec<_> = (0..600000).map(|i| i as f32).collect();
        let path = fixture(&audio);
        for limits in [FULL_WINDOW, SHORT_WINDOW] {
            for (posterior, expected) in [(0.95, "1 2 3 4"), (0.1, "1 3 4")] {
                let recognizer = GapSensitive { posterior, limits };
                let cancellation = Cancellation::new();
                let result = OfflineBatch {
                    recognizer: &recognizer,
                    source: WavSource::open(&path).unwrap(),
                    config: AsrConfig::interactive_for(limits).unwrap(),
                    cancellation: &cancellation,
                }
                .run(&mut QuietDetector, |_| Ok(()), std::time::Instant::now())
                .unwrap();
                assert_eq!(result.transcript.text, expected);
            }
        }
        std::fs::remove_file(path).unwrap();
    }

    #[test]
    fn gap_candidates_reject_neighbour_echoes_but_keep_real_repetitions() {
        let t = |id, frame| TimedToken {
            token_id: id,
            start_sample: frame * ENCODER_STEP as u64,
            confidence: 0.95,
            duration_samples: ENCODER_STEP as u64,
        };
        let left = t(1, 0);
        let right = t(9, 40);
        let tokens = [t(1, 3), t(1, 7), t(2, 20), t(9, 36), t(9, 39)];
        let result = gap_candidate(
            &tokens,
            &left,
            &right,
            ENCODER_STEP as u64,
            40 * ENCODER_STEP as u64,
        );
        assert_eq!(
            result.iter().map(|t| t.token_id).collect::<Vec<_>>(),
            [1, 2]
        );
    }

    #[test]
    fn gap_repair_never_anchors_on_window_zero_or_crosses_utterances() {
        let token = |time| TimedToken {
            token_id: 1,
            start_sample: time,
            confidence: 0.99,
            duration_samples: ENCODER_STEP as u64,
        };
        let utterances = [(0, 160000), (200000, 400000)];
        let heads = BTreeSet::from([(1, 200000)]);
        assert!(!reliable_gap_anchors(
            &token(150000),
            &token(210000),
            &utterances,
            &heads
        ));
        assert!(!reliable_gap_anchors(
            &token(200000),
            &token(240000),
            &utterances,
            &heads
        ));
        assert!(reliable_gap_anchors(
            &token(220000),
            &token(260000),
            &utterances,
            &heads
        ));
    }

    struct QuietDetector;
    impl VoiceDetector for QuietDetector {
        fn probability(&mut self, _: &[f32]) -> Result<f32> {
            Ok(0.0)
        }
    }
    fn fixture(audio: &[f32]) -> PathBuf {
        let path = std::env::temp_dir().join(format!(
            "mozu-batch-{}-{}.wav",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        let mut writer = hound::WavWriter::create(
            &path,
            hound::WavSpec {
                channels: 1,
                sample_rate: SAMPLE_RATE,
                bits_per_sample: 32,
                sample_format: hound::SampleFormat::Float,
            },
        )
        .unwrap();
        for sample in audio {
            writer.write_sample(*sample).unwrap();
        }
        writer.finalize().unwrap();
        path
    }

    #[test]
    fn offline_merge_can_recover_tokens_before_a_previously_decoded_seam() {
        // A later hypothesis recovers a phrase between shared anchors, including unspaced pieces.
        let audio: Vec<_> = (0..600000).map(|i| i as f32).collect();
        let path = fixture(&audio);
        let recognizer = RevisedWindow;
        let cancellation = Cancellation::new();
        let mut updates = Vec::new();
        let result = OfflineBatch {
            recognizer: &recognizer,
            source: WavSource::open(&path).unwrap(),
            config: AsrConfig::interactive(),
            cancellation: &cancellation,
        }
        .run(
            &mut QuietDetector,
            |p| {
                updates.push(p.processed_samples);
                Ok(())
            },
            std::time::Instant::now(),
        );
        std::fs::remove_file(path).unwrap();
        let result = result.unwrap();
        assert_eq!(result.transcript.text, "1 2 3 4 5 6");
        assert_eq!(result.transcript.audio_samples, 600000);
        assert!(result.peak_buffered_samples <= MAX_SAMPLES);
        assert_eq!(updates, [MAX_SAMPLES as u64, 600000]);
    }

    fn merge_fixture(left: &[TimedToken], right: &[TimedToken], overlap: u64) -> Vec<TimedToken> {
        let splice = reconcile_windows(left, right, overlap);
        left[..splice.retained_tokens]
            .iter()
            .chain(&splice.suffix)
            .cloned()
            .collect()
    }

    #[test]
    fn japanese_continuations_and_real_repetitions_survive_merging() {
        let t = |token_id, frame| TimedToken {
            token_id,
            start_sample: frame * ENCODER_STEP as u64,
            confidence: 0.95,
            duration_samples: ENCODER_STEP as u64,
        };
        let left = [t(1, 100), t(2, 101), t(3, 102)];
        let right = [
            t(2, 101),
            t(3, 102),
            t(4, 103),
            t(5, 104),
            t(6, 105),
            t(6, 106),
        ];
        let merged = merge_fixture(&left, &right, 32000);
        assert_eq!(
            merged.iter().map(|t| t.token_id).collect::<Vec<_>>(),
            [1, 2, 3, 4, 5, 6, 6]
        );
        let unmatched = merge_fixture(&left[..2], &right[1..], 32000);
        assert!(unmatched.iter().any(|t| t.token_id == 5));
        assert_eq!(
            merged.last().unwrap().start_sample,
            106 * ENCODER_STEP as u64
        );
    }

    #[test]
    fn long_windows_preserve_boundary_markers_and_end_align_the_tail() {
        for limits in [FULL_WINDOW, SHORT_WINDOW] {
            let maximum = limits.maximum_samples();
            for len in [maximum + 1, maximum * 2 + 123, maximum * 3] {
                let mut audio = vec![0.0001; len];
                let mut positions = vec![0, maximum - 1, maximum, len - 1];
                positions.extend((0..len).step_by(12031));
                positions.sort_unstable();
                positions.dedup();
                for &position in &positions {
                    audio[position] = 7.0;
                }
                let path = fixture(&audio);
                let recognizer = Markers::with_limits(limits);
                let cancellation = Cancellation::new();
                let result = OfflineBatch {
                    recognizer: &recognizer,
                    source: WavSource::open(&path).unwrap(),
                    config: AsrConfig::interactive_for(limits).unwrap(),
                    cancellation: &cancellation,
                }
                .run(&mut QuietDetector, |_| Ok(()), std::time::Instant::now());
                std::fs::remove_file(path).unwrap();
                let result = result.unwrap();
                assert_eq!(
                    result
                        .transcript
                        .tokens
                        .iter()
                        .map(|t| t.start_sample as usize)
                        .collect::<Vec<_>>(),
                    positions,
                    "length={len}"
                );
                assert_eq!(result.transcript.audio_samples, len as u64);
                assert!(result.peak_buffered_samples <= maximum);
            }
        }
    }

    #[test]
    fn digital_silence_never_decodes_and_quiet_speech_is_preserved() {
        let silence = vec![0.0; MAX_SAMPLES + 65536];
        let path = fixture(&silence);
        let recognizer = Markers::new();
        let cancellation = Cancellation::new();
        let result = OfflineBatch {
            recognizer: &recognizer,
            source: WavSource::open(&path).unwrap(),
            config: AsrConfig::interactive(),
            cancellation: &cancellation,
        }
        .run(&mut QuietDetector, |_| Ok(()), std::time::Instant::now());
        std::fs::remove_file(path).unwrap();
        assert!(result.unwrap().transcript.tokens.is_empty());
        assert_eq!(recognizer.calls.get(), 0);
        let mut audio = silence;
        audio[10000..20000].fill(0.0001);
        audio[14000] = 7.0;
        audio[300000..310000].fill(0.0001);
        audio[304000] = 7.0;
        let path = fixture(&audio);
        let result = OfflineBatch {
            recognizer: &recognizer,
            source: WavSource::open(&path).unwrap(),
            config: AsrConfig::interactive(),
            cancellation: &cancellation,
        }
        .run(&mut QuietDetector, |_| Ok(()), std::time::Instant::now());
        std::fs::remove_file(path).unwrap();
        assert_eq!(
            result
                .unwrap()
                .transcript
                .tokens
                .iter()
                .map(|t| t.start_sample)
                .collect::<Vec<_>>(),
            [14000, 304000]
        );
    }

    #[test]
    fn cancelling_from_the_final_callback_cancels_silent_and_short_jobs() {
        for len in [1, MAX_SAMPLES + 1] {
            let path = fixture(&vec![0.0; len]);
            let recognizer = Markers::new();
            let cancellation = Cancellation::new();
            let source = WavSource::open(&path).unwrap();
            let progress = |_| {
                cancellation.cancel();
                Ok(())
            };
            let result = if len <= MAX_SAMPLES {
                transcribe_single_window(
                    &recognizer,
                    source,
                    &cancellation,
                    progress,
                    std::time::Instant::now(),
                )
            } else {
                OfflineBatch {
                    recognizer: &recognizer,
                    source,
                    config: AsrConfig::interactive(),
                    cancellation: &cancellation,
                }
                .run(&mut QuietDetector, progress, std::time::Instant::now())
            };
            std::fs::remove_file(path).unwrap();
            assert!(matches!(result, Err(Error::Cancelled)));
        }
    }

    #[test]
    fn callback_error_and_cancellation_never_return_success() {
        let path = fixture(&vec![0.0001; MAX_SAMPLES + 1]);
        let cancellation = Cancellation::new();
        let job = || OfflineBatch {
            recognizer: &RevisedWindow,
            source: WavSource::open(&path).unwrap(),
            config: AsrConfig::interactive(),
            cancellation: &cancellation,
        };
        let result = job().run(
            &mut QuietDetector,
            |_| Err(Error::Invalid("callback failed".into())),
            std::time::Instant::now(),
        );
        assert!(matches!(result, Err(Error::Invalid(message)) if message == "callback failed"));
        cancellation.cancel();
        assert!(matches!(
            job().run(&mut QuietDetector, |_| Ok(()), std::time::Instant::now()),
            Err(Error::Cancelled)
        ));
        std::fs::remove_file(path).unwrap();
    }

    #[test]
    fn short_batch_decodes_once_and_preserves_full_clip_timing() {
        for limits in [FULL_WINDOW, SHORT_WINDOW] {
            let length = limits.maximum_samples().min(345600);
            let mut audio = vec![0.0001; length];
            let positions: Vec<usize> = [512, 120000, 163839, 163840, 250000, length - 1]
                .into_iter()
                .filter(|&position| position < length)
                .collect();
            for &position in &positions {
                audio[position] = 7.0;
            }
            let path = fixture(&audio);
            let source = WavSource::open(&path).unwrap();
            let recognizer = Markers::with_limits(limits);
            let mut updates = Vec::new();
            let result = transcribe_single_window(
                &recognizer,
                source,
                &Cancellation::new(),
                |progress| {
                    updates.push(progress);
                    Ok(())
                },
                std::time::Instant::now(),
            );
            std::fs::remove_file(path).unwrap();
            let result = result.unwrap();
            assert_eq!(recognizer.calls.get(), 1);
            assert_eq!(result.transcript.audio_samples, audio.len() as u64);
            assert_eq!(
                result
                    .transcript
                    .tokens
                    .iter()
                    .map(|t| t.start_sample)
                    .collect::<Vec<_>>(),
                positions.iter().map(|&p| p as u64).collect::<Vec<_>>()
            );
            assert_eq!(updates.len(), 1);
            assert_eq!(updates[0].processed_samples, audio.len() as u64);
            assert_eq!(updates[0].total_samples, audio.len() as u64);
            assert_eq!(result.peak_buffered_samples, audio.len());
        }
    }
}