forge_workspace/

bench.rs

//! Benching a model on this machine's own material: which models a run can
//! load, the config each one runs under, and the state and results the
//! page reads.
//!
//! The corpus and the runner are `forge-dictate`'s; this module is the
//! workspace's half - it knows the roles, the installed set and the store.

use std::path::Path;
use std::sync::Arc;
use std::sync::atomic::Ordering;

use forge_dictate::ModelSpec;

use crate::dictate::SetRecording;
use crate::{DispatchError, Workspace};

/// Which slot a bench target runs in, in the names the wire uses.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum BenchRole {
    Transcribing,
    Cleanup,
}

/// One model a bench can load.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct BenchTarget {
    pub file: String,
    pub role: BenchRole,
    /// Whether `forge.toml` pins the role this target would take - the
    /// page offers no activation for one, and a bench still runs.
    pub pinned: bool,
}

/// Where the last bench got to, as the page draws it.
#[derive(Debug, Clone, Default, PartialEq, serde::Serialize, serde::Deserialize)]
#[serde(tag = "state", rename_all = "snake_case")]
pub enum BenchState {
    #[default]
    Idle,
    /// `clip` of `clips` is counting clips done, and `so_far` is the share
    /// of the compared ones that agreed word-for-word so far, when any
    /// clip carried something to compare.
    Running {
        target: BenchTarget,
        tier: forge_dictate::bench::Tier,
        clip: usize,
        clips: usize,
        so_far: Option<f64>,
    },
    /// The run stopped with a reason. Nothing was saved.
    Failed { target: BenchTarget, reason: String },
}

/// What one finished bench measured, kept under what it was about.
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct BenchResult {
    pub target: BenchTarget,
    pub tier: forge_dictate::bench::Tier,
    pub metrics: forge_dictate::bench::Metrics,
    /// RFC 3339.
    pub at: String,
    /// The corpus it ran over, so two numbers are only compared when these
    /// agree.
    pub corpus: forge_dictate::bench::CorpusId,
}

/// Where the speaking bench got to, as the page draws it.
#[derive(Debug, Clone, Default, PartialEq, serde::Serialize, serde::Deserialize)]
#[serde(tag = "state", rename_all = "snake_case")]
pub enum SpeakingBenchState {
    #[default]
    Idle,
    /// `clip` of `clips` transcripts have been spoken, heard back and
    /// scored.
    Running { clip: usize, clips: usize },
    /// The run stopped with a reason. Nothing was saved.
    Failed { reason: String },
}

/// What one speaking bench measured, kept under the model and corpus it ran.
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct SpeakingBenchResult {
    /// The speaking model's file, as the page names it.
    pub file: String,
    pub metrics: forge_tts::bench::TtsBenchResult,
    /// RFC 3339.
    pub at: String,
    /// The corpus it ran over, on the same terms as the words bench's.
    pub corpus: forge_dictate::bench::CorpusId,
}

/// What the live roles run on this machine, for a bench that runs outside
/// forge.
#[cfg(any(test, feature = "testing"))]
#[derive(Debug)]
pub struct LiveRoles {
    /// The model each role runs, in role order, with where the choice came
    /// from.
    pub models: Vec<(crate::dictate::DictateRole, crate::install::ActiveModel)>,
    /// The engine configuration those roles describe, as the boot builds it.
    pub config: forge_dictate::Config,
    /// Where the weights live: the configured directory, else the platform
    /// cache directory.
    pub models_dir: std::path::PathBuf,
}

/// A copy of a store another process holds, removed when it goes.
#[cfg(any(test, feature = "testing"))]
struct StoreCopy(std::path::PathBuf);

#[cfg(any(test, feature = "testing"))]
impl StoreCopy {
    fn take(live: &Path) -> Result<Self, String> {
        static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
        let path = std::env::temp_dir().join(format!(
            "forge-store-{}-{}.redb",
            std::process::id(),
            NEXT.fetch_add(1, Ordering::Relaxed)
        ));
        std::fs::copy(live, &path).map_err(|error| {
            format!(
                "the store at {} could not be read ({error}); has forge run with this config dir \
                 yet?",
                live.display()
            )
        })?;
        Ok(Self(path))
    }
}

#[cfg(any(test, feature = "testing"))]
impl Drop for StoreCopy {
    fn drop(&mut self) {
        let _ = std::fs::remove_file(&self.0);
    }
}

/// The models the live roles resolve to, read by a process that is not
/// forge.
///
/// redb takes an exclusive lock, so a running forge holds the live store; the
/// rows are read off a copy. On macOS the copy is an APFS clone, so it is a
/// point-in-time snapshot; a plain byte copy racing a commit can land on the
/// previous one, so these answers are the store as of the copy rather than
/// what a running engine holds mid-activation. A role whose config key names
/// a variant this machine has not installed is an error rather than a feed
/// lookup: the caller is a bench, and a bench fetches nothing.
#[cfg(any(test, feature = "testing"))]
pub fn resolved_live_roles(config_dir: &Path, app_support: &Path) -> Result<LiveRoles, String> {
    let loaded = crate::config::load_from_dir(config_dir)
        .map_err(|error| format!("forge.toml could not be read: {error}"))?;
    let settings = &loaded.dictate;
    if !settings.enabled {
        return Err("[dictate] enabled = false: dictation runs no role on this machine, so the \
                    bench has nothing to measure"
            .to_owned());
    }
    let Some(models_dir) = settings.models_dir() else {
        return Err("no models directory resolves: [dictate] models_dir is unset and there is no \
                    platform cache directory"
            .to_owned());
    };

    let live = crate::store::path_for(app_support, config_dir);
    let copy = StoreCopy::take(&live)?;
    let db = crate::store::Db::open(&copy.0).map_err(|error| {
        format!(
            "the store at {} could not be read through a copy at {}: {error}",
            live.display(),
            copy.0.display()
        )
    })?;
    let installed = crate::store::dictate_models::installed(&db).map_err(|error| {
        format!("the installed models could not be read from {}: {error}", live.display())
    })?;

    let mut models = Vec::new();
    for (role, pin) in settings.model_specs() {
        let configured = match role {
            crate::dictate::DictateRole::Transcribing => settings.transcribe_model.as_deref(),
            crate::dictate::DictateRole::Normalization => settings.cleanup_model.as_deref(),
            // The bench walks the dictate roles; speaking has its own run.
            crate::dictate::DictateRole::Speaking => None,
        };
        // The pick is read only when the key does not decide, as the boot
        // does: a row that cannot matter must not fail the bench.
        let pick = if configured.is_none() {
            crate::store::dictate_models::active(&db, role).map_err(|error| {
                format!(
                    "the {} row could not be read from {}: {error}",
                    role.label(),
                    live.display()
                )
            })?
        } else {
            None
        };
        match crate::install::resolve_role(role, pin, configured, &installed, pick) {
            crate::install::RoleResolution::Chosen(model) => models.push((role, model)),
            crate::install::RoleResolution::NeedsFeed { key, variant } => {
                return Err(format!(
                    "[dictate] {key} names {variant}, which this machine has not installed; the \
                     bench measures what a live role runs and never fetches"
                ));
            }
        }
    }

    let config = crate::dictate::preflight_config(settings, &models);
    Ok(LiveRoles { models, config, models_dir })
}

/// How many saved results the page reads.
const RESULTS_SHOWN: usize = 50;

/// One recording of the read-aloud passage, as the page lists it: what the
/// row draws and what the bench scores, one row per clip.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct ReadAloudRecording {
    /// The take's own directory name, which is what the page deletes by.
    pub id: String,
    pub duration_ms: u64,
    /// The wav's own byte length, which is what the recording costs on disk.
    pub bytes: u64,
    /// The wav's sha256, lowercase hex: the recording's identity, and what
    /// two recordings of the same reading share.
    pub sha256: String,
    /// RFC 3339, off the stamp the recording is named by.
    pub at: String,
}

/// The read-aloud set as the page reads it: the recordings this machine has,
/// whether one is being recorded right now, the passage they are read from,
/// the terms a run scores them on, and the last recording's failure when
/// there is one - a write that failed after the stop has no dispatch left to
/// answer, so the read carries it.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct ReadAloudState {
    /// Oldest first, so a page that appends draws a list that grows down.
    pub recordings: Vec<ReadAloudRecording>,
    pub recording: bool,
    pub error: Option<String>,
    pub passage: String,
    /// The words of the passage a run scores term accuracy on - the figure
    /// only this corpus can produce.
    pub terms: Vec<String>,
}

/// The set directory's recordings, oldest first, read off the take
/// directories themselves: the name carries the stamp, `meta.json` the
/// length the recorder wrote, and the wav its size and its own sha. A
/// directory without its meta is one the store was interrupted writing, and
/// is skipped rather than listed as a clip the bench would then skip too.
fn read_aloud_recordings(dir: &Path) -> Vec<ReadAloudRecording> {
    let Ok(entries) = std::fs::read_dir(dir) else {
        return Vec::new();
    };
    let mut names: Vec<String> = entries
        .flatten()
        .map(|entry| entry.file_name().to_string_lossy().into_owned())
        .filter(|name| {
            name.strip_prefix("take-").is_some_and(|rest| {
                !rest.is_empty() && rest.bytes().all(|byte| byte.is_ascii_digit())
            })
        })
        .collect();
    names.sort();

    names
        .into_iter()
        .filter_map(|id| {
            let millis = id.strip_prefix("take-")?.parse::<i64>().ok()?;
            let take = dir.join(&id);
            let meta = std::fs::read_to_string(take.join("meta.json")).ok()?;
            let meta: serde_json::Value = serde_json::from_str(&meta).ok()?;
            let wav = take.join("output.wav");
            let at = time::OffsetDateTime::from_unix_timestamp(millis.checked_div(1000)?)
                .ok()?
                .format(&time::format_description::well_known::Rfc3339)
                .ok()?;
            Some(ReadAloudRecording {
                id,
                duration_ms: meta.get("duration_ms")?.as_u64()?,
                bytes: std::fs::metadata(&wav).ok()?.len(),
                sha256: meta.get("sha256")?.as_str()?.to_owned(),
                at,
            })
        })
        .collect()
}

/// The config one target runs under: the base with the target's file in
/// its role's slot.
///
/// The spec carries no URL and no digest - a bench loads what is already
/// on disk and never fetches - and the size is the file's own, so the
/// engine's check is against the bytes that are there.
pub fn config_for(
    base: &forge_dictate::Config,
    models_dir: &Path,
    target: &BenchTarget,
) -> Result<forge_dictate::Config, forge_dictate::Error> {
    let path = models_dir.join(&target.file);
    let size = std::fs::metadata(&path)
        .map_err(|source| forge_dictate::Error::Io { path: path.clone(), source })?
        .len();
    let spec = ModelSpec {
        file: target.file.clone(),
        url: String::new(),
        size,
        sha256: None,
        facts: forge_dictate::ModelFacts::default(),
    };

    let mut cfg = base.clone();
    // **A bench run is not the user's dictation.** Its clips would otherwise
    // land in the per-take diagnostics store - filling the shelf the bench
    // itself scores against, so every run grew its own corpus.
    cfg.diagnostics_dir = None;
    match target.role {
        BenchRole::Transcribing => cfg.asr_model = spec,
        BenchRole::Cleanup => cfg.normalizer = Some(spec),
    }
    Ok(cfg)
}

impl Workspace {
    /// The bench's state, for the page's read.
    pub fn dictate_bench(&self) -> BenchState {
        self.dictate_bench.lock().clone()
    }

    /// The directory the read-aloud set lives under, beside forge's other
    /// machine-local stores; `None` when no app-support dir resolves. A
    /// test points it at its own dir rather than the real one.
    fn read_aloud_dir(&self) -> Option<std::path::PathBuf> {
        #[cfg(any(test, feature = "testing"))]
        if let Some(dir) = self.test_read_aloud_dir.lock().clone() {
            return Some(dir);
        }
        forge_sdk::app_support_dir().ok().map(|dir| dir.join("dictate-read-aloud"))
    }

    /// The read-aloud set, as the page draws it: the recordings this machine
    /// has, whether one is being recorded right now, the passage they are
    /// read from, and the last write's failure when there was one.
    pub fn read_aloud_state(&self) -> ReadAloudState {
        let dir = self.read_aloud_dir();
        let passage = forge_dictate::bench::READ_ALOUD_PASSAGE.to_owned();
        ReadAloudState {
            recordings: dir.as_deref().map(read_aloud_recordings).unwrap_or_default(),
            recording: self.dictate_runtime.lock().set_recording.is_some(),
            error: self.read_aloud_error.lock().clone(),
            terms: forge_dictate::bench::terms(&passage),
            passage,
        }
    }

    /// Drop one recording from the read-aloud set, by the id the page read.
    ///
    /// The id is a directory name this server wrote, and it is checked
    /// against that shape before anything is removed: a page-supplied path
    /// must never name a directory outside the set.
    pub(crate) fn delete_read_aloud(&self, id: &str) -> Result<(), DispatchError> {
        let Some(dir) = self.read_aloud_dir() else {
            return Err(DispatchError::ReadAloudUnavailable {
                reason: "no app-support directory resolves".to_owned(),
            });
        };
        let keep_pattern =
            |rest: &str| !rest.is_empty() && rest.bytes().all(|byte| byte.is_ascii_digit());
        if !id.strip_prefix("take-").is_some_and(keep_pattern) {
            return Err(DispatchError::ReadAloudUnavailable {
                reason: format!("{id} is not a recording this set holds"),
            });
        }
        let path = dir.join(id);
        if let Err(error) = std::fs::remove_dir_all(&path)
            && error.kind() != std::io::ErrorKind::NotFound
        {
            tracing::warn!(%error, id, "read-aloud: the recording was not deleted");
            return Err(DispatchError::ReadAloudUnavailable { reason: error.to_string() });
        }
        Ok(())
    }

    /// Begin recording the read-aloud set: the page's own microphone feeds
    /// the frames, and no engine and no transcript are involved - the
    /// passage's words are known, so recording transcribes nothing.
    ///
    /// One capture at a time, like every other: refused while a take is
    /// live, and while a recording is already running - the refusal names
    /// whatever holds it.
    pub(crate) fn start_read_aloud(&self, initiator: Option<u64>) -> Result<(), DispatchError> {
        if !self.config_arc().dictate.enabled {
            return Err(DispatchError::DictateOff);
        }
        let mut runtime = self.dictate_runtime.lock();
        if let Some(holder) = runtime.live_holder() {
            return Err(DispatchError::TakeLive { holder });
        }
        *self.read_aloud_error.lock() = None;
        runtime.set_recording = Some(SetRecording { initiator, samples: Vec::new() });
        Ok(())
    }

    /// Feed a client's frames into the running recording, when that
    /// connection is the one that started it.
    ///
    /// The audio is capped at the same span a take holds, so a recording
    /// nobody stops cannot grow without bound; the frames past the cap are
    /// dropped rather than silently stored as a shorter reading.
    pub fn read_aloud_push(&self, samples: &[f32], initiator: Option<u64>) -> bool {
        let mut runtime = self.dictate_runtime.lock();
        let Some(recording) = runtime.set_recording.as_mut() else {
            return false;
        };
        if recording.initiator != initiator {
            return false;
        }
        let cap = usize::try_from(self.config_arc().dictate.max_capture_minutes)
            .unwrap_or(usize::MAX)
            .saturating_mul(60)
            .saturating_mul(usize::try_from(forge_dictate::SAMPLE_RATE).unwrap_or(16_000));
        let room = cap.saturating_sub(recording.samples.len());
        let take = room.min(samples.len());
        recording.samples.extend_from_slice(&samples[..take]);
        take > 0
    }

    /// Stop the recording: `keep` writes it as the set, and a stop that does
    /// not keeps nothing. The write runs off the runtime thread, and its
    /// failure lands in the read rather than in this answer - the dispatch
    /// has already gone by the time the bytes hit the disk.
    pub(crate) fn finish_read_aloud(
        self: &Arc<Self>,
        keep: bool,
        initiator: Option<u64>,
    ) -> Result<(), DispatchError> {
        let samples = {
            let mut runtime = self.dictate_runtime.lock();
            let started_by_this_connection = runtime
                .set_recording
                .as_ref()
                .is_some_and(|recording| recording.initiator == initiator);
            if !started_by_this_connection {
                return Err(DispatchError::ReadAloudNotRecording);
            }
            let Some(recording) = runtime.set_recording.take() else {
                return Err(DispatchError::ReadAloudNotRecording);
            };
            recording.samples
        };
        if !keep {
            return Ok(());
        }
        let Some(dir) = self.read_aloud_dir() else {
            *self.read_aloud_error.lock() = Some("no app-support directory resolves".to_owned());
            return Ok(());
        };
        if samples.is_empty() {
            *self.read_aloud_error.lock() = Some("nothing was captured".to_owned());
            return Ok(());
        }

        let this = Arc::clone(self);
        tokio::task::spawn_blocking(move || {
            let outcome = forge_dictate::bench::store_read_aloud(&dir, &samples);
            match outcome {
                Ok(()) => {}
                Err(error) => {
                    tracing::warn!(
                        event_name = "read_aloud_write_failed",
                        %error,
                        dir = %dir.display(),
                        "the read-aloud set was not written"
                    );
                    *this.read_aloud_error.lock() = Some(error.to_string());
                }
            }
            this.push_models();
        });
        Ok(())
    }

    /// Every saved result, newest first, capped at what the page draws.
    pub fn bench_results(&self) -> Vec<BenchResult> {
        let db = self.db.lock();
        db.as_ref()
            .map(|db| {
                let mut rows = crate::store::bench_results::results(db).unwrap_or_default();
                rows.truncate(RESULTS_SHOWN);
                rows
            })
            .unwrap_or_default()
    }

    /// Run one model over one tier's corpus on this machine's own material.
    ///
    /// Refused while a take is live (the microphone is the machine's, and a
    /// bench would hold a second engine through it) and while another bench
    /// runs - one at a time, like a download. The outcome rides
    /// [`SessionUpdate::DictateModelsChanged`](crate::SessionUpdate::DictateModelsChanged),
    /// the way every other dictate action's does.
    pub(crate) fn start_bench(
        self: &Arc<Self>,
        target: BenchTarget,
        tier: forge_dictate::bench::Tier,
    ) -> Result<(), DispatchError> {
        if !self.config_arc().dictate.enabled {
            return Err(DispatchError::DictateOff);
        }
        self.claim_bench()?;
        if let Some(holder) = self.dictate_runtime.lock().live_holder() {
            self.release_bench();
            return Err(DispatchError::TakeLive { holder });
        }
        *self.dictate_bench.lock() =
            BenchState::Running { target: target.clone(), tier, clip: 0, clips: 0, so_far: None };
        self.dictate_bench_cancel.store(false, Ordering::Relaxed);

        let this = Arc::clone(self);
        tokio::spawn(async move {
            let runner = Arc::clone(&this);
            runner.run_bench(target, tier).await;
            this.push_models();
        });
        Ok(())
    }

    /// Drop one saved result: the row the page asked to delete, by its key.
    pub(crate) fn delete_bench_result(
        &self,
        target: &BenchTarget,
        tier: forge_dictate::bench::Tier,
        corpus: &str,
    ) -> Result<(), DispatchError> {
        let db = self.db.lock();
        if let Some(db) = db.as_ref() {
            let key =
                crate::store::bench_results::key_parts(target.role, &target.file, tier, corpus);
            if let Err(error) = crate::store::bench_results::remove(db, &key) {
                tracing::warn!(
                    event_name = "dictate_bench_delete_failed",
                    %error,
                    "the saved result was not deleted"
                );
            }
        }
        Ok(())
    }

    /// Stop the run in flight. What it measured so far is discarded rather
    /// than saved: a partial corpus is not a result.
    pub(crate) fn stop_bench(&self) -> Result<(), DispatchError> {
        if !matches!(*self.dictate_bench.lock(), BenchState::Running { .. }) {
            return Err(DispatchError::BenchNotRunning);
        }
        self.dictate_bench_cancel.store(true, Ordering::Relaxed);
        Ok(())
    }

    /// Whether a bench holds the machine, which is what a swap waits on.
    /// Both benches count: each loads its own engines, and a swap under
    /// either would pull a model out from under a run.
    pub(crate) fn bench_running(&self) -> bool {
        self.bench_claim.load(Ordering::SeqCst)
            || matches!(*self.dictate_bench.lock(), BenchState::Running { .. })
            || matches!(*self.speaking_bench.lock(), SpeakingBenchState::Running { .. })
    }

    /// Take the one bench slot, or answer that another run holds it. The
    /// claim is a single atomic, so two presses cannot both pass the check
    /// and load two engines.
    fn claim_bench(&self) -> Result<(), DispatchError> {
        if self.bench_running() {
            return Err(DispatchError::BenchRunning);
        }
        self.bench_claim
            .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
            .map(|_| ())
            .map_err(|_| DispatchError::BenchRunning)
    }

    fn release_bench(&self) {
        self.bench_claim.store(false, Ordering::SeqCst);
    }

    /// The speaking bench's state, for the page's read.
    pub fn speaking_bench(&self) -> SpeakingBenchState {
        self.speaking_bench.lock().clone()
    }

    /// Every saved speaking result, newest first, capped like the words
    /// bench's list.
    pub fn speaking_bench_results(&self) -> Vec<SpeakingBenchResult> {
        let db = self.db.lock();
        db.as_ref()
            .map(|db| {
                let mut rows = crate::store::speaking_results::results(db).unwrap_or_default();
                rows.truncate(RESULTS_SHOWN);
                rows
            })
            .unwrap_or_default()
    }

    /// Score the speaking loop on the public corpus: synthesize each
    /// transcript, hear it back through the settled recognizer, and record
    /// the round trip. Refused while a take is live or another bench runs -
    /// one at a time, like the words bench.
    ///
    /// The run gets a thread of its own with a current-thread runtime: it
    /// awaits the engine's chunks and blocks on recognition, so it must not
    /// hold a worker from the server's runtime for minutes.
    pub(crate) fn start_speaking_bench(self: &Arc<Self>) -> Result<(), DispatchError> {
        let config = self.config_arc();
        if !config.tts.enabled {
            return Err(DispatchError::VoiceOff { reason: "[tts] enabled is false".to_owned() });
        }
        if !config.dictate.enabled {
            return Err(DispatchError::VoiceOff {
                reason: "[dictate] enabled is false, so the recognizer half cannot run".to_owned(),
            });
        }
        self.claim_bench()?;
        if let Some(holder) = self.dictate_runtime.lock().live_holder() {
            self.release_bench();
            return Err(DispatchError::TakeLive { holder });
        }
        *self.speaking_bench.lock() = SpeakingBenchState::Running { clip: 0, clips: 0 };
        self.speaking_bench_cancel.store(false, Ordering::Relaxed);

        let this = Arc::clone(self);
        let spawned = std::thread::Builder::new().name("forge-tts-bench".into()).spawn(move || {
            let runtime = match tokio::runtime::Builder::new_current_thread().enable_all().build() {
                Ok(runtime) => runtime,
                Err(error) => {
                    this.settle_speaking_bench(Err(format!(
                        "the bench runtime did not build: {error}"
                    )));
                    this.push_models();
                    return;
                }
            };
            runtime.block_on(Arc::clone(&this).run_speaking_bench());
            this.push_models();
        });
        if let Err(error) = spawned {
            self.release_bench();
            *self.speaking_bench.lock() = SpeakingBenchState::Failed {
                reason: format!("the bench thread did not start: {error}"),
            };
        }
        Ok(())
    }

    /// Stop the speaking run in flight. What it measured so far is
    /// discarded rather than saved: a partial corpus is not a result.
    pub(crate) fn stop_speaking_bench(&self) -> Result<(), DispatchError> {
        if !matches!(*self.speaking_bench.lock(), SpeakingBenchState::Running { .. }) {
            return Err(DispatchError::BenchNotRunning);
        }
        self.speaking_bench_cancel.store(true, Ordering::Relaxed);
        Ok(())
    }

    /// Gather the public corpus, load both engines, and run the loop.
    async fn run_speaking_bench(self: Arc<Self>) {
        let settings = self.config_arc().dictate.clone();
        let active = self.active_models();
        let cfg = crate::dictate::preflight_config(&settings, &active);
        let Some(models_dir) = settings.models_dir() else {
            return self.settle_speaking_bench(Err("no models directory is configured".to_owned()));
        };
        let takes_dir = cfg.diagnostics_dir.clone().unwrap_or_else(|| models_dir.clone());
        let read_aloud_dir = self.read_aloud_dir().unwrap_or_else(|| takes_dir.clone());
        // The bundled public slice lives beside the models, installed on
        // first use, as the asset both directions read.
        let public_dir = models_dir.join("public-corpus");
        if let Err(error) = forge_dictate::bench::install_public_corpus(&models_dir) {
            return self.settle_speaking_bench(Err(format!(
                "the bundled public corpus could not be installed: {error}"
            )));
        }

        let corpus = match forge_dictate::bench::corpus(
            forge_dictate::bench::Tier::Public,
            &takes_dir,
            &read_aloud_dir,
            &public_dir,
        ) {
            Ok(corpus) => corpus,
            Err(error) => return self.settle_speaking_bench(Err(error.to_string())),
        };
        let clips: Vec<forge_tts::bench::BenchClip> = corpus
            .clips
            .iter()
            .filter_map(|clip| clip.truth.clone())
            .map(|transcript| forge_tts::bench::BenchClip { transcript })
            .collect();
        if clips.is_empty() {
            return self.settle_speaking_bench(Err(
                "nothing to bench: the bundled public corpus holds no clips".to_owned(),
            ));
        }
        let total = clips.len();
        *self.speaking_bench.lock() = SpeakingBenchState::Running { clip: 0, clips: total };

        let Some(files) = self.speaking_files() else {
            return self.settle_speaking_bench(Err(
                "the speaking model is not installed; install it from the models page".to_owned(),
            ));
        };
        let engine = match forge_tts::KokoroTts::load(&files) {
            Ok(engine) => engine,
            Err(error) => return self.settle_speaking_bench(Err(error.to_string())),
        };
        let recognizer = match forge_dictate::Engine::new(cfg) {
            Ok(engine) => engine,
            Err(error) => return self.settle_speaking_bench(Err(error.to_string())),
        };
        if let Err(error) = recognizer.wait_ready() {
            return self.settle_speaking_bench(Err(error.to_string()));
        }

        // A recognizer failure is the bench's failure, not a clip the model
        // said nothing on: the first one is kept and the run settles with
        // it, so a broken pair never reads as a perfect score.
        let heard_error: std::sync::Mutex<Option<String>> = std::sync::Mutex::new(None);
        let transcribe = |pcm: &[f32], rate: u32| {
            let record = |said: String| {
                if let Ok(mut slot) = heard_error.lock()
                    && slot.is_none()
                {
                    *slot = Some(said);
                }
                String::new()
            };
            let resampled =
                match forge_dictate::resample_mono(pcm, rate, forge_dictate::SAMPLE_RATE) {
                    Ok(resampled) => resampled,
                    Err(error) => return record(error.to_string()),
                };
            match recognizer
                .transcribe(forge_dictate::Samples::mono(resampled))
                .and_then(forge_dictate::Ticket::recv)
            {
                Ok(forge_dictate::Outcome::Transcript(transcript)) => transcript.text,
                Ok(other) => record(format!("a clip produced no transcript: {other:?}")),
                Err(error) => record(error.to_string()),
            }
        };

        let this = Arc::clone(&self);
        let outcome = forge_tts::bench::run(&engine, &clips, transcribe, |clip| {
            *this.speaking_bench.lock() = SpeakingBenchState::Running { clip, clips: total };
            this.push_models();
            if this.speaking_bench_cancel.load(Ordering::Relaxed) {
                std::ops::ControlFlow::Break(())
            } else {
                std::ops::ControlFlow::Continue(())
            }
        })
        .await;
        if let Some(said) = heard_error.lock().ok().and_then(|slot| slot.clone()) {
            return self.settle_speaking_bench(Err(said));
        }
        match outcome {
            Ok(metrics) => {
                let file = files.model.file_name().map_or_else(
                    || files.model.display().to_string(),
                    |name| name.to_string_lossy().into_owned(),
                );
                self.settle_speaking_bench(Ok(SpeakingBenchResult {
                    file,
                    metrics,
                    at: crate::install::rfc3339_now(),
                    corpus: corpus.id,
                }));
            }
            Err(error) => self.settle_speaking_bench(Err(error.to_string())),
        }
    }

    /// A speaking run that ended: a result is saved, a stop is idle, and
    /// any other ending is a failure with its own reason.
    fn settle_speaking_bench(&self, outcome: Result<SpeakingBenchResult, String>) {
        self.release_bench();
        match outcome {
            Ok(result) => {
                self.save_speaking_result(&result);
                *self.speaking_bench.lock() = SpeakingBenchState::Idle;
            }
            Err(reason) => {
                if self.speaking_bench_cancel.load(Ordering::Relaxed) {
                    *self.speaking_bench.lock() = SpeakingBenchState::Idle;
                } else {
                    *self.speaking_bench.lock() = SpeakingBenchState::Failed { reason };
                }
            }
        }
    }

    fn save_speaking_result(&self, result: &SpeakingBenchResult) {
        let db = self.db.lock();
        if let Some(db) = db.as_ref()
            && let Err(error) = crate::store::speaking_results::save(db, result)
        {
            tracing::warn!(
                event_name = "speaking_bench_save_failed",
                %error,
                file = %result.file,
                "the speaking bench's result was not saved; the numbers are gone with the run"
            );
        }
    }

    /// Gather the corpus, build the target's config, and run it.
    async fn run_bench(self: Arc<Self>, target: BenchTarget, tier: forge_dictate::bench::Tier) {
        let settings = self.config_arc().dictate.clone();
        let active = self.active_models();
        let cfg = crate::dictate::preflight_config(&settings, &active);
        let Some(models_dir) = settings.models_dir() else {
            return self.fail_bench(target, "no models directory is configured".to_owned());
        };
        let takes_dir = cfg.diagnostics_dir.clone();
        let read_aloud_dir = self.read_aloud_dir();
        // The bundled public slice lives beside the models, installed on
        // first use: one directory both directions read.
        let public_dir = models_dir.join("public-corpus");
        // The closure takes its own copy: the caller keeps the target for
        // the failure paths below.
        let ran = target.clone();

        let this = Arc::clone(&self);
        let outcome = tokio::task::spawn_blocking(move || {
            let Some(takes_dir) = takes_dir else {
                return Err("no diagnostics directory is configured".to_owned());
            };
            let read_aloud_dir = read_aloud_dir.unwrap_or_else(|| takes_dir.clone());
            if tier == forge_dictate::bench::Tier::Public
                && let Err(error) = forge_dictate::bench::install_public_corpus(&models_dir)
            {
                return Err(format!("the bundled public corpus could not be installed: {error}"));
            }
            let corpus =
                forge_dictate::bench::corpus(tier, &takes_dir, &read_aloud_dir, &public_dir)
                    .map_err(|error| error.to_string())?;
            if corpus.clips.is_empty() {
                return Err(match tier {
                    forge_dictate::bench::Tier::Consensus => {
                        "nothing to bench: no takes have been saved here yet - dictate a take, \
                         or record the read-aloud passage and score that"
                            .to_owned()
                    }
                    forge_dictate::bench::Tier::ReadAloud => {
                        "nothing to bench: the read-aloud passage has not been recorded yet"
                            .to_owned()
                    }
                    forge_dictate::bench::Tier::Public => {
                        "nothing to bench: the bundled public corpus holds no clips".to_owned()
                    }
                });
            }
            let run_cfg = config_for(&cfg, &models_dir, &ran).map_err(|e| e.to_string())?;

            let mut compared = 0_usize;
            let mut agreed = 0_usize;
            let clips = corpus.clips.len();
            let outcome = forge_dictate::bench::run(&run_cfg, &corpus, |clip| {
                if clip.matched.is_some() {
                    compared += 1;
                    if clip.matched == Some(true) {
                        agreed += 1;
                    }
                }
                *this.dictate_bench.lock() = BenchState::Running {
                    target: ran.clone(),
                    tier,
                    clip: clip.index + 1,
                    clips,
                    so_far: (compared > 0).then(|| forge_dictate::bench::ratio(agreed, compared)),
                };
                this.push_models();
                if this.dictate_bench_cancel.load(Ordering::Relaxed) {
                    std::ops::ControlFlow::Break(())
                } else {
                    std::ops::ControlFlow::Continue(())
                }
            });
            Ok::<_, String>((outcome, corpus.id))
        })
        .await;

        match outcome {
            Ok(Ok((Ok(metrics), corpus))) => {
                let result = BenchResult {
                    target: target.clone(),
                    tier,
                    metrics,
                    at: crate::install::rfc3339_now(),
                    corpus,
                };
                self.settle_bench_success(&result);
            }
            Ok(Ok((Err(error), _))) => self.settle_bench(&target, error.to_string()),
            Ok(Err(reason)) => self.settle_bench(&target, reason),
            Err(join) => self.settle_bench(&target, join.to_string()),
        }
    }

    /// A run that produced a result: save it, release the one bench slot,
    /// and go idle. The release is here rather than only on the failure
    /// paths, or one successful run would hold the slot for the process's
    /// life.
    fn settle_bench_success(&self, result: &BenchResult) {
        self.save_bench_result(result);
        self.release_bench();
        *self.dictate_bench.lock() = BenchState::Idle;
    }

    /// A run that ended without a result: idle when it was stopped on
    /// request - a partial corpus is not a result, and nothing is saved -
    /// and failed with its own reason otherwise.
    fn settle_bench(&self, target: &BenchTarget, reason: String) {
        self.release_bench();
        if self.dictate_bench_cancel.load(Ordering::Relaxed) {
            *self.dictate_bench.lock() = BenchState::Idle;
        } else {
            self.fail_bench(target.clone(), reason);
        }
    }

    fn fail_bench(&self, target: BenchTarget, reason: String) {
        self.release_bench();
        *self.dictate_bench.lock() = BenchState::Failed { target, reason };
    }

    fn save_bench_result(&self, result: &BenchResult) {
        let db = self.db.lock();
        if let Some(db) = db.as_ref()
            && let Err(error) = crate::store::bench_results::save(db, result)
        {
            tracing::warn!(
                event_name = "dictate_bench_save_failed",
                %error,
                file = %result.target.file,
                "the bench's result was not saved; the numbers are gone with the run"
            );
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::SeatUpdate;

    /// A file present on disk, so the target rules are about the rules.
    fn present(dir: &Path, file: &str) {
        std::fs::write(dir.join(file), b"weights").unwrap();
    }

    fn target(file: &str, role: BenchRole) -> BenchTarget {
        BenchTarget { file: file.to_owned(), role, pinned: false }
    }

    fn metrics(clips: usize) -> forge_dictate::bench::Metrics {
        forge_dictate::bench::Metrics {
            clips,
            audio_seconds: 12.5,
            wall_seconds: 1.25,
            xrt_wall: 10.0,
            term_accuracy: Some(0.75),
            wer: Some(0.05),
            matched: Some((2, 3)),
            stages_ms: forge_dictate::bench::StageTotals::default(),
        }
    }

    fn corpus_id(sha: &str) -> forge_dictate::bench::CorpusId {
        forge_dictate::bench::CorpusId { clips: 3, audio_seconds: 12, sha256: sha.to_owned() }
    }

    /// The config a target runs under: the base with the target's file in
    /// its slot, no URL and no digest - the bench fetches nothing - and the
    /// size the file's own.
    #[test]
    fn a_targets_config_swaps_its_own_slot_and_leaves_the_other_alone() {
        let dir = tempfile::tempdir().unwrap();
        present(dir.path(), "candidate.gguf");
        let base = forge_dictate::Config::default();

        let for_it = config_for(
            &base,
            dir.path(),
            &BenchTarget {
                file: "candidate.gguf".to_owned(),
                role: BenchRole::Transcribing,
                pinned: false,
            },
        )
        .expect("the file is on disk");

        assert_eq!(for_it.asr_model.file, "candidate.gguf");
        assert!(for_it.asr_model.url.is_empty(), "a bench never fetches");
        assert!(for_it.asr_model.sha256.is_none(), "no digest is published for one");
        assert_eq!(for_it.asr_model.size, 7, "the size is the bytes on disk");
        assert!(
            for_it.diagnostics_dir.is_none(),
            "a bench run must not write takes into the store it scores against"
        );
        assert_eq!(
            for_it.normalizer.as_ref().map(|spec| spec.file.clone()),
            base.normalizer.as_ref().map(|spec| spec.file.clone()),
            "the other role keeps its model"
        );

        let for_cleanup = config_for(
            &base,
            dir.path(),
            &BenchTarget {
                file: "candidate.gguf".to_owned(),
                role: BenchRole::Cleanup,
                pinned: false,
            },
        )
        .expect("the file is on disk");
        assert_eq!(
            for_cleanup.normalizer.as_ref().map(|spec| spec.file.as_str()),
            Some("candidate.gguf")
        );
        assert_eq!(for_cleanup.asr_model.file, base.asr_model.file);
    }

    /// A target whose file vanished between the list and the run fails
    /// naming the path rather than loading something else.
    #[test]
    fn a_target_whose_file_is_gone_fails_by_name() {
        let dir = tempfile::tempdir().unwrap();

        let err = config_for(
            &forge_dictate::Config::default(),
            dir.path(),
            &target("gone.gguf", BenchRole::Transcribing),
        )
        .expect_err("no file, no config");

        assert!(err.to_string().contains("gone.gguf"), "got: {err}");
    }

    /// Both benches load engines, so neither starts while the other runs.
    #[tokio::test]
    async fn the_two_benches_refuse_each_other_by_name() {
        let (ws, _updates, _config, _models) = voice_stub();
        *ws.dictate_bench.lock() = BenchState::Running {
            target: target("asr.gguf", BenchRole::Transcribing),
            tier: forge_dictate::bench::Tier::Consensus,
            clip: 0,
            clips: 0,
            so_far: None,
        };
        let err = ws.start_speaking_bench().expect_err("one bench at a time");
        assert!(matches!(err, crate::DispatchError::BenchRunning), "got: {err:?}");

        *ws.dictate_bench.lock() = BenchState::Idle;
        *ws.speaking_bench.lock() = SpeakingBenchState::Running { clip: 0, clips: 1 };
        let err = ws
            .dispatch(crate::Command::DictateBench {
                target: target("asr.gguf", BenchRole::Transcribing),
                tier: forge_dictate::bench::Tier::Consensus,
            })
            .expect_err("the speaking bench holds the machine");
        assert!(matches!(err, crate::DispatchError::BenchRunning), "got: {err:?}");
    }

    /// A stop with nothing running is its own refusal.
    #[test]
    fn a_speaking_stop_with_nothing_running_is_refused() {
        let (ws, _updates, _config, _models) = voice_stub();
        let err = ws.dispatch(crate::Command::VoiceBenchStop).expect_err("nothing to stop");
        assert!(matches!(err, crate::DispatchError::BenchNotRunning), "got: {err:?}");
    }

    /// A stopped speaking run settles to idle and saves nothing: a partial
    /// corpus is not a result.
    #[test]
    fn a_stopped_speaking_run_settles_to_idle_and_saves_nothing() {
        let (ws, _updates, _config, _models) = voice_stub();
        let dir = tempfile::tempdir().unwrap();
        ws.install_db_for_test(crate::store::Db::open(&dir.path().join("db.redb")).unwrap());
        *ws.speaking_bench.lock() = SpeakingBenchState::Running { clip: 3, clips: 12 };
        ws.speaking_bench_cancel.store(true, Ordering::Relaxed);

        ws.settle_speaking_bench(Err("stopped".to_owned()));

        assert!(matches!(*ws.speaking_bench.lock(), SpeakingBenchState::Idle));
        assert!(ws.speaking_bench_results().is_empty(), "a stop saves nothing");
    }

    /// A run that failed without a stop settles to failed with its reason.
    #[test]
    fn a_failed_speaking_run_settles_to_failed() {
        let (ws, _updates, _config, _models) = voice_stub();
        *ws.speaking_bench.lock() = SpeakingBenchState::Running { clip: 0, clips: 1 };

        ws.settle_speaking_bench(Err("the model is missing".to_owned()));

        assert!(
            matches!(&*ws.speaking_bench.lock(), SpeakingBenchState::Failed { reason } if reason.contains("missing")),
            "got: {:?}",
            *ws.speaking_bench.lock(),
        );
    }

    /// **Speech switched off refuses by name**: there is no speaking model
    /// for a run to load.
    #[test]
    fn a_speaking_bench_with_speech_off_is_refused_by_name() {
        let (ws, _updates) = Workspace::testing_stub();
        let err = ws.start_speaking_bench().expect_err("no [tts] section to run");
        assert!(matches!(err, crate::DispatchError::VoiceOff { .. }), "got: {err:?}");
    }

    /// **The first speaking run installs the bundled slice, then says what
    /// it is actually missing.** A machine with no corpus needs nothing
    /// fetched; the run reaches its model and fails naming that instead.
    #[tokio::test]
    async fn a_speaking_bench_installs_the_bundled_corpus_before_scoring() {
        let (ws, mut updates, _config, models) = voice_stub();
        ws.dispatch(crate::Command::VoiceBenchStart).expect("the run starts and fails on its own");

        let state = tokio::time::timeout(std::time::Duration::from_secs(15), async {
            loop {
                match updates.recv().await {
                    Some(frame) => {
                        if let crate::SessionUpdate::DictateModelsChanged { models } =
                            std::sync::Arc::unwrap_or_clone(frame.update)
                            && matches!(models.speaking_bench, SpeakingBenchState::Failed { .. })
                        {
                            break models.speaking_bench;
                        }
                    }
                    None => panic!("the subscription must stay attached"),
                }
            }
        })
        .await
        .expect("the failure must land inside fifteen seconds");
        let SpeakingBenchState::Failed { reason } = state else {
            panic!("a run with no speaking model must land failed");
        };
        assert!(!reason.contains("public corpus"), "the bundled corpus is not a blocker: {reason}");
        assert!(reason.contains("speaking model"), "the missing model is named: {reason}");

        let installed = forge_dictate::bench::public_corpus(&models.path().join("public-corpus"))
            .expect("the bundled slice installed");
        assert_eq!(installed.len(), 32, "the whole slice landed beside the models");
    }

    /// **A words-bench run over the public tier installs the slice too** -
    /// the same first use, so both directions score the corpus the binary
    /// carries rather than something a machine had to find.
    #[tokio::test]
    async fn a_public_words_bench_installs_the_bundled_corpus_before_scoring() {
        let (ws, mut updates, _config, models) = voice_stub();
        ws.dispatch(crate::Command::DictateBench {
            target: target("asr.gguf", BenchRole::Transcribing),
            tier: forge_dictate::bench::Tier::Public,
        })
        .expect("the run starts and fails on its own");

        tokio::time::timeout(std::time::Duration::from_secs(15), async {
            loop {
                match updates.recv().await {
                    Some(frame) => {
                        if let crate::SessionUpdate::DictateModelsChanged { models } =
                            std::sync::Arc::unwrap_or_clone(frame.update)
                            && matches!(models.bench, BenchState::Failed { .. })
                        {
                            break;
                        }
                    }
                    None => panic!("the subscription must stay attached"),
                }
            }
        })
        .await
        .expect("the failure must land inside fifteen seconds");

        let installed = forge_dictate::bench::public_corpus(&models.path().join("public-corpus"))
            .expect("the bundled slice installed");
        assert_eq!(installed.len(), 32, "the whole slice landed beside the models");
    }

    /// A workspace with both directions pointed at temp dirs, so a speaking
    /// run reaches the corpus before it would load a model.
    fn voice_stub() -> (
        Arc<Workspace>,
        tokio::sync::mpsc::UnboundedReceiver<SeatUpdate>,
        tempfile::TempDir,
        tempfile::TempDir,
    ) {
        let config_dir = tempfile::tempdir().unwrap();
        let models = tempfile::tempdir().unwrap();
        let mut config = crate::config::LoadedConfig::empty_for_test();
        config.dictate.enabled = true;
        config.dictate.models_dir = Some(models.path().to_string_lossy().into_owned());
        config.tts.enabled = true;
        config.tts.models_dir = Some(models.path().to_string_lossy().into_owned());
        let (ws, updates) =
            Workspace::testing_stub_with_config(config_dir.path().to_path_buf(), config)
                .expect("the stub builds");
        (ws, updates, config_dir, models)
    }

    /// **A bench refuses by name while a take is live** - the microphone is
    /// the machine's, and a bench would hold a second engine through it.
    #[test]
    fn a_bench_refuses_by_name_while_a_take_is_live() {
        let (ws, _updates, _models) = crate::catalogue::tests_catalogue_view::enabled_stub();
        let (stop, _stop_rx) = tokio::sync::mpsc::channel(1);
        ws.dictate_runtime.lock().recordings.insert(
            crate::SessionSlot::new("Busytools", "forge", "worker"),
            crate::dictate::LiveRecording { stop, sink: None, initiator: None },
        );

        let err = ws
            .dispatch(crate::Command::DictateBench {
                target: target("asr.gguf", BenchRole::Transcribing),
                tier: forge_dictate::bench::Tier::Consensus,
            })
            .expect_err("a live take refuses the run");

        assert!(
            matches!(&err, crate::DispatchError::TakeLive { holder } if holder == "Busytools/forge/worker"),
            "the refusal names the seat whose take is live, got: {err:?}"
        );
        assert_eq!(ws.dictate_bench(), BenchState::Idle, "and nothing started");
    }

    /// A stop with nothing running is its own refusal, and a run that was
    /// stopped settles to idle with nothing saved - **a partial corpus is
    /// not a result.**
    #[test]
    fn a_stopped_run_settles_to_idle_and_saves_nothing() {
        let (ws, _updates) = Workspace::testing_stub();
        let dir = tempfile::tempdir().unwrap();
        ws.install_db_for_test(crate::store::Db::open(&dir.path().join("db.redb")).unwrap());

        let err = ws.dispatch(crate::Command::DictateBenchStop).expect_err("nothing to stop");
        assert!(matches!(err, crate::DispatchError::BenchNotRunning), "got: {err:?}");

        let target = target("asr.gguf", BenchRole::Transcribing);
        *ws.dictate_bench.lock() = BenchState::Running {
            target: target.clone(),
            tier: forge_dictate::bench::Tier::Consensus,
            clip: 3,
            clips: 12,
            so_far: Some(0.5),
        };
        ws.dispatch(crate::Command::DictateBenchStop).expect("a running bench stops");

        ws.settle_bench(&target, forge_dictate::Error::Cancelled.to_string());

        assert_eq!(ws.dictate_bench(), BenchState::Idle, "a stop is not a failure");
        assert!(ws.bench_results().is_empty(), "nothing measured part-way is kept");
    }

    /// A successful run releases the one bench slot. A leak here would
    /// refuse every later bench and every install for the process's life.
    #[test]
    fn a_successful_bench_releases_the_claim() {
        let (ws, _updates) = Workspace::testing_stub();
        let dir = tempfile::tempdir().unwrap();
        ws.install_db_for_test(crate::store::Db::open(&dir.path().join("db.redb")).unwrap());
        ws.claim_bench().expect("the claim is free");

        ws.settle_bench_success(&BenchResult {
            target: target("asr.gguf", BenchRole::Transcribing),
            tier: forge_dictate::bench::Tier::Consensus,
            metrics: metrics(3),
            at: "2026-11-02T00:00:00Z".to_owned(),
            corpus: corpus_id("abc"),
        });

        assert!(!ws.bench_running(), "the claim is released");
        assert_eq!(ws.dictate_bench(), BenchState::Idle);
    }

    /// A settled speaking run releases the one bench slot too.
    #[test]
    fn a_settled_speaking_run_releases_the_claim() {
        let (ws, _updates, _config, _models) = voice_stub();
        ws.claim_bench().expect("the claim is free");

        ws.settle_speaking_bench(Err("stopped".to_owned()));

        assert!(!ws.bench_running(), "the claim is released");
    }

    /// A failed run releases the slot, so a second bench can start.
    #[test]
    fn a_failed_bench_leaves_the_slot_free() {
        let (ws, _updates, _config, _models) = voice_stub();
        ws.claim_bench().expect("the claim is free");

        ws.fail_bench(
            target("asr.gguf", BenchRole::Transcribing),
            "the model would not load".to_owned(),
        );

        ws.claim_bench().expect("the claim is free again");
    }

    /// A stopped run releases the slot, so a second bench can start.
    #[test]
    fn a_stopped_bench_leaves_the_slot_free() {
        let (ws, _updates, _config, _models) = voice_stub();
        ws.claim_bench().expect("the claim is free");
        ws.dictate_bench_cancel.store(true, Ordering::Relaxed);

        ws.settle_bench(&target("asr.gguf", BenchRole::Transcribing), "stopped".to_owned());

        ws.claim_bench().expect("the claim is free again");
    }

    /// A run that ends any other way lands failed, with its own reason.
    #[test]
    fn a_run_that_ends_without_a_result_lands_failed() {
        let (ws, _updates) = Workspace::testing_stub();
        let target = target("asr.gguf", BenchRole::Transcribing);

        ws.settle_bench(&target, "the model would not load".to_owned());

        let BenchState::Failed { target: failed, reason } = ws.dictate_bench() else {
            panic!("a failure must land as one, got {:?}", ws.dictate_bench());
        };
        assert_eq!(failed.file, "asr.gguf");
        assert_eq!(reason, "the model would not load");
    }

    /// A result round-trips through the store newest first, and a re-run
    /// over the same corpus replaces its row rather than stacking one.
    #[test]
    fn a_result_round_trips_through_the_store_newest_first() {
        let (ws, _updates) = Workspace::testing_stub();
        let dir = tempfile::tempdir().unwrap();
        ws.install_db_for_test(crate::store::Db::open(&dir.path().join("db.redb")).unwrap());
        let target = target("asr.gguf", BenchRole::Transcribing);

        let older = BenchResult {
            target: target.clone(),
            tier: forge_dictate::bench::Tier::Consensus,
            metrics: metrics(15),
            at: "2026-10-06T10:00:00Z".to_owned(),
            corpus: corpus_id("aaaa"),
        };
        let mut newer = older.clone();
        newer.at = "2026-10-06T11:00:00Z".to_owned();
        newer.corpus = corpus_id("bbbb");
        ws.save_bench_result(&older);
        ws.save_bench_result(&newer);

        let rows = ws.bench_results();

        assert_eq!(rows.len(), 2, "two corpora are two results");
        assert_eq!(rows[0].at, "2026-10-06T11:00:00Z", "newest first");
        assert_eq!(rows[0].metrics.clips, 15, "the metrics cross whole");
        assert_eq!(rows[0].metrics.term_accuracy, Some(0.75));

        // The same corpus again: the row is replaced, not duplicated.
        let mut rerun = older.clone();
        rerun.at = "2026-10-06T12:00:00Z".to_owned();
        ws.save_bench_result(&rerun);
        let rows = ws.bench_results();
        assert_eq!(rows.len(), 2, "a re-run replaces its own row");
        assert_eq!(
            rows.iter().filter(|row| row.corpus.sha256 == "aaaa").count(),
            1,
            "one row per (role, file, tier, corpus)"
        );
    }

    /// The last models frame queued so far, draining what came before it.
    fn drained_models(
        updates: &mut tokio::sync::mpsc::UnboundedReceiver<SeatUpdate>,
    ) -> Option<crate::catalogue::DictateModelsSnapshot> {
        let mut found = None;
        while let Ok(frame) = updates.try_recv() {
            if let crate::SessionUpdate::DictateModelsChanged { models } =
                std::sync::Arc::unwrap_or_clone(frame.update)
            {
                found = Some(models);
            }
        }
        found
    }

    /// **Starting and stopping a recording both push the read.** The card is
    /// drawn from a frame like every other fact on the page, so a command
    /// that moves the recording without a frame leaves the press looking
    /// dead.
    #[test]
    fn starting_and_stopping_a_recording_push_the_models_read() {
        let (ws, mut updates, _models) = crate::catalogue::tests_catalogue_view::enabled_stub();
        let dir = tempfile::tempdir().unwrap();
        *ws.test_read_aloud_dir.lock() = Some(dir.path().to_path_buf());

        ws.dispatch(crate::Command::DictateReadAloudStart { initiator: Some(1) })
            .expect("nothing else is capturing");
        let models = drained_models(&mut updates).expect("starting pushes the read");
        assert!(models.read_aloud.recording, "the frame carries the recording");

        ws.dispatch(crate::Command::DictateReadAloudStop { keep: false, initiator: Some(1) })
            .expect("the recording is this connection's");
        let models = drained_models(&mut updates).expect("stopping pushes the read");
        assert!(!models.read_aloud.recording, "the frame carries it stopped");
    }

    /// The next frame that carries a written set, inside a test's patience:
    /// the write runs off the runtime thread, so the stop's own frame lands
    /// before it and this is the one that says the bytes are on disk.
    async fn await_recorded(
        updates: &mut tokio::sync::mpsc::UnboundedReceiver<SeatUpdate>,
    ) -> crate::catalogue::DictateModelsSnapshot {
        tokio::time::timeout(std::time::Duration::from_secs(15), async {
            loop {
                match updates
                    .recv()
                    .await
                    .map(|frame| std::sync::Arc::unwrap_or_clone(frame.update))
                {
                    Some(crate::SessionUpdate::DictateModelsChanged { models }) => {
                        if !models.read_aloud.recordings.is_empty() {
                            break models;
                        }
                    }
                    Some(_) => {}
                    None => panic!("the subscription must stay attached"),
                }
            }
        })
        .await
        .expect("the set must land inside fifteen seconds")
    }

    /// **A kept recording becomes the set.** The frames the page fed land as
    /// the passage's own take - and only the recording connection's frames
    /// do: another connection's are not the recording's to feed. A stop that
    /// does not keep writes nothing.
    #[tokio::test]
    async fn a_kept_recording_becomes_the_set_and_a_dropped_one_writes_nothing() {
        let (ws, mut updates, _models) = crate::catalogue::tests_catalogue_view::enabled_stub();
        let dir = tempfile::tempdir().unwrap();
        *ws.test_read_aloud_dir.lock() = Some(dir.path().to_path_buf());

        ws.dispatch(crate::Command::DictateReadAloudStart { initiator: Some(1) }).unwrap();
        assert!(ws.read_aloud_push(&vec![0.25_f32; 1_600], Some(1)), "its own frames land");
        assert!(!ws.read_aloud_push(&[0.0_f32; 8], Some(2)), "another connection's do not");
        ws.dispatch(crate::Command::DictateReadAloudStop { keep: true, initiator: Some(1) })
            .unwrap();

        let models = await_recorded(&mut updates).await;
        assert_eq!(models.read_aloud.recordings.len(), 1, "the read says the set is here");
        let clips = forge_dictate::bench::read_aloud(dir.path()).expect("the set reads");
        assert_eq!(clips.len(), 1, "a passage and a take are there");
        assert_eq!(clips[0].audio.len(), 1_600, "the frames the page fed");

        // A second recording, dropped rather than kept: nothing replaces the
        // set that stands.
        ws.dispatch(crate::Command::DictateReadAloudStart { initiator: Some(1) }).unwrap();
        assert!(ws.read_aloud_push(&vec![0.5_f32; 800], Some(1)));
        ws.dispatch(crate::Command::DictateReadAloudStop { keep: false, initiator: Some(1) })
            .unwrap();
        let clips = forge_dictate::bench::read_aloud(dir.path()).unwrap();
        assert_eq!(clips.len(), 1, "a dropped recording leaves the set alone");
        assert_eq!(clips[0].audio.len(), 1_600);
    }

    /// **A recording is deleted by the id the page read, and nothing else
    /// is.** The id is a directory name this server wrote, and a page-supplied
    /// path must never name a directory outside the set.
    #[tokio::test]
    async fn a_recording_is_deleted_by_its_id_and_a_foreign_one_is_refused() {
        let (ws, mut updates, _models) = crate::catalogue::tests_catalogue_view::enabled_stub();
        let dir = tempfile::tempdir().unwrap();
        *ws.test_read_aloud_dir.lock() = Some(dir.path().to_path_buf());

        ws.dispatch(crate::Command::DictateReadAloudStart { initiator: Some(1) }).unwrap();
        assert!(ws.read_aloud_push(&vec![0.25_f32; 1_600], Some(1)));
        ws.dispatch(crate::Command::DictateReadAloudStop { keep: true, initiator: Some(1) })
            .unwrap();
        let models = await_recorded(&mut updates).await;
        let id = models.read_aloud.recordings[0].id.clone();
        assert_eq!(models.read_aloud.recordings[0].duration_ms, 100, "a tenth of a second");

        // A path that is not one of this set's own directory names is refused
        // rather than removed.
        let outside = dir.path().join("neighbour");
        std::fs::create_dir_all(&outside).unwrap();
        for id in ["../neighbour", "neighbour", "take-x", "take-"] {
            let err = ws.dispatch(crate::Command::DictateReadAloudDelete { id: id.to_owned() });
            assert!(err.is_err(), "accepted {id:?}");
        }
        assert!(outside.is_dir(), "a foreign directory must survive");

        ws.dispatch(crate::Command::DictateReadAloudDelete { id: id.clone() }).unwrap();
        let models = drained_models(&mut updates).expect("the delete pushes the read");
        assert!(models.read_aloud.recordings.is_empty(), "the recording left the set");
        assert!(!dir.path().join(&id).exists(), "and its directory went with it");
    }

    /// One capture at a time: a take that is live refuses a recording by
    /// name, and a recording counts as live for the guards that wait on
    /// captures.
    #[test]
    fn a_recording_and_a_take_refuse_each_other() {
        let (ws, _updates, _models) = crate::catalogue::tests_catalogue_view::enabled_stub();
        let dir = tempfile::tempdir().unwrap();
        *ws.test_read_aloud_dir.lock() = Some(dir.path().to_path_buf());

        ws.dispatch(crate::Command::DictateReadAloudStart { initiator: Some(1) }).unwrap();
        assert_eq!(
            ws.dictate_runtime.lock().live_holder().as_deref(),
            Some("the read-aloud recording"),
            "a recording is a live capture"
        );
        let err = ws
            .dispatch(crate::Command::DictateReadAloudStart { initiator: Some(1) })
            .expect_err("one recording at a time");
        assert!(
            matches!(err, crate::DispatchError::TakeLive { ref holder } if holder == "the read-aloud recording"),
            "got: {err:?}"
        );

        let (stop, _stop_rx) = tokio::sync::mpsc::channel(1);
        ws.dictate_runtime.lock().recordings.insert(
            crate::SessionSlot::new("Busytools", "forge", "worker"),
            crate::dictate::LiveRecording { stop, sink: None, initiator: None },
        );
        ws.dispatch(crate::Command::DictateReadAloudStop { keep: false, initiator: Some(1) })
            .unwrap();
        let err = ws
            .dispatch(crate::Command::DictateReadAloudStart { initiator: Some(1) })
            .expect_err("a take holds the capture");
        assert!(
            matches!(&err, crate::DispatchError::TakeLive { holder } if holder == "Busytools/forge/worker"),
            "the refusal names the seat whose take is live, got: {err:?}"
        );
    }
}