mozu-audio 0.2.0

//! Runtime-independent audio services with dedicated model-owning threads.
//! Models remain local to their worker and are retained between requests.

use crate::{
    Cancellation, ComputeConfig, ComputeUnits, HaasEngine, InputLimits,
    result::AsrResult,
    streaming::StreamingSession,
    vad::{
        SileroVad, VadAnalysis, VadResult, VadSegment, VadSegmentationConfig, VadState,
        VadStreamResult, VadStreamState,
    },
    vocabulary::VocabularyRequest,
    window::{AsrConfig, AsrUpdate, JobId, JobUpdate, Transcript, UpdateKind, WindowConfig},
};
use async_channel::{Receiver, Sender, TrySendError, bounded};
use std::{
    collections::{BTreeMap, HashMap},
    future::Future,
    path::{Path, PathBuf},
    sync::{
        Arc,
        atomic::{AtomicU8, Ordering},
    },
    thread::{self, JoinHandle},
};

pub type Result<T> = std::result::Result<T, ServiceError>;

#[derive(Clone, Debug, thiserror::Error)]
pub enum ServiceError {
    #[error("Invalid audio service configuration: {0}")]
    Configuration(String),
    #[error("Cannot start audio worker: {0}")]
    WorkerStart(String),
    #[error("Audio worker stopped before completing the operation")]
    WorkerStopped,
    #[error("Audio worker panicked: {0}")]
    WorkerPanic(String),
    #[error("Audio service is shutting down")]
    Shutdown,
    #[error("ASR service already has an active job: job={job_id}")]
    Busy { job_id: JobId },
    #[error("Audio source is not registered: source={source_id}")]
    UnknownSource { source_id: SourceId },
    #[error("Audio source registry is full: capacity={capacity}; remove an unused source first")]
    SourceLimit { capacity: usize },
    #[error("ASR job is closed: job={job_id}")]
    JobClosed { job_id: JobId },
    #[error("Audio request cancelled: job={job_id}")]
    Cancelled { job_id: JobId },
    #[error("ASR input queue is full: capacity={capacity}")]
    InputFull { capacity: usize },
    #[error(
        "ASR update queue is full: job={job_id}, capacity={capacity}; consume updates concurrently"
    )]
    OutputFull { job_id: JobId, capacity: usize },
    #[error(
        "Invalid ASR audio: samples={samples}, non_finite={non_finite}; expected finite mono16kHz and <=16000 samples"
    )]
    InvalidAudio { samples: usize, non_finite: bool },
    #[error("ASR failed: operation={operation}, job={job_id:?}, error={source}")]
    Asr {
        operation: &'static str,
        job_id: Option<JobId>,
        #[source]
        source: Arc<crate::Error>,
    },
    #[error("VAD failed: operation={operation}, request={job_id:?}, error={source}")]
    Vad {
        operation: &'static str,
        job_id: Option<JobId>,
        #[source]
        source: Arc<crate::Error>,
    },
}

fn cancellation_error(job_id: Option<JobId>) -> ServiceError {
    match job_id {
        Some(job_id) => ServiceError::Cancelled { job_id },
        None => ServiceError::Shutdown,
    }
}

fn asr_error(operation: &'static str, job_id: Option<JobId>, source: crate::Error) -> ServiceError {
    match source {
        crate::Error::Cancelled => cancellation_error(job_id),
        source => ServiceError::Asr {
            operation,
            job_id,
            source: Arc::new(source),
        },
    }
}
fn vad_error(operation: &'static str, job_id: Option<JobId>, source: crate::Error) -> ServiceError {
    match source {
        crate::Error::Cancelled => cancellation_error(job_id),
        source => ServiceError::Vad {
            operation,
            job_id,
            source: Arc::new(source),
        },
    }
}
fn reply<T>(sender: Sender<Result<T>>, result: Result<T>) -> bool {
    match sender.try_send(result) {
        Ok(()) => true,
        // A cancelled caller has released its one-shot receiver.
        Err(TrySendError::Closed(_)) => false,
        Err(TrySendError::Full(_)) => {
            unreachable!("a one-shot reply has one producer and one message")
        }
    }
}

#[derive(Clone, Debug)]
pub enum DecoderPaths {
    Fused { predictor: PathBuf },
    Split { prediction: PathBuf, joint: PathBuf },
}

/// ASR assets without an endpoint detector model.
#[derive(Clone, Debug)]
pub struct AsrModelPaths {
    pub assets: PathBuf,
    pub encoder: PathBuf,
    pub decoder: DecoderPaths,
}
impl AsrModelPaths {
    fn load_with_compute(&self, compute: ComputeConfig) -> crate::Result<HaasEngine> {
        match &self.decoder {
            DecoderPaths::Fused { predictor } => {
                HaasEngine::load_with_compute(&self.assets, &self.encoder, predictor, compute)
            }
            DecoderPaths::Split { prediction, joint } => HaasEngine::load_split_with_compute(
                &self.assets,
                &self.encoder,
                prediction,
                joint,
                compute,
            ),
        }
    }
}

#[derive(Clone, Debug)]
pub struct ModelPaths {
    pub assets: PathBuf,
    pub encoder: PathBuf,
    pub decoder: DecoderPaths,
    pub vad: PathBuf,
}
impl ModelPaths {
    fn asr(&self) -> AsrModelPaths {
        AsrModelPaths {
            assets: self.assets.clone(),
            encoder: self.encoder.clone(),
            decoder: self.decoder.clone(),
        }
    }
}

/// Queue sizes count commands and regular updates. One extra update slot is
/// reserved for terminal success or error, so overload never hides termination.
#[derive(Clone, Copy, Debug)]
pub struct ServiceConfig {
    pub input_capacity: usize,
    pub update_capacity: usize,
}
impl ServiceConfig {
    fn validate(self) -> Result<()> {
        if self.input_capacity == 0
            || self.update_capacity == 0
            || self.update_capacity.checked_add(1).is_none()
        {
            return Err(ServiceError::Configuration(format!(
                "input_capacity={}, update_capacity={}",
                self.input_capacity, self.update_capacity
            )));
        }
        Ok(())
    }
}

/// One caller-assigned input identity, independent of the job generation.
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize)]
#[serde(transparent)]
pub struct SourceId(String);
impl SourceId {
    pub fn new(value: &str) -> Result<Self> {
        if value.is_empty()
            || value.trim() != value
            || value.len() > 256
            || value.chars().any(char::is_control)
        {
            return Err(ServiceError::Configuration(format!(
                "Invalid audio source identity: value={value:?}, bytes={}",
                value.len()
            )));
        }
        Ok(Self(value.to_owned()))
    }
    /// Source used by the original single-source start methods.
    pub fn primary() -> Self {
        Self("default".to_owned())
    }
    pub fn as_str(&self) -> &str {
        &self.0
    }
}
impl std::fmt::Display for SourceId {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str(&self.0)
    }
}

/// One retained model set with a bounded registry of independent audio sources.
/// Completed sources retain their last result until restarted or removed.
#[derive(Clone, Copy, Debug)]
pub struct ServiceOptions {
    pub queues: ServiceConfig,
    pub source_capacity: usize,
    pub compute: ComputeConfig,
    pub vad_compute: ComputeUnits,
}
impl ServiceOptions {
    pub fn single_source(queues: ServiceConfig) -> Self {
        Self {
            queues,
            source_capacity: 1,
            compute: ComputeConfig::standard(),
            vad_compute: ComputeUnits::CpuOnly,
        }
    }
    fn validate(self) -> Result<()> {
        self.queues.validate()?;
        if self.source_capacity == 0 {
            return Err(ServiceError::Configuration(
                "source_capacity must be positive: source_capacity=0".to_owned(),
            ));
        }
        Ok(())
    }
}

#[derive(Clone, Debug)]
pub enum SessionStatus {
    Running,
    Completed(Arc<AsrResult>),
    Failed(ServiceError),
}
/// Worker-ordered state, including updates not yet consumed by the event receiver.
#[derive(Clone, Debug)]
pub struct SessionSnapshot {
    pub source_id: SourceId,
    pub job_id: JobId,
    pub status: SessionStatus,
    pub latest_update: Option<JobUpdate>,
    /// Full current tokens and text; audio_samples includes processed input
    /// that has not yet produced a recognition window.
    pub transcript: Transcript,
}

const OPEN: u8 = 0;
const CANCEL_REQUESTED: u8 = 1;
const TERMINAL: u8 = 2;

struct JobControl {
    state: AtomicU8,
    cancellation: Cancellation,
}
impl JobControl {
    fn new() -> Self {
        Self {
            state: AtomicU8::new(OPEN),
            cancellation: Cancellation::new(),
        }
    }
    fn is_open(&self) -> bool {
        self.state.load(Ordering::Acquire) == OPEN
    }
    fn cancel(&self) {
        if self
            .state
            .compare_exchange(OPEN, CANCEL_REQUESTED, Ordering::AcqRel, Ordering::Acquire)
            .is_ok()
        {
            self.cancellation.cancel();
        }
    }
    fn is_cancelled(&self) -> bool {
        self.state.load(Ordering::Acquire) == CANCEL_REQUESTED
    }
}

enum SessionConfig {
    Windowed(WindowConfig),
    Vad(AsrConfig),
    VocabularyWindowed {
        config: WindowConfig,
        vocabulary: VocabularyRequest,
    },
    VocabularyVad {
        config: AsrConfig,
        vocabulary: VocabularyRequest,
    },
}
impl SessionConfig {
    fn validate_for(&self, limits: InputLimits) -> crate::Result<()> {
        match self {
            Self::Windowed(config) | Self::VocabularyWindowed { config, .. } => {
                config.validate_for(limits)
            }
            Self::Vad(config) | Self::VocabularyVad { config, .. } => config.validate_for(limits),
        }
    }
}
struct StartJob {
    id: JobId,
    source: SourceId,
    config: SessionConfig,
    control: Arc<JobControl>,
    updates: Sender<Result<JobUpdate>>,
    completion: Sender<Result<AsrResult>>,
    reply: Sender<Result<()>>,
}
enum Command {
    Start(StartJob),
    Audio {
        id: JobId,
        samples: Vec<f32>,
    },
    Finish {
        id: JobId,
    },
    Snapshot {
        source: SourceId,
        expected_job: Option<JobId>,
        reply: Sender<Result<SessionSnapshot>>,
    },
    Snapshots {
        reply: Sender<Result<Vec<SessionSnapshot>>>,
    },
    RemoveSource {
        source: SourceId,
        reply: Sender<Result<()>>,
    },
    Wake,
}

fn wake(commands: &Sender<Command>) {
    match commands.try_send(Command::Wake) {
        Ok(()) => {}
        // A full queue already wakes the worker; a closed queue is already stopping.
        Err(TrySendError::Full(_)) | Err(TrySendError::Closed(_)) => {}
    }
}

struct Worker<C> {
    commands: Sender<C>,
    shutdown: Cancellation,
    exited: Receiver<()>,
    thread: Option<JoinHandle<()>>,
}
impl<C> Worker<C> {
    fn stop(&self) {
        self.shutdown.cancel();
        self.commands.close();
    }
    async fn join(&mut self) -> Result<()> {
        self.stop();
        // The sender is a worker-lifetime guard and never sends a value.
        if self.exited.recv().await.is_ok() {
            return Err(ServiceError::WorkerStopped);
        }
        let handle = self.thread.take().ok_or(ServiceError::WorkerStopped)?;
        handle.join().map_err(|payload| {
            let message = if let Some(text) = payload.downcast_ref::<String>() {
                text.clone()
            } else if let Some(text) = payload.downcast_ref::<&str>() {
                (*text).to_owned()
            } else {
                "non-string panic payload".to_owned()
            };
            ServiceError::WorkerPanic(message)
        })
    }
}
impl<C> Drop for Worker<C> {
    fn drop(&mut self) {
        // Dropping signals shutdown without blocking the caller on an in-flight
        // Core ML prediction. shutdown().await explicitly waits for thread exit.
        self.stop();
    }
}
impl<C: Send + 'static> Worker<C> {
    fn spawn<F, R, M, Ready, Run>(
        name: &str,
        capacity: usize,
        load: F,
        metadata: Ready,
        run: Run,
    ) -> Result<(Self, Receiver<Result<M>>)>
    where
        F: FnOnce() -> Result<R> + Send + 'static,
        R: 'static,
        M: Send + 'static,
        Ready: FnOnce(&R) -> M + Send + 'static,
        Run: FnOnce(&R, Receiver<C>, Cancellation) + Send + 'static,
    {
        if capacity == 0 {
            return Err(ServiceError::Configuration(
                "input capacity must be positive: capacity=0".to_owned(),
            ));
        }
        let (commands, receiver) = bounded(capacity);
        let (ready, readiness) = bounded(1);
        let (lifetime, exited) = bounded(1);
        let shutdown = Cancellation::new();
        let worker_shutdown = shutdown.clone();
        let thread = thread::Builder::new()
            .name(name.to_owned())
            .spawn(move || {
                match load() {
                    Ok(runtime) => {
                        if reply(ready, Ok(metadata(&runtime))) {
                            run(&runtime, receiver, worker_shutdown);
                        }
                    }
                    Err(error) => {
                        reply(ready, Err(error));
                    }
                }
                drop(lifetime);
            })
            .map_err(|error| ServiceError::WorkerStart(error.to_string()))?;
        Ok((
            Self {
                commands,
                shutdown,
                exited,
                thread: Some(thread),
            },
            readiness,
        ))
    }
}

/// Models are loaded, used and released on one dedicated thread. Futures use
/// async-channel and can be polled by any executor; no Tokio runtime is required.
pub struct AsrService {
    worker: Worker<Command>,
    limits: InputLimits,
    config: ServiceConfig,
}
impl AsrService {
    pub async fn load(paths: ModelPaths, config: ServiceConfig) -> Result<Self> {
        Self::load_runtime(move || HaasRuntime::load(paths), config).await
    }
    /// Load only ASR models. VAD jobs require load() with an explicit VAD model.
    pub async fn load_windowed(paths: AsrModelPaths, config: ServiceConfig) -> Result<Self> {
        Self::load_runtime(move || HaasRuntime::load_windowed(paths), config).await
    }
    pub async fn load_with_options(paths: ModelPaths, options: ServiceOptions) -> Result<Self> {
        options.validate()?;
        Self::load_runtime_with_capacity(
            move || HaasRuntime::load_with_compute(paths, options.compute, options.vad_compute),
            options.queues,
            options.source_capacity,
        )
        .await
    }
    pub async fn load_windowed_with_options(
        paths: AsrModelPaths,
        options: ServiceOptions,
    ) -> Result<Self> {
        options.validate()?;
        Self::load_runtime_with_capacity(
            move || HaasRuntime::load_windowed_with_compute(paths, options.compute),
            options.queues,
            options.source_capacity,
        )
        .await
    }
    async fn load_runtime<F, R>(load: F, config: ServiceConfig) -> Result<Self>
    where
        F: FnOnce() -> crate::Result<R> + Send + 'static,
        R: Runtime + 'static,
    {
        Self::load_runtime_with_capacity(load, config, 1).await
    }
    async fn load_runtime_with_capacity<F, R>(
        load: F,
        config: ServiceConfig,
        source_capacity: usize,
    ) -> Result<Self>
    where
        F: FnOnce() -> crate::Result<R> + Send + 'static,
        R: Runtime + 'static,
    {
        config.validate()?;
        if source_capacity == 0 {
            return Err(ServiceError::Configuration(
                "source_capacity must be positive: source_capacity=0".to_owned(),
            ));
        }
        let (worker, readiness) = Worker::spawn(
            "mozu-asr",
            config.input_capacity,
            move || load().map_err(|error| asr_error("load models", None, error)),
            R::limits,
            move |runtime, receiver, shutdown| {
                run_worker(runtime, receiver, shutdown, config, source_capacity)
            },
        )?;
        let limits = readiness
            .recv()
            .await
            .map_err(|_| ServiceError::WorkerStopped)??;
        Ok(Self {
            worker,
            limits,
            config,
        })
    }
    pub fn input_limits(&self) -> InputLimits {
        self.limits
    }
    /// Start a fresh job on the retained models. A concurrent job returns Busy.
    /// Consume job.updates while feeding audio or awaiting input.finish().
    pub async fn start(&self, config: AsrConfig) -> Result<StreamingJob> {
        self.start_config(SourceId::primary(), SessionConfig::Vad(config))
            .await
    }
    /// Decode on complete windows and finish, without endpoint-triggered flushes.
    pub async fn start_windowed(&self, config: WindowConfig) -> Result<StreamingJob> {
        self.start_config(SourceId::primary(), SessionConfig::Windowed(config))
            .await
    }
    pub async fn start_source(&self, source: SourceId, config: AsrConfig) -> Result<StreamingJob> {
        self.start_config(source, SessionConfig::Vad(config)).await
    }
    pub async fn start_source_windowed(
        &self,
        source: SourceId,
        config: WindowConfig,
    ) -> Result<StreamingJob> {
        self.start_config(source, SessionConfig::Windowed(config))
            .await
    }
    pub async fn start_source_with_vocabulary(
        &self,
        source: SourceId,
        config: AsrConfig,
        vocabulary: VocabularyRequest,
    ) -> Result<StreamingJob> {
        self.start_config(source, SessionConfig::VocabularyVad { config, vocabulary })
            .await
    }
    pub async fn start_source_windowed_with_vocabulary(
        &self,
        source: SourceId,
        config: WindowConfig,
        vocabulary: VocabularyRequest,
    ) -> Result<StreamingJob> {
        self.start_config(
            source,
            SessionConfig::VocabularyWindowed { config, vocabulary },
        )
        .await
    }
    pub async fn snapshot(&self, source: &SourceId) -> Result<SessionSnapshot> {
        request_snapshot(&self.worker.commands, source.clone(), None).await
    }
    pub async fn snapshots(&self) -> Result<Vec<SessionSnapshot>> {
        let (reply, response) = bounded(1);
        self.worker
            .commands
            .send(Command::Snapshots { reply })
            .await
            .map_err(|_| ServiceError::WorkerStopped)?;
        response
            .recv()
            .await
            .map_err(|_| ServiceError::WorkerStopped)?
    }
    /// Cancel the source's active job and release its retained snapshot.
    pub async fn remove_source(&self, source: &SourceId) -> Result<()> {
        let (reply, response) = bounded(1);
        self.worker
            .commands
            .send(Command::RemoveSource {
                source: source.clone(),
                reply,
            })
            .await
            .map_err(|_| ServiceError::WorkerStopped)?;
        response
            .recv()
            .await
            .map_err(|_| ServiceError::WorkerStopped)?
    }
    async fn start_config(&self, source: SourceId, config: SessionConfig) -> Result<StreamingJob> {
        config
            .validate_for(self.limits)
            .map_err(|error| asr_error("validate job configuration", None, error))?;
        if self.worker.shutdown.check().is_err() {
            return Err(ServiceError::Shutdown);
        }
        let id = JobId::new();
        let control = Arc::new(JobControl::new());
        let (updates, receiver) = bounded(self.config.update_capacity + 1);
        let (completion, finished) = bounded(1);
        let (reply, response) = bounded(1);
        let job = StreamingJob {
            input: StreamingInput {
                id,
                source: source.clone(),
                commands: self.worker.commands.clone(),
                control: control.clone(),
                completion: finished,
                closed: false,
                capacity: self.config.input_capacity,
            },
            updates: UpdateStream {
                id,
                receiver,
                control: control.clone(),
                commands: self.worker.commands.clone(),
                ended: false,
                latest: None,
            },
        };
        self.worker
            .commands
            .send(Command::Start(StartJob {
                id,
                source,
                config,
                control,
                updates,
                completion,
                reply,
            }))
            .await
            .map_err(|_| ServiceError::WorkerStopped)?;
        response
            .recv()
            .await
            .map_err(|_| ServiceError::WorkerStopped)??;
        Ok(job)
    }
    /// Cancel unfinished work, release worker-owned models, and wait for exit.
    pub async fn shutdown(mut self) -> Result<()> {
        self.worker.join().await
    }
}

/// Asynchronous Silero inference on one model-owning worker. Audio is finite mono
/// 16 kHz f32. Requests are ordered and the bounded queue applies backpressure.
/// Streaming state belongs to the caller: await a result before submitting the
/// next chunk of that stream. Independent streams can share the retained model.
/// Dropping a request future cancels that request without cancelling the caller
/// token or sibling requests. Cancellation is checked between inference steps;
/// shutdown().await waits for the current prediction and model release.
pub struct VadService {
    worker: Worker<Box<dyn VadTask>>,
}
impl VadService {
    pub async fn load(model_path: PathBuf, queue_capacity: usize) -> Result<Self> {
        Self::load_with_compute(model_path, queue_capacity, ComputeUnits::CpuOnly).await
    }
    pub async fn load_with_compute(
        model_path: PathBuf,
        queue_capacity: usize,
        compute: ComputeUnits,
    ) -> Result<Self> {
        Self::load_runtime(
            move || SileroVad::load_with_compute(&model_path, compute),
            queue_capacity,
        )
        .await
    }
    async fn load_runtime<F, R>(load: F, queue_capacity: usize) -> Result<Self>
    where
        F: FnOnce() -> crate::Result<R> + Send + 'static,
        R: VadRuntime + 'static,
    {
        let (worker, readiness) = Worker::spawn(
            "mozu-vad",
            queue_capacity,
            move || load().map_err(|error| vad_error("load model", None, error)),
            |_| (),
            |runtime: &R, receiver: Receiver<Box<dyn VadTask>>, shutdown| {
                // Drain requests after close so every accepted request receives
                // an explicit shutdown result, including queued requests.
                while let Ok(task) = receiver.recv_blocking() {
                    task.execute(runtime, &shutdown);
                }
            },
        )?;
        readiness
            .recv()
            .await
            .map_err(|_| ServiceError::WorkerStopped)??;
        Ok(Self { worker })
    }
    /// Infer 1..=4096 samples and return a new recurrent model state.
    pub fn process_chunk(
        &self,
        audio: Vec<f32>,
        state: VadState,
        threshold: f32,
        cancellation: Cancellation,
    ) -> impl Future<Output = Result<VadResult>> + Send + use<> {
        self.request("process chunk", cancellation, move |model, token| {
            model.process_chunk(&audio, &state, threshold, token)
        })
    }
    /// Analyze an entire buffer with a fresh recurrent state.
    pub fn process_samples(
        &self,
        audio: Vec<f32>,
        threshold: f32,
        cancellation: Cancellation,
    ) -> impl Future<Output = Result<VadAnalysis>> + Send + use<> {
        self.request("process samples", cancellation, move |model, token| {
            model.process_samples(&audio, threshold, token)
        })
    }
    /// Analyze a mono 16 kHz WAV file using bounded reads.
    pub fn process_wav(
        &self,
        path: PathBuf,
        threshold: f32,
        cancellation: Cancellation,
    ) -> impl Future<Output = Result<VadAnalysis>> + Send + use<> {
        self.request("process WAV", cancellation, move |model, token| {
            model.process_wav(&path, threshold, token)
        })
    }
    /// Decode and convert a system-supported audio file on the VAD worker.
    pub fn process_file(
        &self,
        path: PathBuf,
        threshold: f32,
        cancellation: Cancellation,
    ) -> impl Future<Output = Result<VadAnalysis>> + Send + use<> {
        self.request("process file", cancellation, move |model, token| {
            model.process_file(&path, threshold, token)
        })
    }
    pub fn segment_file(
        &self,
        path: PathBuf,
        config: VadSegmentationConfig,
        cancellation: Cancellation,
    ) -> impl Future<Output = Result<Vec<VadSegment>>> + Send + use<> {
        self.request("segment file", cancellation, move |model, token| {
            model.segment_file(&path, &config, token)
        })
    }
    /// Advance one stream and return its new state and optional speech event.
    pub fn process_streaming_chunk(
        &self,
        audio: Vec<f32>,
        state: VadStreamState,
        config: VadSegmentationConfig,
        cancellation: Cancellation,
    ) -> impl Future<Output = Result<VadStreamResult>> + Send + use<> {
        self.request(
            "process streaming chunk",
            cancellation,
            move |model, token| model.process_streaming_chunk(&audio, &state, &config, token),
        )
    }
    pub fn segment_samples(
        &self,
        audio: Vec<f32>,
        config: VadSegmentationConfig,
        cancellation: Cancellation,
    ) -> impl Future<Output = Result<Vec<VadSegment>>> + Send + use<> {
        self.request("segment samples", cancellation, move |model, token| {
            model.segment_samples(&audio, &config, token)
        })
    }
    pub fn segment_wav(
        &self,
        path: PathBuf,
        config: VadSegmentationConfig,
        cancellation: Cancellation,
    ) -> impl Future<Output = Result<Vec<VadSegment>>> + Send + use<> {
        self.request("segment WAV", cancellation, move |model, token| {
            model.segment_wav(&path, &config, token)
        })
    }
    /// Cancel active and queued requests, release the model, and wait for exit.
    pub async fn shutdown(mut self) -> Result<()> {
        self.worker.join().await
    }
    fn request<T, F>(
        &self,
        operation: &'static str,
        cancellation: Cancellation,
        execute: F,
    ) -> impl Future<Output = Result<T>> + Send + use<T, F>
    where
        T: Send + 'static,
        F: FnOnce(&dyn VadRuntime, &Cancellation) -> crate::Result<T> + Send + 'static,
    {
        let commands = self.worker.commands.clone();
        let shutdown = self.worker.shutdown.clone();
        let cancellation = RequestCancellation(Cancellation::linked(&cancellation, &shutdown));
        async move {
            let id = JobId::new();
            request_status(id, &cancellation.0, &shutdown)?;
            let (sender, response) = bounded(1);
            let task = VadRequest {
                id,
                operation,
                cancellation: cancellation.0.clone(),
                reply: sender,
                execute,
            };
            let channel_error = || {
                request_status(id, &cancellation.0, &shutdown)
                    .err()
                    .unwrap_or(ServiceError::WorkerStopped)
            };
            commands
                .send(Box::new(task))
                .await
                .map_err(|_| channel_error())?;
            response.recv().await.map_err(|_| channel_error())?
        }
    }
}

// Cancels only this request when its future is dropped; caller tokens and sibling
// requests remain usable. Model predictions already in progress are cooperative.
struct RequestCancellation(Cancellation);
impl Drop for RequestCancellation {
    fn drop(&mut self) {
        self.0.cancel();
    }
}
fn request_status(id: JobId, cancellation: &Cancellation, shutdown: &Cancellation) -> Result<()> {
    if shutdown.check().is_err() {
        return Err(ServiceError::Shutdown);
    }
    cancellation
        .check()
        .map_err(|_| ServiceError::Cancelled { job_id: id })
}
trait VadTask: Send {
    fn execute(self: Box<Self>, runtime: &dyn VadRuntime, shutdown: &Cancellation);
}
struct VadRequest<T, F> {
    id: JobId,
    operation: &'static str,
    cancellation: Cancellation,
    reply: Sender<Result<T>>,
    execute: F,
}
impl<T, F> VadTask for VadRequest<T, F>
where
    T: Send + 'static,
    F: FnOnce(&dyn VadRuntime, &Cancellation) -> crate::Result<T> + Send + 'static,
{
    fn execute(self: Box<Self>, runtime: &dyn VadRuntime, shutdown: &Cancellation) {
        let VadRequest {
            id,
            operation,
            cancellation,
            reply: sender,
            execute,
        } = *self;
        let result = request_status(id, &cancellation, shutdown).and_then(|()| {
            let result = execute(runtime, &cancellation)
                .map_err(|error| vad_error(operation, Some(id), error));
            request_status(id, &cancellation, shutdown)?;
            result
        });
        reply(sender, result);
    }
}

trait VadRuntime {
    fn process_chunk(
        &self,
        audio: &[f32],
        state: &VadState,
        threshold: f32,
        cancellation: &Cancellation,
    ) -> crate::Result<VadResult>;
    fn process_samples(
        &self,
        audio: &[f32],
        threshold: f32,
        cancellation: &Cancellation,
    ) -> crate::Result<VadAnalysis>;
    fn process_wav(
        &self,
        path: &Path,
        threshold: f32,
        cancellation: &Cancellation,
    ) -> crate::Result<VadAnalysis>;
    fn process_file(
        &self,
        path: &Path,
        threshold: f32,
        cancellation: &Cancellation,
    ) -> crate::Result<VadAnalysis>;
    fn segment_file(
        &self,
        path: &Path,
        config: &VadSegmentationConfig,
        cancellation: &Cancellation,
    ) -> crate::Result<Vec<VadSegment>>;
    fn process_streaming_chunk(
        &self,
        audio: &[f32],
        state: &VadStreamState,
        config: &VadSegmentationConfig,
        cancellation: &Cancellation,
    ) -> crate::Result<VadStreamResult>;
    fn segment_samples(
        &self,
        audio: &[f32],
        config: &VadSegmentationConfig,
        cancellation: &Cancellation,
    ) -> crate::Result<Vec<VadSegment>>;
    fn segment_wav(
        &self,
        path: &Path,
        config: &VadSegmentationConfig,
        cancellation: &Cancellation,
    ) -> crate::Result<Vec<VadSegment>>;
}
impl VadRuntime for SileroVad {
    fn process_chunk(
        &self,
        audio: &[f32],
        state: &VadState,
        threshold: f32,
        cancellation: &Cancellation,
    ) -> crate::Result<VadResult> {
        self.process_chunk(audio, state, threshold, cancellation)
    }
    fn process_samples(
        &self,
        audio: &[f32],
        threshold: f32,
        cancellation: &Cancellation,
    ) -> crate::Result<VadAnalysis> {
        self.process_samples(audio, threshold, cancellation)
    }
    fn process_wav(
        &self,
        path: &Path,
        threshold: f32,
        cancellation: &Cancellation,
    ) -> crate::Result<VadAnalysis> {
        self.process_wav(path, threshold, cancellation)
    }
    fn process_file(
        &self,
        path: &Path,
        threshold: f32,
        cancellation: &Cancellation,
    ) -> crate::Result<VadAnalysis> {
        self.process_file(path, threshold, cancellation)
    }
    fn segment_file(
        &self,
        path: &Path,
        config: &VadSegmentationConfig,
        cancellation: &Cancellation,
    ) -> crate::Result<Vec<VadSegment>> {
        self.segment_file(path, config, cancellation)
    }
    fn process_streaming_chunk(
        &self,
        audio: &[f32],
        state: &VadStreamState,
        config: &VadSegmentationConfig,
        cancellation: &Cancellation,
    ) -> crate::Result<VadStreamResult> {
        self.process_streaming_chunk(audio, state, config, cancellation)
    }
    fn segment_samples(
        &self,
        audio: &[f32],
        config: &VadSegmentationConfig,
        cancellation: &Cancellation,
    ) -> crate::Result<Vec<VadSegment>> {
        self.segment_samples(audio, config, cancellation)
    }
    fn segment_wav(
        &self,
        path: &Path,
        config: &VadSegmentationConfig,
        cancellation: &Cancellation,
    ) -> crate::Result<Vec<VadSegment>> {
        self.segment_wav(path, config, cancellation)
    }
}

pub struct StreamingJob {
    pub input: StreamingInput,
    pub updates: UpdateStream,
}

/// One ordered audio producer. Each accepted block is <=1s mono16kHz f32.
/// Dropping an unfinished producer cancels its job. Input overload is explicit.
pub struct StreamingInput {
    id: JobId,
    source: SourceId,
    commands: Sender<Command>,
    control: Arc<JobControl>,
    completion: Receiver<Result<AsrResult>>,
    closed: bool,
    capacity: usize,
}
impl StreamingInput {
    pub fn job_id(&self) -> JobId {
        self.id
    }
    pub fn source_id(&self) -> &SourceId {
        &self.source
    }
    pub async fn snapshot(&self) -> Result<SessionSnapshot> {
        request_snapshot(&self.commands, self.source.clone(), Some(self.id)).await
    }
    fn audio_command(&self, audio: &[f32]) -> Result<Command> {
        if self.closed || !self.control.is_open() {
            return Err(ServiceError::JobClosed { job_id: self.id });
        }
        let non_finite = audio.iter().any(|sample| !sample.is_finite());
        if audio.len() > 16000 || non_finite {
            return Err(ServiceError::InvalidAudio {
                samples: audio.len(),
                non_finite,
            });
        }
        Ok(Command::Audio {
            id: self.id,
            samples: audio.to_vec(),
        })
    }
    /// Wait for input capacity; success means enqueued, not yet recognized.
    /// Later cancellation or a terminal job error still aborts queued input.
    pub async fn push_audio(&mut self, audio: &[f32]) -> Result<()> {
        self.commands
            .send(self.audio_command(audio)?)
            .await
            .map_err(|_| ServiceError::WorkerStopped)
    }
    /// Never waits. On InputFull, the caller retains audio and can retry it.
    pub fn try_push_audio(&mut self, audio: &[f32]) -> Result<()> {
        self.commands
            .try_send(self.audio_command(audio)?)
            .map_err(|error| match error {
                TrySendError::Full(_) => ServiceError::InputFull {
                    capacity: self.capacity,
                },
                TrySendError::Closed(_) => ServiceError::WorkerStopped,
            })
    }
    /// Drain accepted audio, publish one final update, and return the transcript.
    /// Consume updates concurrently; unread regular updates can cause OutputFull.
    pub async fn finish(self) -> Result<Transcript> {
        self.finish_result().await.map(AsrResult::into_transcript)
    }
    /// Finish with measured ASR metadata and vocabulary pieces.
    pub async fn finish_result(mut self) -> Result<AsrResult> {
        if self.control.is_open() {
            self.commands
                .send(Command::Finish { id: self.id })
                .await
                .map_err(|_| ServiceError::WorkerStopped)?;
        }
        self.closed = true;
        self.completion
            .recv()
            .await
            .map_err(|_| ServiceError::WorkerStopped)?
    }
    /// Abort queued work and wait until this job stops, without a success terminal.
    pub async fn cancel(mut self) -> Result<()> {
        self.control.cancel();
        self.closed = true;
        wake(&self.commands);
        match self
            .completion
            .recv()
            .await
            .map_err(|_| ServiceError::WorkerStopped)?
        {
            Err(ServiceError::Cancelled { .. }) => Ok(()),
            Err(error) => Err(error),
            Ok(_) => Err(ServiceError::JobClosed { job_id: self.id }),
        }
    }
}
impl Drop for StreamingInput {
    fn drop(&mut self) {
        if !self.closed {
            self.control.cancel();
            wake(&self.commands);
        }
    }
}

/// Single-consumer ordered updates. A final update or returned error terminates
/// iteration; premature channel closure is an error. Dropping it cancels the job.
pub struct UpdateStream {
    id: JobId,
    receiver: Receiver<Result<JobUpdate>>,
    control: Arc<JobControl>,
    commands: Sender<Command>,
    ended: bool,
    latest: Option<JobUpdate>,
}
impl UpdateStream {
    pub fn job_id(&self) -> JobId {
        self.id
    }
    /// Last update consumed by this receiver, not a snapshot of unconsumed work.
    pub fn latest(&self) -> Option<&JobUpdate> {
        self.latest.as_ref()
    }
    pub async fn next(&mut self) -> Result<Option<JobUpdate>> {
        if self.ended {
            return Ok(None);
        }
        match self.receiver.recv().await {
            Ok(Ok(event)) => {
                self.ended = event.update.kind == UpdateKind::Final;
                self.latest = Some(event.clone());
                Ok(Some(event))
            }
            Ok(Err(error)) => {
                self.ended = true;
                Err(error)
            }
            Err(_) => {
                self.ended = true;
                Err(ServiceError::WorkerStopped)
            }
        }
    }
}
impl Drop for UpdateStream {
    fn drop(&mut self) {
        if !self.ended {
            self.control.cancel();
            wake(&self.commands);
        }
    }
}

trait Session {
    fn push(&mut self, audio: &[f32], cancellation: &Cancellation)
    -> crate::Result<Vec<AsrUpdate>>;
    fn finish(&mut self, cancellation: &Cancellation) -> crate::Result<(AsrResult, AsrUpdate)>;
}
impl Session for StreamingSession<'_> {
    fn push(
        &mut self,
        audio: &[f32],
        cancellation: &Cancellation,
    ) -> crate::Result<Vec<AsrUpdate>> {
        self.push_audio(audio, cancellation)
    }
    fn finish(&mut self, cancellation: &Cancellation) -> crate::Result<(AsrResult, AsrUpdate)> {
        self.finish_result(cancellation)
    }
}
trait Runtime {
    fn limits(&self) -> InputLimits;
    fn start(&self, config: SessionConfig) -> crate::Result<Box<dyn Session + '_>>;
}
struct HaasRuntime {
    engine: HaasEngine,
    vad: Option<SileroVad>,
}
impl HaasRuntime {
    fn load(paths: ModelPaths) -> crate::Result<Self> {
        Self::load_with_compute(paths, ComputeConfig::standard(), ComputeUnits::CpuOnly)
    }
    fn load_with_compute(
        paths: ModelPaths,
        compute: ComputeConfig,
        vad_compute: ComputeUnits,
    ) -> crate::Result<Self> {
        let engine = paths.asr().load_with_compute(compute)?;
        Ok(Self {
            engine,
            vad: Some(SileroVad::load_with_compute(&paths.vad, vad_compute)?),
        })
    }
    fn load_windowed(paths: AsrModelPaths) -> crate::Result<Self> {
        Self::load_windowed_with_compute(paths, ComputeConfig::standard())
    }
    fn load_windowed_with_compute(
        paths: AsrModelPaths,
        compute: ComputeConfig,
    ) -> crate::Result<Self> {
        Ok(Self {
            engine: paths.load_with_compute(compute)?,
            vad: None,
        })
    }
}
impl Runtime for HaasRuntime {
    fn limits(&self) -> InputLimits {
        self.engine.input_limits()
    }
    fn start(&self, config: SessionConfig) -> crate::Result<Box<dyn Session + '_>> {
        let session = match config {
            SessionConfig::Windowed(config) => StreamingSession::windowed(&self.engine, config)?,
            SessionConfig::Vad(config) => StreamingSession::new(&self.engine, self.vad.as_ref().ok_or_else(|| crate::Error::Invalid("VAD model is not loaded; use start_windowed or load models with an explicit VAD path".to_owned()))?, config)?,
            SessionConfig::VocabularyWindowed { config, vocabulary } => {
                let compiled = Arc::new(self.engine.compile_vocabulary(vocabulary.vocabulary, &vocabulary.tokenizer_model)?);
                StreamingSession::windowed_with_vocabulary(&self.engine, config, compiled)?
            }
            SessionConfig::VocabularyVad { config, vocabulary } => {
                let compiled = Arc::new(self.engine.compile_vocabulary(vocabulary.vocabulary, &vocabulary.tokenizer_model)?);
                let vad = self.vad.as_ref().ok_or_else(|| crate::Error::Invalid("VAD vocabulary job requires an explicitly loaded VAD model".to_owned()))?;
                StreamingSession::new_with_vocabulary(&self.engine, vad, config, compiled)?
            }
        };
        Ok(Box::new(session))
    }
}

struct ActiveJob<'a> {
    id: JobId,
    source: SourceId,
    latest: Option<JobUpdate>,
    current: Transcript,
    session: Box<dyn Session + 'a>,
    control: Arc<JobControl>,
    updates: Sender<Result<JobUpdate>>,
    completion: Sender<Result<AsrResult>>,
    capacity: usize,
}
impl ActiveJob<'_> {
    fn publish(&mut self, update: AsrUpdate) -> Result<()> {
        if self.updates.len() >= self.capacity {
            return Err(ServiceError::OutputFull {
                job_id: self.id,
                capacity: self.capacity,
            });
        }
        let revision = &update.revision;
        if revision.previous_token_count != self.current.tokens.len()
            || revision.retained_token_count > self.current.tokens.len()
        {
            return Err(asr_error(
                "publish token revision",
                Some(self.id),
                crate::Error::Invalid(format!(
                    "Invalid service revision: current_tokens={}, previous={}, retained={}",
                    self.current.tokens.len(),
                    revision.previous_token_count,
                    revision.retained_token_count
                )),
            ));
        }
        let event = JobUpdate {
            job_id: self.id,
            update,
        };
        self.updates
            .try_send(Ok(event.clone()))
            .map_err(|error| match error {
                TrySendError::Full(_) => ServiceError::OutputFull {
                    job_id: self.id,
                    capacity: self.capacity,
                },
                TrySendError::Closed(_) => ServiceError::JobClosed { job_id: self.id },
            })?;
        self.current
            .tokens
            .truncate(event.update.revision.retained_token_count);
        self.current
            .tokens
            .extend_from_slice(&event.update.revision.appended_tokens);
        self.current.text.clone_from(&event.update.confirmed_text);
        self.current.text.push_str(&event.update.provisional_text);
        self.latest = Some(event);
        Ok(())
    }
    fn snapshot(&self) -> SessionSnapshot {
        SessionSnapshot {
            source_id: self.source.clone(),
            job_id: self.id,
            status: SessionStatus::Running,
            latest_update: self.latest.clone(),
            transcript: self.current.clone(),
        }
    }
    fn close(self, result: Result<(AsrResult, AsrUpdate)>) -> SessionSnapshot {
        let result = if self.control.state.swap(TERMINAL, Ordering::AcqRel) == CANCEL_REQUESTED {
            Err(ServiceError::Cancelled { job_id: self.id })
        } else {
            result
        };
        let terminal = result
            .as_ref()
            .map(|(_, update)| JobUpdate {
                job_id: self.id,
                update: update.clone(),
            })
            .map_err(Clone::clone);
        // Only this worker sends updates. Regular publication never uses the
        // reserved terminal slot, even when the consumer is not scheduled.
        let latest = match &terminal {
            Ok(update) => Some(update.clone()),
            Err(_) => self.latest,
        };
        let outcome = match self.updates.try_send(terminal) {
            Ok(()) => result.map(|(transcript, _)| transcript),
            Err(TrySendError::Closed(_)) => {
                result.and(Err(ServiceError::JobClosed { job_id: self.id }))
            }
            Err(TrySendError::Full(_)) => {
                unreachable!("regular publication reserves one terminal slot")
            }
        };
        self.updates.close();
        let transcript = match &outcome {
            Ok(result) => result.clone().into_transcript(),
            Err(_) => self.current,
        };
        let status = match &outcome {
            Ok(result) => SessionStatus::Completed(Arc::new(result.clone())),
            Err(error) => SessionStatus::Failed(error.clone()),
        };
        let snapshot = SessionSnapshot {
            source_id: self.source,
            job_id: self.id,
            status,
            latest_update: latest,
            transcript,
        };
        reply(self.completion, outcome);
        snapshot
    }
}

async fn request_snapshot(
    commands: &Sender<Command>,
    source: SourceId,
    expected_job: Option<JobId>,
) -> Result<SessionSnapshot> {
    let (reply, response) = bounded(1);
    commands
        .send(Command::Snapshot {
            source,
            expected_job,
            reply,
        })
        .await
        .map_err(|_| ServiceError::WorkerStopped)?;
    response
        .recv()
        .await
        .map_err(|_| ServiceError::WorkerStopped)?
}

struct SessionRegistry<'a, R: Runtime> {
    runtime: &'a R,
    active: HashMap<JobId, ActiveJob<'a>>,
    sources: BTreeMap<SourceId, SessionSnapshot>,
    source_capacity: usize,
    update_capacity: usize,
}
impl<'a, R: Runtime> SessionRegistry<'a, R> {
    fn start(&mut self, request: StartJob) {
        let StartJob {
            id,
            source,
            config,
            control,
            updates,
            completion,
            reply,
        } = request;
        let result = if let Some(existing) = self
            .sources
            .get(&source)
            .filter(|entry| matches!(entry.status, SessionStatus::Running))
        {
            Err(ServiceError::Busy {
                job_id: existing.job_id,
            })
        } else if !self.sources.contains_key(&source) && self.sources.len() >= self.source_capacity
        {
            Err(ServiceError::SourceLimit {
                capacity: self.source_capacity,
            })
        } else {
            let runtime = self.runtime;
            runtime
                .start(config)
                .map(|session| {
                    let job = ActiveJob {
                        id,
                        source: source.clone(),
                        latest: None,
                        current: Transcript {
                            text: String::new(),
                            tokens: Vec::new(),
                            audio_samples: 0,
                        },
                        session,
                        control,
                        updates,
                        completion,
                        capacity: self.update_capacity,
                    };
                    self.sources.insert(source, job.snapshot());
                    self.active.insert(id, job);
                })
                .map_err(|error| asr_error("start job", Some(id), error))
        };
        self::reply(reply, result);
    }
    fn retire(&mut self, id: JobId, result: Result<(AsrResult, AsrUpdate)>) {
        let job = self
            .active
            .remove(&id)
            .expect("only active jobs can retire");
        let snapshot = job.close(result);
        self.sources.insert(snapshot.source_id.clone(), snapshot);
    }
    fn cancel_pending(&mut self) {
        let cancelled: Vec<_> = self
            .active
            .values()
            .filter(|job| job.control.is_cancelled())
            .map(|job| job.id)
            .collect();
        for id in cancelled {
            self.retire(id, Err(ServiceError::Cancelled { job_id: id }));
        }
    }
    fn push(&mut self, id: JobId, samples: &[f32]) {
        if let Some(job) = self.active.get_mut(&id) {
            let result = job
                .session
                .push(samples, &job.control.cancellation)
                .map_err(|error| asr_error("push audio", Some(id), error))
                .and_then(|updates| {
                    job.current.audio_samples = job
                        .current
                        .audio_samples
                        .checked_add(samples.len() as u64)
                        .ok_or_else(|| {
                            asr_error(
                                "count source input",
                                Some(id),
                                crate::Error::Invalid("Source sample count overflow".to_owned()),
                            )
                        })?;
                    updates
                        .into_iter()
                        .try_for_each(|update| job.publish(update))
                });
            if let Err(error) = result {
                self.retire(id, Err(error));
            }
        }
        // Input for a terminated job is retired after its explicit terminal error.
    }
    fn finish(&mut self, id: JobId) {
        if let Some(job) = self.active.get_mut(&id) {
            let result = job
                .session
                .finish(&job.control.cancellation)
                .map_err(|error| asr_error("finish job", Some(id), error));
            self.retire(id, result);
        }
    }
    fn snapshot(&self, source: &SourceId) -> Result<SessionSnapshot> {
        let snapshot = self
            .sources
            .get(source)
            .ok_or_else(|| ServiceError::UnknownSource {
                source_id: source.clone(),
            })?;
        match snapshot.status {
            SessionStatus::Running => Ok(self
                .active
                .get(&snapshot.job_id)
                .expect("a running source owns an active job")
                .snapshot()),
            _ => Ok(snapshot.clone()),
        }
    }
    fn snapshots(&self) -> Result<Vec<SessionSnapshot>> {
        self.sources
            .keys()
            .map(|source| self.snapshot(source))
            .collect()
    }
    fn remove(&mut self, source: &SourceId) -> Result<()> {
        let snapshot = self.snapshot(source)?;
        if let Some(job) = self.active.get(&snapshot.job_id) {
            job.control.cancel();
            self.retire(
                snapshot.job_id,
                Err(ServiceError::Cancelled {
                    job_id: snapshot.job_id,
                }),
            );
        }
        self.sources.remove(source);
        Ok(())
    }
    fn shutdown(&mut self) {
        let ids: Vec<_> = self.active.keys().copied().collect();
        for id in ids {
            self.retire(id, Err(ServiceError::Shutdown));
        }
    }
}

fn run_worker(
    runtime: &impl Runtime,
    commands: Receiver<Command>,
    shutdown: Cancellation,
    config: ServiceConfig,
    source_capacity: usize,
) {
    let mut sessions = SessionRegistry {
        runtime,
        active: HashMap::new(),
        sources: BTreeMap::new(),
        source_capacity,
        update_capacity: config.update_capacity,
    };
    while let Ok(command) = commands.recv_blocking() {
        if shutdown.check().is_err() {
            break;
        }
        sessions.cancel_pending();
        match command {
            Command::Start(request) => sessions.start(request),
            Command::Audio { id, samples } => sessions.push(id, &samples),
            Command::Finish { id } => sessions.finish(id),
            Command::Snapshot {
                source,
                expected_job,
                reply,
            } => {
                let result = sessions.snapshot(&source).and_then(|snapshot| {
                    if let Some(id) = expected_job
                        && snapshot.job_id != id
                    {
                        return Err(ServiceError::JobClosed { job_id: id });
                    }
                    Ok(snapshot)
                });
                self::reply(reply, result);
            }
            Command::Snapshots { reply } => {
                self::reply(reply, sessions.snapshots());
            }
            Command::RemoveSource { source, reply } => {
                self::reply(reply, sessions.remove(&source));
            }
            Command::Wake => {}
        }
    }
    sessions.shutdown();
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::window::{
        ENCODER_STEP, TimedToken, WindowRecognizer,
        test_support::{FULL_WINDOW, Markers},
    };
    use futures_lite::future::{block_on, poll_once, zip};
    use std::{
        sync::{Mutex, mpsc},
        time::Duration,
    };

    struct MarkerRuntime {
        markers: Markers,
    }
    fn configured_session<'a>(
        recognizer: &'a dyn WindowRecognizer,
        config: SessionConfig,
    ) -> crate::Result<StreamingSession<'a>> {
        match config {
            SessionConfig::Windowed(config) => {
                StreamingSession::with_windowed_recognizer(recognizer, config)
            }
            SessionConfig::Vad(config) => StreamingSession::with_recognizer(recognizer, config),
            SessionConfig::VocabularyWindowed { .. } | SessionConfig::VocabularyVad { .. } => Err(
                crate::Error::Invalid("Vocabulary requests require a Haas runtime".to_owned()),
            ),
        }
    }
    impl Runtime for MarkerRuntime {
        fn limits(&self) -> InputLimits {
            FULL_WINDOW
        }
        fn start(&self, config: SessionConfig) -> crate::Result<Box<dyn Session + '_>> {
            Ok(Box::new(configured_session(&self.markers, config)?))
        }
    }
    async fn markers(config: ServiceConfig) -> AsrService {
        AsrService::load_runtime(
            || {
                Ok(MarkerRuntime {
                    markers: Markers::new(),
                })
            },
            config,
        )
        .await
        .unwrap()
    }
    fn audio(token: usize) -> Vec<f32> {
        let mut values = vec![0.0001; 16000];
        values[8000] = token as f32;
        values
    }
    async fn collect(updates: &mut UpdateStream) -> Result<Vec<JobUpdate>> {
        let mut events = Vec::new();
        while let Some(event) = updates.next().await? {
            events.push(event);
        }
        Ok(events)
    }
    fn assert_replay(events: &[JobUpdate], transcript: &Transcript, id: JobId) {
        let mut tokens: Vec<TimedToken> = Vec::new();
        let mut stable = 0;
        for (index, event) in events.iter().enumerate() {
            assert_eq!(event.job_id, id);
            assert_eq!(event.update.sequence as usize, index + 1);
            let revision = &event.update.revision;
            assert_eq!(revision.previous_token_count, tokens.len());
            assert!(revision.retained_token_count >= stable);
            tokens.truncate(revision.retained_token_count);
            tokens.extend_from_slice(&revision.appended_tokens);
            assert!(
                revision.stable_token_count >= stable
                    && revision.stable_token_count <= tokens.len()
            );
            stable = revision.stable_token_count;
            if let Some(span) = &event.update.recognition_span {
                assert!(
                    span.window.start <= span.center.start && span.center.end <= span.window.end
                );
                assert!(span.window.end <= event.update.received_samples);
            }
        }
        assert_eq!(tokens, transcript.tokens);
        assert_eq!(stable, tokens.len());
        assert_eq!(
            events
                .iter()
                .filter(|e| e.update.kind == UpdateKind::Final)
                .count(),
            1
        );
    }

    #[test]
    fn shared_sources_keep_independent_audio_and_query_unconsumed_worker_state() {
        block_on(async {
            let loads = Arc::new(std::sync::atomic::AtomicUsize::new(0));
            let counted = loads.clone();
            let service = AsrService::load_runtime_with_capacity(
                move || {
                    counted.fetch_add(1, Ordering::SeqCst);
                    Ok(MarkerRuntime {
                        markers: Markers::new(),
                    })
                },
                ServiceConfig {
                    input_capacity: 4,
                    update_capacity: 64,
                },
                2,
            )
            .await
            .unwrap();
            let microphone = SourceId::new("microphone").unwrap();
            let system = SourceId::new("system").unwrap();
            let config = WindowConfig {
                center_samples: 16000 / ENCODER_STEP * ENCODER_STEP,
                left_samples: 10240,
                right_samples: 0,
            };
            let StreamingJob {
                input: mut mic_input,
                updates: mut mic_updates,
            } = service
                .start_source_windowed(microphone.clone(), config.clone())
                .await
                .unwrap();
            let StreamingJob {
                input: mut sys_input,
                updates: mut sys_updates,
            } = service
                .start_source_windowed(system.clone(), config)
                .await
                .unwrap();
            let mic_id = mic_input.job_id();
            let sys_id = sys_input.job_id();
            assert_ne!(mic_id, sys_id);
            for token in 1..=3 {
                mic_input.push_audio(&audio(token)).await.unwrap();
                sys_input.push_audio(&audio(token + 10)).await.unwrap();
            }
            let mic_snapshot = mic_input.snapshot().await.unwrap();
            let sys_snapshot = service.snapshot(&system).await.unwrap();
            assert_eq!(mic_snapshot.job_id, mic_id);
            assert_eq!(sys_snapshot.job_id, sys_id);
            assert!(mic_updates.latest().is_none());
            assert!(sys_updates.latest().is_none());
            let mic_update = mic_snapshot.latest_update.unwrap();
            let sys_update = sys_snapshot.latest_update.unwrap();
            assert!(mic_update.update.received_samples > 0);
            assert!(sys_update.update.received_samples > 0);
            assert!(
                mic_update
                    .update
                    .revision
                    .appended_tokens
                    .iter()
                    .all(|token| token.token_id < 10)
            );
            assert!(
                sys_update
                    .update
                    .revision
                    .appended_tokens
                    .iter()
                    .all(|token| token.token_id > 10)
            );
            let (mic_result, sys_result) = zip(mic_input.finish(), sys_input.finish()).await;
            let mic_result = mic_result.unwrap();
            let sys_result = sys_result.unwrap();
            assert_eq!(
                mic_result
                    .tokens
                    .iter()
                    .map(|token| token.token_id)
                    .collect::<Vec<_>>(),
                vec![1, 2, 3]
            );
            assert_eq!(
                sys_result
                    .tokens
                    .iter()
                    .map(|token| token.token_id)
                    .collect::<Vec<_>>(),
                vec![11, 12, 13]
            );
            assert_replay(
                &collect(&mut mic_updates).await.unwrap(),
                &mic_result,
                mic_id,
            );
            assert_replay(
                &collect(&mut sys_updates).await.unwrap(),
                &sys_result,
                sys_id,
            );
            assert!(matches!(
                service.snapshot(&microphone).await.unwrap().status,
                SessionStatus::Completed(_)
            ));
            assert_eq!(service.snapshots().await.unwrap().len(), 2);
            assert_eq!(loads.load(Ordering::SeqCst), 1);
            service.shutdown().await.unwrap();
        });
    }

    #[test]
    fn source_removal_and_overload_do_not_cancel_siblings_or_mix_job_generations() {
        block_on(async {
            let service = AsrService::load_runtime_with_capacity(
                || {
                    Ok(MarkerRuntime {
                        markers: Markers::new(),
                    })
                },
                ServiceConfig {
                    input_capacity: 4,
                    update_capacity: 8,
                },
                2,
            )
            .await
            .unwrap();
            let first = SourceId::new("first").unwrap();
            let second = SourceId::new("second").unwrap();
            let third = SourceId::new("third").unwrap();
            let StreamingJob {
                input: first_input,
                updates: mut first_updates,
            } = service
                .start_source(first.clone(), AsrConfig::interactive())
                .await
                .unwrap();
            let StreamingJob {
                input: mut second_input,
                updates: mut second_updates,
            } = service
                .start_source(second.clone(), AsrConfig::interactive())
                .await
                .unwrap();
            let second_id = second_input.job_id();
            assert!(matches!(
                service
                    .start_source(first.clone(), AsrConfig::interactive())
                    .await,
                Err(ServiceError::Busy { .. })
            ));
            assert!(matches!(
                service
                    .start_source(third.clone(), AsrConfig::interactive())
                    .await,
                Err(ServiceError::SourceLimit { capacity: 2 })
            ));
            let previous_id = first_input.job_id();
            service.remove_source(&first).await.unwrap();
            assert!(
                matches!(first_updates.next().await, Err(ServiceError::Cancelled { job_id }) if job_id == previous_id)
            );
            let StreamingJob {
                input: replacement,
                mut updates,
            } = service
                .start_source(first.clone(), AsrConfig::interactive())
                .await
                .unwrap();
            assert_ne!(replacement.job_id(), previous_id);
            assert!(
                matches!(first_input.snapshot().await, Err(ServiceError::JobClosed { job_id }) if job_id == previous_id)
            );
            replacement.cancel().await.unwrap();
            assert!(matches!(
                updates.next().await,
                Err(ServiceError::Cancelled { .. })
            ));
            second_input.push_audio(&audio(7)).await.unwrap();
            let transcript = second_input.finish().await.unwrap();
            assert_eq!(
                transcript
                    .tokens
                    .iter()
                    .map(|token| token.token_id)
                    .collect::<Vec<_>>(),
                vec![7]
            );
            assert_replay(
                &collect(&mut second_updates).await.unwrap(),
                &transcript,
                second_id,
            );
            service.remove_source(&first).await.unwrap();
            assert!(matches!(
                service.snapshot(&first).await,
                Err(ServiceError::UnknownSource { .. })
            ));
            let job = service
                .start_source(third, AsrConfig::interactive())
                .await
                .unwrap();
            job.input.cancel().await.unwrap();
            service.shutdown().await.unwrap();
        });
    }

    #[test]
    fn drains_accepted_audio_replays_events_and_reuses_models_with_distinct_jobs() {
        block_on(async {
            let service = markers(ServiceConfig {
                input_capacity: 2,
                update_capacity: 64,
            })
            .await;
            let mut previous = None;
            for _ in 0..2 {
                let StreamingJob {
                    mut input,
                    mut updates,
                } = service.start(AsrConfig::interactive()).await.unwrap();
                let id = input.job_id();
                assert_ne!(previous, Some(id));
                previous = Some(id);
                let producer = async {
                    for token in 1..=35 {
                        input.push_audio(&audio(token)).await?;
                    }
                    input.finish().await
                };
                let consumer = collect(&mut updates);
                let (transcript, events) = zip(producer, consumer).await;
                assert!(
                    transcript.is_ok(),
                    "producer={transcript:?}, updates={events:?}"
                );
                let transcript = transcript.unwrap();
                assert_eq!(transcript.audio_samples, 35 * 16000);
                assert_eq!(
                    transcript
                        .tokens
                        .iter()
                        .map(|t| t.token_id)
                        .collect::<Vec<_>>(),
                    (1..=35).collect::<Vec<_>>()
                );
                assert_replay(&events.unwrap(), &transcript, id);
                assert_eq!(updates.latest().unwrap().update.kind, UpdateKind::Final);
            }
            service.shutdown().await.unwrap();
        });
    }
    #[test]
    fn empty_job_emits_one_terminal_without_fabricating_an_inference_span() {
        block_on(async {
            let service = markers(ServiceConfig {
                input_capacity: 1,
                update_capacity: 1,
            })
            .await;
            let StreamingJob { input, mut updates } =
                service.start(AsrConfig::interactive()).await.unwrap();
            let id = input.job_id();
            let transcript = input.finish().await.unwrap();
            let events = collect(&mut updates).await.unwrap();
            assert_eq!(events.len(), 1);
            assert!(events[0].update.recognition_span.is_none());
            assert_replay(&events, &transcript, id);
            service.shutdown().await.unwrap();
        });
    }
    #[test]
    fn service_reuses_models_across_windowed_and_vad_jobs_with_distinct_identities() {
        block_on(async {
            let service = markers(ServiceConfig {
                input_capacity: 2,
                update_capacity: 16,
            })
            .await;
            let StreamingJob {
                mut input,
                mut updates,
            } = service
                .start_windowed(WindowConfig::interactive_for(FULL_WINDOW).unwrap())
                .await
                .unwrap();
            let windowed_id = input.job_id();
            input.push_audio(&audio(7)).await.unwrap();
            let transcript = input.finish().await.unwrap();
            let events = collect(&mut updates).await.unwrap();
            assert_eq!(events.len(), 1);
            assert_eq!(transcript.text, "7");
            assert_replay(&events, &transcript, windowed_id);
            let StreamingJob {
                mut input,
                mut updates,
            } = service.start(AsrConfig::interactive()).await.unwrap();
            let vad_id = input.job_id();
            assert_ne!(vad_id, windowed_id);
            input.push_audio(&audio(9)).await.unwrap();
            let transcript = input.finish().await.unwrap();
            assert_eq!(transcript.text, "9");
            assert_replay(&collect(&mut updates).await.unwrap(), &transcript, vad_id);
            let StreamingJob { input, mut updates } = service
                .start_windowed(WindowConfig::interactive_for(FULL_WINDOW).unwrap())
                .await
                .unwrap();
            input.cancel().await.unwrap();
            assert!(matches!(
                updates.next().await,
                Err(ServiceError::Cancelled { .. })
            ));
            assert!(updates.next().await.unwrap().is_none());
            service.shutdown().await.unwrap();
        });
    }
    #[test]
    fn busy_and_invalid_audio_do_not_replace_or_poison_the_active_job() {
        block_on(async {
            let service = markers(ServiceConfig {
                input_capacity: 4,
                update_capacity: 8,
            })
            .await;
            let StreamingJob {
                mut input,
                mut updates,
            } = service.start(AsrConfig::interactive()).await.unwrap();
            assert!(
                matches!(service.start(AsrConfig::interactive()).await, Err(ServiceError::Busy { job_id }) if job_id == input.job_id())
            );
            assert!(matches!(
                input.try_push_audio(&[f32::NAN]),
                Err(ServiceError::InvalidAudio {
                    non_finite: true,
                    ..
                })
            ));
            assert!(matches!(
                input.push_audio(&vec![0.0; 16001]).await,
                Err(ServiceError::InvalidAudio { samples: 16001, .. })
            ));
            input.push_audio(&audio(7)).await.unwrap();
            assert_eq!(input.finish().await.unwrap().text, "7");
            assert_eq!(
                collect(&mut updates)
                    .await
                    .unwrap()
                    .last()
                    .unwrap()
                    .update
                    .kind,
                UpdateKind::Final
            );
            service.shutdown().await.unwrap();
        });
    }
    #[test]
    fn idle_cancellation_is_an_error_without_a_success_terminal_and_allows_next_job() {
        block_on(async {
            let service = markers(ServiceConfig {
                input_capacity: 1,
                update_capacity: 1,
            })
            .await;
            let StreamingJob { input, mut updates } =
                service.start(AsrConfig::interactive()).await.unwrap();
            let id = input.job_id();
            input.cancel().await.unwrap();
            assert!(
                matches!(updates.next().await, Err(ServiceError::Cancelled { job_id }) if job_id == id)
            );
            assert!(updates.next().await.unwrap().is_none());
            let StreamingJob { input, mut updates } =
                service.start(AsrConfig::interactive()).await.unwrap();
            input.finish().await.unwrap();
            assert_eq!(collect(&mut updates).await.unwrap().len(), 1);
            service.shutdown().await.unwrap();
        });
    }
    #[test]
    fn output_overload_keeps_queued_edits_and_reserves_a_terminal_error_slot() {
        block_on(async {
            let service = markers(ServiceConfig {
                input_capacity: 8,
                update_capacity: 1,
            })
            .await;
            let StreamingJob {
                mut input,
                mut updates,
            } = service.start(AsrConfig::interactive()).await.unwrap();
            for token in 1..=4 {
                input.push_audio(&audio(token)).await.unwrap();
            }
            assert!(matches!(
                input.finish().await,
                Err(ServiceError::OutputFull { capacity: 1, .. })
            ));
            let first = updates.next().await.unwrap().unwrap();
            assert_eq!(first.update.sequence, 1);
            assert_ne!(first.update.kind, UpdateKind::Final);
            assert!(matches!(
                updates.next().await,
                Err(ServiceError::OutputFull { capacity: 1, .. })
            ));
            assert!(updates.next().await.unwrap().is_none());
            service.shutdown().await.unwrap();
        });
    }

    struct GateRuntime {
        markers: Markers,
        entered: mpsc::Sender<()>,
        release: mpsc::Receiver<()>,
    }
    struct GateSession<'a> {
        inner: StreamingSession<'a>,
        entered: &'a mpsc::Sender<()>,
        release: &'a mpsc::Receiver<()>,
        first: bool,
    }
    impl Session for GateSession<'_> {
        fn push(
            &mut self,
            samples: &[f32],
            cancellation: &Cancellation,
        ) -> crate::Result<Vec<AsrUpdate>> {
            if self.first {
                self.first = false;
                self.entered.send(()).unwrap();
                self.release.recv_timeout(Duration::from_secs(5)).unwrap();
            }
            self.inner.push_audio(samples, cancellation)
        }
        fn finish(&mut self, cancellation: &Cancellation) -> crate::Result<(AsrResult, AsrUpdate)> {
            self.inner.finish_result(cancellation)
        }
    }
    impl Runtime for GateRuntime {
        fn limits(&self) -> InputLimits {
            FULL_WINDOW
        }
        fn start(&self, config: SessionConfig) -> crate::Result<Box<dyn Session + '_>> {
            Ok(Box::new(GateSession {
                inner: configured_session(&self.markers, config)?,
                entered: &self.entered,
                release: &self.release,
                first: true,
            }))
        }
    }
    async fn gated(config: ServiceConfig) -> (AsrService, mpsc::Receiver<()>, mpsc::Sender<()>) {
        let (entered, observed) = mpsc::channel();
        let (release, released) = mpsc::channel();
        let service = AsrService::load_runtime(
            move || {
                Ok(GateRuntime {
                    markers: Markers::new(),
                    entered,
                    release: released,
                })
            },
            config,
        )
        .await
        .unwrap();
        (service, observed, release)
    }
    #[test]
    fn full_input_queue_rejects_audio_explicitly_and_retry_preserves_it() {
        block_on(async {
            let (service, entered, release) = gated(ServiceConfig {
                input_capacity: 1,
                update_capacity: 16,
            })
            .await;
            let StreamingJob {
                mut input,
                mut updates,
            } = service.start(AsrConfig::interactive()).await.unwrap();
            input.push_audio(&audio(1)).await.unwrap();
            entered.recv_timeout(Duration::from_secs(5)).unwrap();
            input.try_push_audio(&audio(2)).unwrap();
            let third = audio(3);
            assert!(matches!(
                input.try_push_audio(&third),
                Err(ServiceError::InputFull { capacity: 1 })
            ));
            release.send(()).unwrap();
            input.push_audio(&third).await.unwrap();
            let transcript = input.finish().await.unwrap();
            assert_eq!(transcript.text, "1 2 3");
            assert_replay(
                &collect(&mut updates).await.unwrap(),
                &transcript,
                updates.job_id(),
            );
            service.shutdown().await.unwrap();
        });
    }
    #[test]
    fn cancel_interrupts_inflight_work_and_retires_already_queued_audio() {
        block_on(async {
            let (service, entered, release) = gated(ServiceConfig {
                input_capacity: 1,
                update_capacity: 8,
            })
            .await;
            let StreamingJob {
                mut input,
                mut updates,
            } = service.start(AsrConfig::interactive()).await.unwrap();
            input.push_audio(&audio(1)).await.unwrap();
            entered.recv_timeout(Duration::from_secs(5)).unwrap();
            input.try_push_audio(&audio(2)).unwrap();
            let mut cancellation = Box::pin(input.cancel());
            assert!(poll_once(cancellation.as_mut()).await.is_none());
            release.send(()).unwrap();
            cancellation.await.unwrap();
            assert!(matches!(
                updates.next().await,
                Err(ServiceError::Cancelled { .. })
            ));
            assert!(updates.next().await.unwrap().is_none());
            service.shutdown().await.unwrap();
        });
    }
    #[test]
    fn shutdown_waits_for_inflight_work_without_blocking_its_future() {
        block_on(async {
            let (service, entered, release) = gated(ServiceConfig {
                input_capacity: 1,
                update_capacity: 8,
            })
            .await;
            let StreamingJob {
                mut input,
                mut updates,
            } = service.start(AsrConfig::interactive()).await.unwrap();
            input.push_audio(&audio(1)).await.unwrap();
            entered.recv_timeout(Duration::from_secs(5)).unwrap();
            let mut shutdown = Box::pin(service.shutdown());
            assert!(poll_once(shutdown.as_mut()).await.is_none());
            release.send(()).unwrap();
            shutdown.await.unwrap();
            while let Some(event) = updates.next().await.transpose() {
                match event {
                    Ok(event) => assert_ne!(event.update.kind, UpdateKind::Final),
                    Err(ServiceError::Shutdown) => break,
                    Err(error) => panic!("unexpected shutdown error: {error}"),
                }
            }
            assert!(matches!(input.finish().await, Err(ServiceError::Shutdown)));
        });
    }
    #[test]
    fn dropping_input_or_consumer_cancels_idle_jobs() {
        block_on(async {
            let service = markers(ServiceConfig {
                input_capacity: 1,
                update_capacity: 2,
            })
            .await;
            let StreamingJob { input, mut updates } =
                service.start(AsrConfig::interactive()).await.unwrap();
            drop(input);
            assert!(matches!(
                updates.next().await,
                Err(ServiceError::Cancelled { .. })
            ));
            let StreamingJob { input, updates } =
                service.start(AsrConfig::interactive()).await.unwrap();
            drop(updates);
            assert!(matches!(
                input.finish().await,
                Err(ServiceError::Cancelled { .. })
            ));
            service.shutdown().await.unwrap();
        });
    }

    struct BrokenRuntime;
    struct BrokenSession;
    impl Runtime for BrokenRuntime {
        fn limits(&self) -> InputLimits {
            FULL_WINDOW
        }
        fn start(&self, _: SessionConfig) -> crate::Result<Box<dyn Session + '_>> {
            Ok(Box::new(BrokenSession))
        }
    }
    impl Session for BrokenSession {
        fn push(&mut self, _: &[f32], _: &Cancellation) -> crate::Result<Vec<AsrUpdate>> {
            Err(crate::Error::Invalid(
                "injected inference failure".to_owned(),
            ))
        }
        fn finish(&mut self, _: &Cancellation) -> crate::Result<(AsrResult, AsrUpdate)> {
            panic!("failed job must not be finalized")
        }
    }
    #[test]
    fn model_load_and_inference_failures_are_delivered_with_context() {
        block_on(async {
            let config = ServiceConfig {
                input_capacity: 2,
                update_capacity: 2,
            };
            assert!(matches!(
                AsrService::load_runtime(
                    || Err::<MarkerRuntime, _>(crate::Error::Invalid(
                        "injected load failure".to_owned()
                    )),
                    config
                )
                .await,
                Err(ServiceError::Asr {
                    operation: "load models",
                    job_id: None,
                    ..
                })
            ));
            let service = AsrService::load_runtime(|| Ok(BrokenRuntime), config)
                .await
                .unwrap();
            let StreamingJob {
                mut input,
                mut updates,
            } = service.start(AsrConfig::interactive()).await.unwrap();
            let id = input.job_id();
            input.push_audio(&audio(1)).await.unwrap();
            assert!(
                matches!(updates.next().await, Err(ServiceError::Asr { operation: "push audio", job_id: Some(job_id), .. }) if job_id == id)
            );
            assert!(matches!(
                input.finish().await,
                Err(ServiceError::Asr {
                    operation: "push audio",
                    ..
                })
            ));
            assert!(updates.next().await.unwrap().is_none());
            service.shutdown().await.unwrap();
        });
    }

    struct AffineRuntime {
        recognizer: std::rc::Rc<AffineRecognizer>,
    }
    struct AffineRecognizer {
        markers: Markers,
        calls: Arc<Mutex<Vec<thread::ThreadId>>>,
        dropped: mpsc::Sender<thread::ThreadId>,
    }
    impl WindowRecognizer for AffineRecognizer {
        fn input_limits(&self) -> InputLimits {
            FULL_WINDOW
        }
        fn infer(
            &self,
            samples: &[f32],
            cancellation: &Cancellation,
        ) -> crate::Result<Vec<TimedToken>> {
            self.calls.lock().unwrap().push(thread::current().id());
            self.markers.infer(samples, cancellation)
        }
        fn token_piece(&self, id: usize) -> crate::Result<crate::result::TokenPiece> {
            self.markers.token_piece(id)
        }
        fn render(&self, tokens: &[TimedToken]) -> crate::Result<String> {
            self.markers.render(tokens)
        }
    }
    impl Drop for AffineRuntime {
        fn drop(&mut self) {
            self.recognizer
                .dropped
                .send(thread::current().id())
                .unwrap();
        }
    }
    impl Runtime for AffineRuntime {
        fn limits(&self) -> InputLimits {
            FULL_WINDOW
        }
        fn start(&self, config: SessionConfig) -> crate::Result<Box<dyn Session + '_>> {
            Ok(Box::new(configured_session(
                self.recognizer.as_ref(),
                config,
            )?))
        }
    }
    #[test]
    fn non_send_models_are_loaded_used_reused_and_dropped_on_one_worker_thread() {
        block_on(async {
            let caller = thread::current().id();
            let (loaded, load_thread) = mpsc::channel();
            let (dropped, drop_thread) = mpsc::channel();
            let calls = Arc::new(Mutex::new(Vec::new()));
            let worker_calls = calls.clone();
            let service = AsrService::load_runtime(
                move || {
                    loaded.send(thread::current().id()).unwrap();
                    Ok(AffineRuntime {
                        recognizer: std::rc::Rc::new(AffineRecognizer {
                            markers: Markers::new(),
                            calls: worker_calls,
                            dropped,
                        }),
                    })
                },
                ServiceConfig {
                    input_capacity: 2,
                    update_capacity: 8,
                },
            )
            .await
            .unwrap();
            let owner = load_thread.recv_timeout(Duration::from_secs(5)).unwrap();
            assert_ne!(owner, caller);
            for _ in 0..2 {
                let StreamingJob {
                    mut input,
                    mut updates,
                } = service.start(AsrConfig::interactive()).await.unwrap();
                input.push_audio(&audio(1)).await.unwrap();
                input.finish().await.unwrap();
                collect(&mut updates).await.unwrap();
            }
            service.shutdown().await.unwrap();
            assert_eq!(
                drop_thread.recv_timeout(Duration::from_secs(5)).unwrap(),
                owner
            );
            let calls = calls.lock().unwrap();
            assert!(calls.len() >= 2 && calls.iter().all(|id| *id == owner));
        });
    }
    #[test]
    fn cancelling_model_loading_releases_the_late_loaded_runtime() {
        block_on(async {
            let (entered, entry) = mpsc::channel();
            let (release, released) = mpsc::channel();
            let (dropped, drop_thread) = mpsc::channel();
            let mut loading = Box::pin(AsrService::load_runtime(
                move || {
                    entered.send(()).unwrap();
                    released.recv_timeout(Duration::from_secs(5)).unwrap();
                    Ok(AffineRuntime {
                        recognizer: std::rc::Rc::new(AffineRecognizer {
                            markers: Markers::new(),
                            calls: Arc::new(Mutex::new(Vec::new())),
                            dropped,
                        }),
                    })
                },
                ServiceConfig {
                    input_capacity: 1,
                    update_capacity: 1,
                },
            ));
            assert!(poll_once(loading.as_mut()).await.is_none());
            entry.recv_timeout(Duration::from_secs(5)).unwrap();
            drop(loading);
            release.send(()).unwrap();
            drop_thread.recv_timeout(Duration::from_secs(5)).unwrap();
        });
    }
    #[test]
    fn invalid_capacity_and_job_config_are_rejected_without_a_worker_panic() {
        block_on(async {
            for config in [
                ServiceConfig {
                    input_capacity: 0,
                    update_capacity: 1,
                },
                ServiceConfig {
                    input_capacity: 1,
                    update_capacity: 0,
                },
                ServiceConfig {
                    input_capacity: 1,
                    update_capacity: usize::MAX,
                },
            ] {
                assert!(matches!(
                    AsrService::load_runtime(
                        || Ok(MarkerRuntime {
                            markers: Markers::new()
                        }),
                        config
                    )
                    .await,
                    Err(ServiceError::Configuration(_))
                ));
            }
            let service = markers(ServiceConfig {
                input_capacity: 1,
                update_capacity: 1,
            })
            .await;
            let mut config = AsrConfig::interactive();
            config.center_samples = 0;
            assert!(matches!(
                service.start(config).await,
                Err(ServiceError::Asr {
                    operation: "validate job configuration",
                    ..
                })
            ));
            service.shutdown().await.unwrap();
        });
    }

    struct VadProbe {
        calls: Arc<Mutex<Vec<(thread::ThreadId, &'static str)>>>,
        entered: mpsc::Receiver<()>,
        release: mpsc::Sender<()>,
        dropped: mpsc::Receiver<thread::ThreadId>,
        owner: thread::ThreadId,
    }
    struct TestVad {
        // An intentionally non-Send runtime must never leave its owner thread.
        owner: std::rc::Rc<thread::ThreadId>,
        calls: Arc<Mutex<Vec<(thread::ThreadId, &'static str)>>>,
        entered: mpsc::Sender<()>,
        release: mpsc::Receiver<()>,
        dropped: mpsc::Sender<thread::ThreadId>,
    }
    impl TestVad {
        fn record(&self, operation: &'static str) {
            assert_eq!(*self.owner, thread::current().id());
            self.calls
                .lock()
                .unwrap()
                .push((thread::current().id(), operation));
        }
        fn analysis(&self, audio: &[f32]) -> VadAnalysis {
            VadAnalysis {
                frames: Vec::new(),
                audio_samples: audio.len() as u64,
                output_state: VadState::initial(),
            }
        }
    }
    impl Drop for TestVad {
        fn drop(&mut self) {
            self.dropped.send(thread::current().id()).unwrap();
        }
    }
    impl VadRuntime for TestVad {
        fn process_chunk(
            &self,
            audio: &[f32],
            state: &VadState,
            threshold: f32,
            token: &Cancellation,
        ) -> crate::Result<VadResult> {
            self.record("chunk");
            token.check()?;
            let mut output_state = state.clone();
            output_state.context[0] = audio[0];
            Ok(VadResult {
                probability: threshold,
                is_voice_active: true,
                processing_seconds: 0.0,
                output_state,
            })
        }
        fn process_samples(
            &self,
            audio: &[f32],
            _: f32,
            token: &Cancellation,
        ) -> crate::Result<VadAnalysis> {
            self.record("samples");
            token.check()?;
            if audio == [99.0] {
                self.entered.send(()).unwrap();
                self.release.recv_timeout(Duration::from_secs(5)).unwrap();
            }
            if audio == [-1.0] {
                return Err(crate::Error::Invalid("test inference failure".to_owned()));
            }
            assert_ne!(audio, [-2.0], "test model panic");
            Ok(self.analysis(audio))
        }
        fn process_wav(
            &self,
            path: &Path,
            _: f32,
            token: &Cancellation,
        ) -> crate::Result<VadAnalysis> {
            self.record("wav");
            token.check()?;
            assert_eq!(path, Path::new("test.wav"));
            Ok(self.analysis(&[1.0]))
        }
        fn process_file(
            &self,
            path: &Path,
            threshold: f32,
            cancellation: &Cancellation,
        ) -> crate::Result<VadAnalysis> {
            self.process_wav(path, threshold, cancellation)
        }
        fn segment_file(
            &self,
            path: &Path,
            config: &VadSegmentationConfig,
            cancellation: &Cancellation,
        ) -> crate::Result<Vec<VadSegment>> {
            self.segment_wav(path, config, cancellation)
        }
        fn process_streaming_chunk(
            &self,
            audio: &[f32],
            state: &VadStreamState,
            config: &VadSegmentationConfig,
            token: &Cancellation,
        ) -> crate::Result<VadStreamResult> {
            self.record("stream");
            token.check()?;
            assert_eq!(config.positive_threshold, 0.85);
            let mut state = state.clone();
            state.processed_samples += audio.len() as u64;
            Ok(VadStreamResult {
                state,
                event: None,
                probability: 0.5,
                processing_seconds: 0.0,
            })
        }
        fn segment_samples(
            &self,
            audio: &[f32],
            config: &VadSegmentationConfig,
            token: &Cancellation,
        ) -> crate::Result<Vec<VadSegment>> {
            self.record("segment samples");
            token.check()?;
            assert_eq!(config.speech_padding_samples, 1600);
            Ok(vec![VadSegment {
                start_sample: 0,
                end_sample: audio.len() as u64,
            }])
        }
        fn segment_wav(
            &self,
            path: &Path,
            config: &VadSegmentationConfig,
            token: &Cancellation,
        ) -> crate::Result<Vec<VadSegment>> {
            self.record("segment wav");
            token.check()?;
            assert_eq!(config.speech_padding_samples, 1600);
            assert_eq!(path, Path::new("test.wav"));
            Ok(vec![VadSegment {
                start_sample: 0,
                end_sample: 7,
            }])
        }
    }
    async fn test_vad(capacity: usize) -> (VadService, VadProbe) {
        let (entered, entry) = mpsc::channel();
        let (release, released) = mpsc::channel();
        let (dropped, drop_thread) = mpsc::channel();
        let (loaded, load_thread) = mpsc::channel();
        let calls = Arc::new(Mutex::new(Vec::new()));
        let worker_calls = calls.clone();
        let service = VadService::load_runtime(
            move || {
                let owner = thread::current().id();
                loaded.send(owner).unwrap();
                Ok(TestVad {
                    owner: std::rc::Rc::new(owner),
                    calls: worker_calls,
                    entered,
                    release: released,
                    dropped,
                })
            },
            capacity,
        )
        .await
        .unwrap();
        (
            service,
            VadProbe {
                calls,
                entered: entry,
                release,
                dropped: drop_thread,
                owner: load_thread.recv_timeout(Duration::from_secs(5)).unwrap(),
            },
        )
    }
    fn owned_future<T, F: Future<Output = T> + Send + 'static>(future: F) -> F {
        future
    }

    #[test]
    fn vad_reuses_one_non_send_runtime_for_all_operations() {
        block_on(async {
            let (service, probe) = test_vad(2).await;
            assert_ne!(probe.owner, thread::current().id());
            let token = Cancellation::new();
            let initial = VadState::initial();
            let result = owned_future(service.process_chunk(
                vec![0.25],
                initial.clone(),
                0.7,
                token.clone(),
            ))
            .await
            .unwrap();
            assert_eq!(result.probability, 0.7);
            assert_eq!(result.output_state.context[0], 0.25);
            assert_eq!(initial, VadState::initial());
            assert_eq!(
                owned_future(service.process_samples(vec![0.0; 7], 0.85, token.clone()))
                    .await
                    .unwrap()
                    .audio_samples,
                7
            );
            assert_eq!(
                owned_future(service.process_wav("test.wav".into(), 0.85, token.clone()))
                    .await
                    .unwrap()
                    .audio_samples,
                1
            );
            let stream = VadStreamState::initial();
            let result = owned_future(service.process_streaming_chunk(
                vec![0.0; 3],
                stream.clone(),
                VadSegmentationConfig::standard(),
                token.clone(),
            ))
            .await
            .unwrap();
            assert_eq!(result.state.processed_samples, 3);
            assert_eq!(stream.processed_samples, 0);
            assert_eq!(
                owned_future(service.segment_samples(
                    vec![0.0; 5],
                    VadSegmentationConfig::standard(),
                    token.clone()
                ))
                .await
                .unwrap()[0]
                    .end_sample,
                5
            );
            assert_eq!(
                owned_future(service.segment_wav(
                    "test.wav".into(),
                    VadSegmentationConfig::standard(),
                    token.clone()
                ))
                .await
                .unwrap()[0]
                    .end_sample,
                7
            );
            assert_eq!(
                owned_future(service.process_file("test.wav".into(), 0.85, token.clone()))
                    .await
                    .unwrap()
                    .audio_samples,
                1
            );
            assert_eq!(
                owned_future(service.segment_file(
                    "test.wav".into(),
                    VadSegmentationConfig::standard(),
                    token.clone()
                ))
                .await
                .unwrap()[0]
                    .end_sample,
                7
            );
            token.check().unwrap();
            service.shutdown().await.unwrap();
            assert_eq!(
                probe.dropped.recv_timeout(Duration::from_secs(5)).unwrap(),
                probe.owner
            );
            let calls = probe.calls.lock().unwrap();
            assert_eq!(calls.len(), 8);
            assert!(calls.iter().all(|(thread, _)| *thread == probe.owner));
        });
    }
    #[test]
    fn vad_queue_backpressure_and_dropped_requests_preserve_sibling_work() {
        block_on(async {
            let (service, probe) = test_vad(1).await;
            let token = Cancellation::new();
            let mut first = Box::pin(service.process_samples(vec![99.0], 0.85, token.clone()));
            assert!(poll_once(first.as_mut()).await.is_none());
            probe.entered.recv_timeout(Duration::from_secs(5)).unwrap();
            let mut discarded = Box::pin(service.process_samples(vec![1.0], 0.85, token.clone()));
            assert!(poll_once(discarded.as_mut()).await.is_none());
            assert_eq!(service.worker.commands.len(), 1);
            let mut waiting = Box::pin(service.process_samples(vec![2.0; 2], 0.85, token.clone()));
            assert!(poll_once(waiting.as_mut()).await.is_none());
            assert_eq!(service.worker.commands.len(), 1);
            drop(discarded);
            token.check().unwrap();
            probe.release.send(()).unwrap();
            assert_eq!(first.await.unwrap().audio_samples, 1);
            assert_eq!(waiting.await.unwrap().audio_samples, 2);
            service.shutdown().await.unwrap();
            assert_eq!(probe.calls.lock().unwrap().len(), 2);
        });
    }
    #[test]
    fn dropping_an_active_vad_future_cancels_its_token_without_poisoning_the_model() {
        block_on(async {
            let (service, probe) = test_vad(1).await;
            let token = Cancellation::new();
            let mut active = Box::pin(service.process_samples(vec![99.0], 0.85, token.clone()));
            assert!(poll_once(active.as_mut()).await.is_none());
            probe.entered.recv_timeout(Duration::from_secs(5)).unwrap();
            drop(active);
            token.check().unwrap();
            probe.release.send(()).unwrap();
            service
                .process_samples(vec![1.0], 0.85, token)
                .await
                .unwrap();
            service.shutdown().await.unwrap();
            assert_eq!(probe.calls.lock().unwrap().len(), 2);
        });
    }
    #[test]
    fn vad_caller_cancellation_cancels_only_requests_using_that_token() {
        block_on(async {
            let (service, probe) = test_vad(2).await;
            let token = Cancellation::new();
            let mut active = Box::pin(service.process_samples(vec![99.0], 0.85, token.clone()));
            assert!(poll_once(active.as_mut()).await.is_none());
            probe.entered.recv_timeout(Duration::from_secs(5)).unwrap();
            let mut queued = Box::pin(service.process_samples(vec![1.0], 0.85, token.clone()));
            assert!(poll_once(queued.as_mut()).await.is_none());
            token.cancel();
            probe.release.send(()).unwrap();
            assert!(matches!(active.await, Err(ServiceError::Cancelled { .. })));
            assert!(matches!(queued.await, Err(ServiceError::Cancelled { .. })));
            assert!(matches!(
                service.process_samples(vec![1.0], 0.85, token).await,
                Err(ServiceError::Cancelled { .. })
            ));
            service
                .process_samples(vec![2.0], 0.85, Cancellation::new())
                .await
                .unwrap();
            service.shutdown().await.unwrap();
            assert_eq!(probe.calls.lock().unwrap().len(), 2);
        });
    }
    #[test]
    fn vad_shutdown_cancels_active_queued_and_waiting_requests_and_joins_model_owner() {
        block_on(async {
            let (service, probe) = test_vad(1).await;
            let mut active =
                Box::pin(service.process_samples(vec![99.0], 0.85, Cancellation::new()));
            assert!(poll_once(active.as_mut()).await.is_none());
            probe.entered.recv_timeout(Duration::from_secs(5)).unwrap();
            let mut queued =
                Box::pin(service.process_samples(vec![1.0], 0.85, Cancellation::new()));
            assert!(poll_once(queued.as_mut()).await.is_none());
            let mut waiting =
                Box::pin(service.process_samples(vec![2.0], 0.85, Cancellation::new()));
            assert!(poll_once(waiting.as_mut()).await.is_none());
            let mut shutdown = Box::pin(service.shutdown());
            assert!(poll_once(shutdown.as_mut()).await.is_none());
            probe.release.send(()).unwrap();
            assert!(matches!(active.await, Err(ServiceError::Shutdown)));
            assert!(matches!(queued.await, Err(ServiceError::Shutdown)));
            assert!(matches!(waiting.await, Err(ServiceError::Shutdown)));
            shutdown.await.unwrap();
            assert_eq!(probe.calls.lock().unwrap().len(), 1);
            assert_eq!(
                probe.dropped.recv_timeout(Duration::from_secs(5)).unwrap(),
                probe.owner
            );
        });
    }
    #[test]
    fn dropping_vad_service_never_blocks_on_active_inference() {
        block_on(async {
            let (service, probe) = test_vad(1).await;
            let mut active =
                Box::pin(service.process_samples(vec![99.0], 0.85, Cancellation::new()));
            assert!(poll_once(active.as_mut()).await.is_none());
            probe.entered.recv_timeout(Duration::from_secs(5)).unwrap();
            drop(service);
            probe.release.send(()).unwrap();
            assert!(matches!(active.await, Err(ServiceError::Shutdown)));
            assert_eq!(
                probe.dropped.recv_timeout(Duration::from_secs(5)).unwrap(),
                probe.owner
            );
        });
    }
    #[test]
    fn vad_errors_keep_context_and_worker_panics_never_become_success() {
        block_on(async {
            assert!(matches!(
                VadService::load_runtime(
                    || -> crate::Result<TestVad> {
                        panic!("capacity must be validated before loading")
                    },
                    0
                )
                .await,
                Err(ServiceError::Configuration(_))
            ));
            assert!(matches!(
                VadService::load_runtime(
                    || -> crate::Result<TestVad> {
                        Err(crate::Error::Invalid("test loading failure".to_owned()))
                    },
                    1
                )
                .await,
                Err(ServiceError::Vad {
                    operation: "load model",
                    job_id: None,
                    ..
                })
            ));
            let (service, _probe) = test_vad(1).await;
            assert!(matches!(
                service
                    .process_samples(vec![-1.0], 0.85, Cancellation::new())
                    .await,
                Err(ServiceError::Vad {
                    operation: "process samples",
                    job_id: Some(_),
                    ..
                })
            ));
            service
                .process_samples(vec![1.0], 0.85, Cancellation::new())
                .await
                .unwrap();
            assert!(matches!(
                service
                    .process_samples(vec![-2.0], 0.85, Cancellation::new())
                    .await,
                Err(ServiceError::WorkerStopped)
            ));
            assert!(
                matches!(service.shutdown().await, Err(ServiceError::WorkerPanic(message)) if message.contains("test model panic"))
            );
        });
    }
    #[test]
    fn linked_cancellation_never_cancels_parents_or_siblings() {
        let caller = Cancellation::new();
        let shutdown = Cancellation::new();
        let first = Cancellation::linked(&caller, &shutdown);
        let second = Cancellation::linked(&caller, &shutdown);
        first.cancel();
        assert!(first.check().is_err());
        caller.check().unwrap();
        shutdown.check().unwrap();
        second.check().unwrap();
        caller.cancel();
        assert!(second.check().is_err());
        shutdown.check().unwrap();
        let caller = Cancellation::new();
        let linked = Cancellation::linked(&caller, &shutdown);
        shutdown.cancel();
        assert!(linked.check().is_err());
        caller.check().unwrap();
    }
    #[test]
    fn async_finish_returns_shared_metadata_without_leaking_between_jobs() {
        block_on(async {
            let service = markers(ServiceConfig {
                input_capacity: 2,
                update_capacity: 2,
            })
            .await;
            for value in [7usize, 9] {
                let StreamingJob {
                    mut input,
                    mut updates,
                } = service
                    .start_windowed(WindowConfig::interactive_for(FULL_WINDOW).unwrap())
                    .await
                    .unwrap();
                let id = input.job_id();
                input.push_audio(&audio(value)).await.unwrap();
                let result = input.finish_result().await.unwrap();
                assert_eq!(result.text, value.to_string());
                assert_eq!(result.tokens[0].piece.text, value.to_string());
                assert_eq!(result.metadata.inference.calls, 1);
                assert_eq!(result.metadata.audio_samples, 16000);
                assert!(result.metadata.inference.stages.is_none());
                assert_replay(
                    &collect(&mut updates).await.unwrap(),
                    &result.into_transcript(),
                    id,
                );
            }
            service.shutdown().await.unwrap();
        });
    }
}