kcode_k1_audio_classification/

lib.rs

pub use kcode_k1_audio_classification_projection::{
    ExecutedAnalysis, FragmentId, FragmentStageV1, FragmentStatus, LlmJobState, LlmJobStatus,
    OverallState, SpeakerLabelV1, StageState, StageStatus,
};

use kcode_k1_audio_classification_projection::{Projection, ProjectionEffect};
use kcode_k1_audio_fragment_runner as runner;
use kcode_k1_audio_fragment_submit as fragment_submit;
use kcode_k1_audio_fragment_transactions as transactions;
use kcode_k1_objects::K1Objects;
use kcode_k1_peering::K1Peering;
use kcode_k1_txn_ordering::{K1TxnOrdering, Subsystem, SubsystemId, TxId};
use kcode_speaker_v3_analysis::Analyzer;
use std::collections::{HashMap, HashSet};
use std::future::Future;
use std::path::Path;
#[cfg(test)]
use std::path::PathBuf;
use std::pin::Pin;
use std::rc::Rc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering as AtomicOrdering};
use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};
use tokio::sync::mpsc;
use tokio::task::JoinHandle as LocalJoinHandle;

const RESTART_ERROR: &str = "analysis interrupted by restart";
const SUBSYSTEM_NAME: &str = "audio-classification";

pub struct AudioClassification {
    projection: Arc<Projection>,
    _ordering: Arc<K1TxnOrdering>,
    peering: Arc<K1Peering>,
    objects: Arc<K1Objects>,
    control: Arc<Control>,
    health: Arc<Health>,
    worker: Option<JoinHandle<()>>,
}

impl AudioClassification {
    pub fn open(
        root: &Path,
        ordering: Arc<K1TxnOrdering>,
        peering: Arc<K1Peering>,
        objects: Arc<K1Objects>,
        analyzer: Analyzer,
    ) -> Result<Self, String> {
        Self::open_with_engine(
            root,
            ordering,
            peering,
            objects,
            Box::new(ProductionEngine { analyzer }),
        )
    }

    fn open_with_engine(
        root: &Path,
        ordering: Arc<K1TxnOrdering>,
        peering: Arc<K1Peering>,
        objects: Arc<K1Objects>,
        engine: Box<dyn Engine>,
    ) -> Result<Self, String> {
        let (projection, cursor) = Projection::open(root, &ordering)?;
        let projection = Arc::new(projection);
        let control = Arc::new(Control::default());
        let health = Arc::new(Health::default());
        let replaying = Arc::new(AtomicBool::new(true));
        let replayed = Arc::new(AtomicU64::new(0));
        let callback = Arc::new(Callback {
            projection: projection.clone(),
            control: control.clone(),
            health: health.clone(),
            replaying: replaying.clone(),
            replayed: replayed.clone(),
        });
        let (sender, receiver) = mpsc::unbounded_channel();
        control.install(sender.clone());
        let worker = spawn_worker(
            engine,
            sender,
            receiver,
            control.clone(),
            health.clone(),
            peering.clone(),
            objects.clone(),
        )?;
        let startup = (|| -> Result<(), String> {
            ordering
                .register_subsystem(subsystem_id()?, cursor, callback)
                .map_err(|error| format!("register audio classification: {error}"))?;
            replaying.store(false, AtomicOrdering::Release);
            record_replay(root, replayed.load(AtomicOrdering::Acquire));
            let interrupted = projection
                .running()
                .map_err(|error| format!("read interrupted analyses: {error}"))?;
            for fragment in interrupted {
                transactions::submit_failure(
                    &peering,
                    fragment.fragment_id,
                    fragment.stage,
                    None,
                    RESTART_ERROR.to_owned(),
                )
                .map_err(|error| format!("persist interrupted analysis: {error}"))?;
            }
            control.activate();
            let queued = projection
                .queued()
                .map_err(|error| format!("read queued analyses: {error}"))?;
            for fragment_id in queued {
                control.start(fragment_id, 0)?;
            }
            health.ensure()
        })();
        if let Err(error) = startup {
            control.shutdown();
            let _ = worker.join();
            return Err(error);
        }
        Ok(Self {
            projection,
            _ordering: ordering,
            peering,
            objects,
            control,
            health,
            worker: Some(worker),
        })
    }

    pub fn submit(&self, ogg_bytes: &[u8]) -> Result<FragmentId, String> {
        self.health.ensure()?;
        fragment_submit::submit(&self.objects, &self.peering, ogg_bytes)
            .map_err(|error| self.submission_error(error))
    }

    pub fn status(&self, fragment_id: FragmentId) -> Result<Option<FragmentStatus>, String> {
        self.health.ensure()?;
        self.projection.status(fragment_id)
    }

    pub fn retry(&self, fragment_id: FragmentId) -> Result<(), String> {
        self.health.ensure()?;
        let status = self
            .projection
            .status(fragment_id)?
            .ok_or_else(|| "unknown audio fragment".to_owned())?;
        if status.state != OverallState::Failed {
            return Err("retry requires Failed state".to_owned());
        }
        if !self
            .control
            .start(fragment_id, status.attempt_count)
            .map_err(|error| self.worker_error(error))?
        {
            return Err("retry is already active".to_owned());
        }
        Ok(())
    }

    pub fn discard(&self, fragment_id: FragmentId) -> Result<(), String> {
        self.health.ensure()?;
        let status = self
            .projection
            .status(fragment_id)?
            .ok_or_else(|| "unknown audio fragment".to_owned())?;
        if status.state == OverallState::Discarded {
            return Ok(());
        }
        if !self.control.reserve_discard(fragment_id) {
            return Err("discard is already active".to_owned());
        }
        match transactions::submit_discard(&self.peering, fragment_id) {
            Ok(_) => {
                self.control.discard_done(fragment_id);
                Ok(())
            }
            Err(error) => {
                if !is_committed_error(&error) {
                    self.control.discard_done(fragment_id);
                }
                Err(self.submission_error(error))
            }
        }
    }

    pub fn submit_labels(
        &self,
        fragment_id: FragmentId,
        labels: Vec<SpeakerLabelV1>,
    ) -> Result<(), String> {
        self.health.ensure()?;
        let interim = self.projection.validate_labels(fragment_id, &labels)?;
        if !self.control.reserve_labels(fragment_id) {
            return Err("label submission is already active".to_owned());
        }
        match transactions::submit_label_confirmation(&self.peering, fragment_id, interim, labels) {
            Ok(_) => {
                self.control.labels_done(fragment_id);
                Ok(())
            }
            Err(error) => {
                if !is_committed_error(&error) {
                    self.control.labels_done(fragment_id);
                }
                Err(self.submission_error(error))
            }
        }
    }

    fn submission_error(&self, error: String) -> String {
        if is_committed_error(&error) {
            self.health
                .fault("transaction commitment is ambiguous".to_owned());
            self.control.shutdown();
        }
        error
    }

    fn worker_error(&self, error: String) -> String {
        self.health.fault(error.clone());
        self.control.shutdown();
        error
    }

    #[cfg(test)]
    fn inject_error_burst(&self, id: FragmentId, errors: Vec<String>) -> Result<(), String> {
        self.projection.inject_errors(id, errors)
    }
}

impl Drop for AudioClassification {
    fn drop(&mut self) {
        self.control.shutdown();
        if let Some(worker) = self.worker.take() {
            let _ = worker.join();
        }
    }
}

#[derive(Default)]
struct Health {
    reopen: AtomicBool,
    diagnostic: Mutex<Option<String>>,
}

impl Health {
    fn ensure(&self) -> Result<(), String> {
        if !self.reopen.load(AtomicOrdering::Acquire) {
            return Ok(());
        }
        let detail = lock(&self.diagnostic)
            .clone()
            .unwrap_or_else(|| "processing fault".to_owned());
        Err(format!("audio classification requires reopen: {detail}"))
    }

    fn fault(&self, diagnostic: String) {
        let mut current = lock(&self.diagnostic);
        if current.is_none() {
            *current = Some(diagnostic);
        }
        self.reopen.store(true, AtomicOrdering::Release);
    }
}

#[derive(Default)]
struct Control {
    state: Mutex<ControlState>,
    live: AtomicBool,
}

#[derive(Default)]
struct ControlState {
    sender: Option<mpsc::UnboundedSender<WorkerCommand>>,
    starting: HashMap<FragmentId, StartReservation>,
    labels: HashSet<FragmentId>,
    discards: HashSet<FragmentId>,
    next_generation: u64,
}

struct StartReservation {
    generation: u64,
    baseline_attempt: u32,
}

impl Control {
    fn install(&self, sender: mpsc::UnboundedSender<WorkerCommand>) {
        lock(&self.state).sender = Some(sender);
    }

    fn activate(&self) {
        self.live.store(true, AtomicOrdering::Release);
    }

    fn start(&self, id: FragmentId, baseline_attempt: u32) -> Result<bool, String> {
        if !self.live.load(AtomicOrdering::Acquire) {
            return Err("audio classification worker is unavailable".to_owned());
        }
        let mut state = lock(&self.state);
        if state.starting.contains_key(&id) {
            return Ok(false);
        }
        state.next_generation = state
            .next_generation
            .checked_add(1)
            .ok_or_else(|| "audio classification worker generation overflow".to_owned())?;
        let generation = state.next_generation;
        let sender = state
            .sender
            .clone()
            .ok_or_else(|| "audio classification worker is unavailable".to_owned())?;
        state.starting.insert(
            id,
            StartReservation {
                generation,
                baseline_attempt,
            },
        );
        drop(state);
        if sender.send(WorkerCommand::Start(id, generation)).is_err() {
            self.finished(id, generation);
            return Err("audio classification worker is unavailable".to_owned());
        }
        Ok(true)
    }

    fn progress_applied(&self, id: FragmentId, attempt_count: u32) {
        let mut state = lock(&self.state);
        if state
            .starting
            .get(&id)
            .is_some_and(|value| attempt_count > value.baseline_attempt)
        {
            state.starting.remove(&id);
        }
    }

    fn abort(&self, id: FragmentId) -> Result<(), String> {
        let sender = {
            let mut state = lock(&self.state);
            state.starting.remove(&id);
            state.labels.remove(&id);
            state.discards.remove(&id);
            state.sender.clone()
        };
        if self.live.load(AtomicOrdering::Acquire)
            && sender.is_some_and(|sender| sender.send(WorkerCommand::Abort(id)).is_err())
        {
            return Err("audio classification worker is unavailable".to_owned());
        }
        Ok(())
    }

    fn finished(&self, id: FragmentId, generation: u64) {
        let mut state = lock(&self.state);
        if state
            .starting
            .get(&id)
            .is_some_and(|value| value.generation == generation)
        {
            state.starting.remove(&id);
        }
    }

    fn reserve_labels(&self, id: FragmentId) -> bool {
        lock(&self.state).labels.insert(id)
    }

    fn labels_done(&self, id: FragmentId) {
        lock(&self.state).labels.remove(&id);
    }

    fn reserve_discard(&self, id: FragmentId) -> bool {
        lock(&self.state).discards.insert(id)
    }

    fn discard_done(&self, id: FragmentId) {
        lock(&self.state).discards.remove(&id);
    }

    fn shutdown(&self) {
        self.live.store(false, AtomicOrdering::Release);
        let sender = lock(&self.state).sender.clone();
        if let Some(sender) = sender {
            let _ = sender.send(WorkerCommand::Stop);
        }
    }
}

struct Callback {
    projection: Arc<Projection>,
    control: Arc<Control>,
    health: Arc<Health>,
    replaying: Arc<AtomicBool>,
    replayed: Arc<AtomicU64>,
}

impl Callback {
    fn fault(&self, diagnostic: String) -> String {
        self.health.fault(diagnostic.clone());
        self.control.shutdown();
        diagnostic
    }

    fn react(&self, fragment_id: FragmentId, effect: ProjectionEffect) -> Result<(), String> {
        match effect {
            ProjectionEffect::Start => {
                if self.control.live.load(AtomicOrdering::Acquire) {
                    self.control.start(fragment_id, 0)?;
                }
            }
            ProjectionEffect::Abort => self.control.abort(fragment_id)?,
            ProjectionEffect::LabelsCommitted => self.control.labels_done(fragment_id),
            ProjectionEffect::None => {
                let status = self
                    .projection
                    .status(fragment_id)?
                    .ok_or_else(|| "applied event has no projected fragment".to_owned())?;
                self.control
                    .progress_applied(fragment_id, status.attempt_count);
            }
        }
        Ok(())
    }
}

impl Subsystem for Callback {
    fn submit_txn(&self, id: TxId, payload: &[u8]) -> Result<(), String> {
        let applied = self
            .projection
            .apply(id, payload)
            .map_err(|error| self.fault(format!("apply audio classification event: {error}")))?;
        if self.replaying.load(AtomicOrdering::Acquire) {
            self.replayed.fetch_add(1, AtomicOrdering::AcqRel);
            return Ok(());
        }
        self.react(applied.fragment_id, applied.effect)
            .map_err(|error| self.fault(format!("apply audio classification effect: {error}")))
    }

    fn reorg(&self) -> Result<(), String> {
        let result = self.projection.clear();
        self.health
            .fault("canonical reorganization requires reopen".to_owned());
        self.control.shutdown();
        result
    }
}

type EngineFuture<'a> = Pin<Box<dyn Future<Output = Result<(), String>> + 'a>>;

trait Engine: Send + 'static {
    fn run<'a>(
        &'a self,
        peering: &'a K1Peering,
        fragment_id: FragmentId,
        ogg_bytes: &'a [u8],
    ) -> EngineFuture<'a>;
}

struct ProductionEngine {
    analyzer: Analyzer,
}

impl Engine for ProductionEngine {
    fn run<'a>(
        &'a self,
        peering: &'a K1Peering,
        fragment_id: FragmentId,
        ogg_bytes: &'a [u8],
    ) -> EngineFuture<'a> {
        Box::pin(runner::run(&self.analyzer, peering, fragment_id, ogg_bytes))
    }
}

enum WorkerCommand {
    Start(FragmentId, u64),
    Abort(FragmentId),
    Finished(FragmentId, u64, Result<(), String>),
    Stop,
}

fn spawn_worker(
    engine: Box<dyn Engine>,
    sender: mpsc::UnboundedSender<WorkerCommand>,
    receiver: mpsc::UnboundedReceiver<WorkerCommand>,
    control: Arc<Control>,
    health: Arc<Health>,
    peering: Arc<K1Peering>,
    objects: Arc<K1Objects>,
) -> Result<JoinHandle<()>, String> {
    thread::Builder::new()
        .name("k1-audio-classification".to_owned())
        .spawn(move || worker_main(engine, sender, receiver, control, health, peering, objects))
        .map_err(|error| format!("start audio classification worker: {error}"))
}

fn worker_main(
    engine: Box<dyn Engine>,
    sender: mpsc::UnboundedSender<WorkerCommand>,
    mut receiver: mpsc::UnboundedReceiver<WorkerCommand>,
    control: Arc<Control>,
    health: Arc<Health>,
    peering: Arc<K1Peering>,
    objects: Arc<K1Objects>,
) {
    let runtime = match tokio::runtime::Builder::new_current_thread().build() {
        Ok(runtime) => runtime,
        Err(error) => {
            health.fault(format!("create worker runtime: {error}"));
            control.shutdown();
            return;
        }
    };
    let local = tokio::task::LocalSet::new();
    runtime.block_on(local.run_until(async move {
        let engine: Rc<dyn Engine> = Rc::from(engine);
        let mut tasks: HashMap<FragmentId, (u64, LocalJoinHandle<()>)> = HashMap::new();
        while let Some(command) = receiver.recv().await {
            match command {
                WorkerCommand::Start(id, generation) => {
                    if let Some((old_generation, old_task)) = tasks.remove(&id) {
                        old_task.abort();
                        control.finished(id, old_generation);
                    }
                    let task_engine = engine.clone();
                    let task_peering = peering.clone();
                    let task_objects = objects.clone();
                    let task_sender = sender.clone();
                    let task = tokio::task::spawn_local(async move {
                        let result =
                            run_fragment(task_engine.as_ref(), &task_peering, &task_objects, id)
                                .await;
                        let _ = task_sender.send(WorkerCommand::Finished(id, generation, result));
                    });
                    tasks.insert(id, (generation, task));
                }
                WorkerCommand::Abort(id) => {
                    if let Some((generation, task)) = tasks.remove(&id) {
                        task.abort();
                        control.finished(id, generation);
                    }
                }
                WorkerCommand::Finished(id, generation, result) => {
                    if tasks
                        .get(&id)
                        .is_some_and(|(current, _)| *current == generation)
                    {
                        tasks.remove(&id);
                        control.finished(id, generation);
                    }
                    if let Err(error) = result {
                        health.fault(format!("runner persistence failure: {error}"));
                        control.shutdown();
                        break;
                    }
                }
                WorkerCommand::Stop => break,
            }
        }
        for (id, (generation, task)) in tasks {
            task.abort();
            control.finished(id, generation);
        }
    }));
}

async fn run_fragment(
    engine: &dyn Engine,
    peering: &K1Peering,
    objects: &K1Objects,
    id: FragmentId,
) -> Result<(), String> {
    let object = match objects.load(id) {
        Ok(Some(object)) if object.file_type == "audio/ogg" => object,
        Ok(Some(_)) => {
            transactions::submit_failure(
                peering,
                id,
                FragmentStageV1::Queue,
                None,
                "audio Object is not audio/ogg".to_owned(),
            )?;
            return Ok(());
        }
        Ok(None) => {
            transactions::submit_failure(
                peering,
                id,
                FragmentStageV1::Queue,
                None,
                "audio Object is unavailable".to_owned(),
            )?;
            return Ok(());
        }
        Err(error) => {
            transactions::submit_failure(
                peering,
                id,
                FragmentStageV1::Queue,
                None,
                format!("load audio Object: {error}"),
            )?;
            return Ok(());
        }
    };
    engine.run(peering, id, &object.data).await
}

fn subsystem_id() -> Result<SubsystemId, String> {
    SubsystemId::from_str(SUBSYSTEM_NAME)
}

fn is_committed_error(error: &str) -> bool {
    error.to_ascii_lowercase().contains("committed")
}

fn lock<T>(mutex: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
    mutex.lock().unwrap_or_else(|error| error.into_inner())
}

#[cfg(test)]
fn record_replay(root: &Path, count: u64) {
    lock(replay_counts()).insert(root.to_path_buf(), count);
}

#[cfg(not(test))]
fn record_replay(_root: &Path, _count: u64) {}

#[cfg(test)]
fn replay_counts() -> &'static Mutex<HashMap<PathBuf, u64>> {
    use std::sync::OnceLock;
    static COUNTS: OnceLock<Mutex<HashMap<PathBuf, u64>>> = OnceLock::new();
    COUNTS.get_or_init(|| Mutex::new(HashMap::new()))
}

#[cfg(test)]
mod tests {
    use super::*;
    use kcode_k1_audio_classification_testkit::{
        AdapterError, AdapterFactory, ClassificationAdapter, OpenRequest,
        SUCCESS_INTERIM_TRANSCRIPT, ScriptedOutcome, TestId, TestJob, TestJobState, TestState,
        TestStatus, run_all,
    };
    use kcode_k1_audio_fragment_transactions::ProgressUpdateV1;
    use kcode_k1_txn_ordering::REGISTER_AT_TIP;
    use kcode_speaker_v3_analysis::{
        AnalysisEnvelope, FeatureVector24, GeminiCohort, OggAudioMetadata, StructuredAnalysis,
        StructuredSpeaker, StructurerProvenance,
    };
    use std::collections::VecDeque;
    use tokio::sync::oneshot;

    const PROJECTION_DATABASE_FILE: &str = "audio-classification.sqlite3";

    struct ScriptedEngine {
        outcomes: Mutex<VecDeque<ScriptedOutcome>>,
    }

    impl Engine for ScriptedEngine {
        fn run<'a>(
            &'a self,
            peering: &'a K1Peering,
            id: FragmentId,
            bytes: &'a [u8],
        ) -> EngineFuture<'a> {
            Box::pin(async move {
                let Some(mut outcome) = lock(&self.outcomes).pop_front() else {
                    return Ok(());
                };
                transactions::submit_progress(
                    peering,
                    id,
                    ProgressUpdateV1::LlmJobStarted {
                        sequence: 1,
                        stage: FragmentStageV1::Transcript,
                        name: "transcript".to_owned(),
                    },
                )?;
                loop {
                    match outcome {
                        ScriptedOutcome::Blocked {
                            gate,
                            outcome: next,
                        } => {
                            let (sender, receiver) = oneshot::channel();
                            thread::Builder::new()
                                .name("audio-test-gate".to_owned())
                                .spawn(move || {
                                    gate.wait_for_release();
                                    let _ = sender.send(());
                                })
                                .map_err(|error| error.to_string())?;
                            receiver
                                .await
                                .map_err(|_| "test gate cancelled".to_owned())?;
                            outcome = *next;
                        }
                        ScriptedOutcome::Success => {
                            transactions::submit_progress(
                                peering,
                                id,
                                ProgressUpdateV1::LlmJobSucceeded { sequence: 1 },
                            )?;
                            transactions::submit_transcription_complete(
                                peering,
                                id,
                                success_analysis(bytes)?,
                            )?;
                            return Ok(());
                        }
                        ScriptedOutcome::Failure { message } => {
                            transactions::submit_failure(
                                peering,
                                id,
                                FragmentStageV1::Transcript,
                                Some(1),
                                message,
                            )?;
                            return Ok(());
                        }
                    }
                }
            })
        }
    }

    fn success_analysis(bytes: &[u8]) -> Result<ExecutedAnalysis, String> {
        let speaker_one = "Speaker 1".parse().map_err(|error| format!("{error}"))?;
        let speaker_two = "Speaker 2".parse().map_err(|error| format!("{error}"))?;
        let provenance = StructurerProvenance {
            model_id: "test-model".to_owned(),
            prompt_revision: "test-prompt".to_owned(),
        };
        Ok(ExecutedAnalysis {
            envelope: AnalysisEnvelope {
                audio: OggAudioMetadata::from_ogg_bytes(bytes)
                    .map_err(|error| error.to_string())?,
                analysis: StructuredAnalysis {
                    transcript: SUCCESS_INTERIM_TRANSCRIPT.to_owned(),
                    speakers: vec![
                        StructuredSpeaker {
                            speaker: speaker_one,
                            language: "English".to_owned(),
                            features: FeatureVector24::default(),
                            features_usable_for_training: true,
                        },
                        StructuredSpeaker {
                            speaker: speaker_two,
                            language: "English".to_owned(),
                            features: FeatureVector24::default(),
                            features_usable_for_training: true,
                        },
                    ],
                },
                gemini: GeminiCohort {
                    model_id: "test-gemini".to_owned(),
                    transcript_prompt_revision: "test-transcript".to_owned(),
                    feature_prompt_revisions: std::array::from_fn(|_| "test-feature".to_owned()),
                    feature_schema_revision: "test-schema".to_owned(),
                },
                structurer: provenance.clone(),
            },
            label_extractor: provenance,
        })
    }

    struct TestFactory;

    struct TestAdapter {
        facade: AudioClassification,
    }

    impl AdapterFactory for TestFactory {
        type Adapter = TestAdapter;

        fn open(&self, request: OpenRequest) -> Result<Self::Adapter, AdapterError> {
            let ordering = Arc::new(
                K1TxnOrdering::open(&request.root.join("ordering")).map_err(adapter_error)?,
            );
            let peering = Arc::new(
                K1Peering::open(&request.root.join("peering"), ordering.clone())
                    .map_err(adapter_error)?,
            );
            let objects = Arc::new(
                K1Objects::open(ordering.clone(), peering.clone()).map_err(adapter_error)?,
            );
            let facade = AudioClassification::open_with_engine(
                &request.root.join("projection"),
                ordering,
                peering,
                objects,
                Box::new(ScriptedEngine {
                    outcomes: Mutex::new(request.outcomes.into()),
                }),
            )
            .map_err(adapter_error)?;
            Ok(TestAdapter { facade })
        }

        fn append_queued_while_closed(
            &self,
            root: &Path,
            bytes: &[u8],
        ) -> Result<TestId, AdapterError> {
            let ordering =
                Arc::new(K1TxnOrdering::open(&root.join("ordering")).map_err(adapter_error)?);
            let peering = Arc::new(
                K1Peering::open(&root.join("peering"), ordering.clone()).map_err(adapter_error)?,
            );
            let objects = Arc::new(
                K1Objects::open(ordering.clone(), peering.clone()).map_err(adapter_error)?,
            );
            ordering
                .register_subsystem(
                    subsystem_id().map_err(adapter_error)?,
                    Some(REGISTER_AT_TIP),
                    Arc::new(NoopSubsystem),
                )
                .map_err(adapter_error)?;
            fragment_submit::submit(&objects, &peering, bytes)
                .map(to_test_id)
                .map_err(adapter_error)
        }

        fn corrupt_projection(&self, root: &Path) -> Result<(), AdapterError> {
            std::fs::write(
                root.join("projection").join(PROJECTION_DATABASE_FILE),
                b"corrupt projection",
            )
            .map_err(|error| adapter_error(error.to_string()))
        }

        fn startup_replay_count(&self, root: &Path) -> Result<u64, AdapterError> {
            lock(replay_counts())
                .get(&root.join("projection"))
                .copied()
                .ok_or_else(|| adapter_error("startup replay count is unavailable".to_owned()))
        }
    }

    impl ClassificationAdapter for TestAdapter {
        fn submit(&self, bytes: &[u8]) -> Result<TestId, AdapterError> {
            self.facade
                .submit(bytes)
                .map(to_test_id)
                .map_err(adapter_error)
        }

        fn status(&self, id: TestId) -> Result<TestStatus, AdapterError> {
            let status = self
                .facade
                .status(from_test_id(id))
                .map_err(adapter_error)?
                .ok_or_else(|| adapter_error("status is unavailable".to_owned()))?;
            Ok(TestStatus {
                state: match status.state {
                    OverallState::Queued => TestState::Queued,
                    OverallState::Running => TestState::Running,
                    OverallState::Failed => TestState::Failed,
                    OverallState::Completed => TestState::Completed,
                    OverallState::Confirmed => TestState::Confirmed,
                    OverallState::Discarded => TestState::Discarded,
                },
                attempt_count: status.attempt_count,
                jobs: status
                    .jobs
                    .into_iter()
                    .map(|job| TestJob {
                        attempt: job.attempt,
                        sequence: job.sequence,
                        state: match job.state {
                            LlmJobState::Running => TestJobState::Running,
                            LlmJobState::Succeeded => TestJobState::Succeeded,
                            LlmJobState::Failed => TestJobState::Failed,
                        },
                    })
                    .collect(),
                interim_transcript: status
                    .analysis
                    .as_ref()
                    .map(|analysis| analysis.envelope.analysis.transcript.clone()),
                final_transcript: status.final_transcript,
                labels: status
                    .confirmed_labels
                    .into_iter()
                    .map(|label| label.person_id)
                    .collect(),
                errors: status.errors,
                errors_truncated: status.errors_truncated,
            })
        }

        fn retry(&self, id: TestId) -> Result<(), AdapterError> {
            self.facade.retry(from_test_id(id)).map_err(adapter_error)
        }

        fn discard(&self, id: TestId) -> Result<(), AdapterError> {
            self.facade.discard(from_test_id(id)).map_err(adapter_error)
        }

        fn submit_labels(&self, id: TestId, labels: Vec<String>) -> Result<(), AdapterError> {
            let fragment_id = from_test_id(id);
            let status = self
                .facade
                .status(fragment_id)
                .map_err(adapter_error)?
                .ok_or_else(|| adapter_error("status is unavailable".to_owned()))?;
            let speakers = status
                .analysis
                .ok_or_else(|| adapter_error("analysis is unavailable".to_owned()))?
                .envelope
                .analysis
                .speakers;
            if speakers.len() != labels.len() {
                return Err(adapter_error("label count mismatch".to_owned()));
            }
            let labels = speakers
                .into_iter()
                .zip(labels)
                .map(|(speaker, person_id)| SpeakerLabelV1 {
                    speaker: speaker.speaker,
                    person_id,
                })
                .collect();
            self.facade
                .submit_labels(fragment_id, labels)
                .map_err(adapter_error)
        }

        fn inject_error_burst(&self, id: TestId, errors: Vec<String>) -> Result<(), AdapterError> {
            self.facade
                .inject_error_burst(from_test_id(id), errors)
                .map_err(adapter_error)
        }
    }

    struct NoopSubsystem;

    impl Subsystem for NoopSubsystem {
        fn submit_txn(&self, _id: TxId, _payload: &[u8]) -> Result<(), String> {
            Ok(())
        }

        fn reorg(&self) -> Result<(), String> {
            Ok(())
        }
    }

    fn to_test_id(id: FragmentId) -> TestId {
        id.into_bytes()
    }

    fn from_test_id(id: TestId) -> FragmentId {
        FragmentId::from_bytes(id)
    }

    fn adapter_error(message: String) -> AdapterError {
        Box::new(std::io::Error::other(message))
    }

    #[test]
    fn published_conformance() {
        run_all(&TestFactory).expect("audio classification conformance");
    }
}