kcode_audio_ingress/

identity.rs

//! Typed speaker-analysis validation and classifier orchestration.

use std::{
    collections::{BTreeSet, HashMap, HashSet},
    sync::Arc,
};

use anyhow::{Context, ensure};
use chrono::{DateTime, Utc};
use kcode_speaker_extract::ExtractionOutcome;
use kcode_speaker_system::{Cohort, IdentifyEvidence, SpeechClassifier, TrainOutcome};
pub use kcode_speaker_system::{FeatureRow, ObservationKey};
use serde::{Deserialize, Serialize};
use uuid::Uuid;

/// Exact classifier provider cohort component.
pub const CLASSIFIER_PROVIDER: &str = "google";
/// Exact classifier model cohort component.
pub const CLASSIFIER_MODEL: &str = "gemini-3.1-pro-preview";
/// Exact classifier prompt-version cohort component.
pub const CLASSIFIER_PROMPT_VERSION: &str = "gemini-speaker-24-freeform/1";
/// Exact classifier feature-schema cohort component.
pub const CLASSIFIER_SCHEMA_VERSION: &str = "gemini-speaker-24-normalized/1";

const READ_ONLY_IDENTIFY_THRESHOLD: f64 = 1e308;

/// One validated chunk-local utterance parsed from the raw Gemini response.
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ParsedUtterance {
    /// Exact chunk-local speaker label.
    pub speaker: String,
    /// Lowercase ISO 639-3 language code for this utterance.
    pub language: String,
    /// Complete utterance in the language spoken.
    pub original_text: String,
    /// Complete English translation, or an empty string for English.
    pub english_translation: String,
    /// Corrected natural version for audibly non-native speech, when applicable.
    pub corrected_natural_text: Option<String>,
    /// Concise grammar, vocabulary, pronunciation, stress, and rhythm coaching.
    pub coaching: Vec<String>,
    /// Concise audible annotations that belong to this utterance.
    pub annotations: Vec<String>,
}

/// One typed speaker row parsed from a raw Gemini response.
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ParsedSpeaker {
    /// Exact chunk-local label used by the utterances.
    pub local_label: String,
    /// Lowercase ISO 639-3 code for the speaker's primary spoken language.
    pub primary_language: Option<String>,
    /// A validated 24-value row, absent when the extractor withheld a profile.
    pub feature_row: Option<FeatureRow>,
}

/// Complete validated structure parsed from one raw Gemini response.
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ParsedChunk {
    /// Complete ordered utterances for the chunk.
    pub utterances: Vec<ParsedUtterance>,
    /// Useful whole-chunk notes.
    pub notes: Vec<String>,
    /// Whether Gemini's whole-clip assessment was valid.
    pub clip_valid: bool,
    /// Brief invalidity reason, present exactly when `clip_valid` is false.
    pub clip_validity_reason: Option<String>,
    /// Exactly one typed row for each chunk-local speaker.
    pub speakers: Vec<ParsedSpeaker>,
}

/// Classifier evidence retained for one chunk-local speaker.
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
pub struct CandidateMapping {
    /// Best candidate's caller-owned full name.
    pub full_name: String,
    /// Best candidate's raw classifier cost.
    pub cost: f64,
    /// Raw classifier confidence evidence, not a probability.
    pub confidence: f64,
    /// Optional runner-up full name.
    pub runner_up_full_name: Option<String>,
    /// Optional runner-up raw cost.
    pub runner_up_cost: Option<f64>,
    /// Raw background-population cost.
    pub background_population_cost: f64,
}

/// One deterministic classifier observation in a correction packet.
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
pub struct CorrectionObservation {
    /// Chunk-local speaker label.
    pub local_label: String,
    /// Stable zero-based ordinal after labels are sorted.
    pub speaker_ordinal: u32,
    /// Deterministic persisted training and correction key.
    pub observation_key: ObservationKey,
    /// Best available read-only classifier evidence.
    pub candidate: Option<CandidateMapping>,
    /// Best candidate's full name, even when identity quality is insufficient.
    pub identified_full_name: Option<String>,
    /// Human-confirmed full name, when the confirmation API has been applied.
    pub confirmed_full_name: Option<String>,
}

/// One complete chunk in a recording-level correction packet.
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
pub struct CorrectionChunk {
    /// Zero-based chronological chunk index.
    pub chunk_index: usize,
    /// Total recording chunk count.
    pub chunk_count: usize,
    /// Source-audio start in milliseconds.
    pub audio_start_ms: u64,
    /// Source-audio end in milliseconds.
    pub audio_end_ms: u64,
    /// Complete raw Gemini response without normalization.
    pub raw_gemini_response: String,
    /// Validated GPT-parsed structure.
    pub parsed: ParsedChunk,
    /// Read-only classifier mappings for every parsed speaker.
    pub observations: Vec<CorrectionObservation>,
    /// Whether validity, confidence, background-bracketing, and one-to-one checks passed.
    pub clean: bool,
}

/// Durable identity-confirmation lifecycle for a correction packet.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ConfirmationState {
    /// No classifier observations from this packet were intentionally retained.
    Unconfirmed,
    /// The recording was clean and all observations were automatically trained.
    AutomaticallyTrained,
    /// Exact observation-level human confirmations were applied.
    Confirmed,
}

/// Complete transport-neutral correction packet for one recording.
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
pub struct CorrectionPacket {
    /// Stable recording UUID.
    pub recording_id: Uuid,
    /// Stable application user identifier associated with provider usage.
    pub user_id: String,
    /// Lowercase SHA-256 identity of the retained original bytes.
    pub sha256: String,
    /// Sanitized original filename.
    pub original_filename: String,
    /// Original retained file size in bytes.
    pub size_bytes: u64,
    /// Instant at which the recording began.
    pub recorded_at: DateTime<Utc>,
    /// Whether every chunk passed the recording-level clean gate.
    pub clean: bool,
    /// Total chronological chunk count.
    pub chunk_count: usize,
    /// Every raw response, parsed structure, row, mapping, key, and interval.
    pub chunks: Vec<CorrectionChunk>,
    /// Current durable confirmation and training state.
    pub confirmation_state: ConfirmationState,
}

/// One observation-level full-name confirmation.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct ObservationConfirmation {
    /// Exact deterministic observation key from the correction packet.
    pub observation_key: ObservationKey,
    /// Caller-confirmed full name.
    pub confirmed_full_name: String,
}

/// Exact confirmations for one completed recording.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct RecordingConfirmation {
    /// Completed recording receiving the confirmations.
    pub recording_id: Uuid,
    /// One confirmation for every known observation, with no extras.
    pub observations: Vec<ObservationConfirmation>,
}

#[derive(Clone)]
pub(crate) struct ClassificationContext {
    pub(crate) recording_id: Uuid,
    pub(crate) user_id: String,
    pub(crate) sha256: String,
    pub(crate) original_filename: String,
    pub(crate) size_bytes: u64,
    pub(crate) recorded_at: DateTime<Utc>,
    pub(crate) classifier: Arc<SpeechClassifier>,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct TranscriptChunk {
    utterances: Vec<ParsedUtterance>,
    notes: Vec<String>,
    speakers: Vec<TranscriptSpeaker>,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct TranscriptSpeaker {
    local_label: String,
    speaker_ordinal: u16,
}

pub(crate) fn parse_and_validate_chunk(
    response: &str,
    extraction: &ExtractionOutcome,
    chunk_duration_seconds: f64,
) -> anyhow::Result<ParsedChunk> {
    ensure!(
        !response.trim().is_empty(),
        "GPT parser returned an empty response"
    );
    let transcript: TranscriptChunk = serde_json::from_str(response)
        .context("GPT parser response is not one valid transcript JSON object")?;

    let (profile_count, clip_valid, clip_validity_reason) = match extraction {
        ExtractionOutcome::Scored(scored) if scored.additional_speakers.is_empty() => {
            (scored.speakers.len(), true, None)
        }
        ExtractionOutcome::Scored(scored) => (
            scored.speakers.len() + scored.additional_speakers.len(),
            false,
            Some("One or more speakers lacked a complete feature profile.".to_owned()),
        ),
        ExtractionOutcome::Unscorable {
            reason,
            additional_speakers,
        } => (additional_speakers.len(), false, Some(reason.clone())),
    };
    ensure!(
        transcript.speakers.len() == profile_count,
        "transcript and normalized speaker counts differ"
    );

    let mut speakers = transcript.speakers;
    speakers.sort_by_key(|speaker| speaker.speaker_ordinal);
    ensure!(
        speakers
            .iter()
            .enumerate()
            .all(|(index, speaker)| usize::from(speaker.speaker_ordinal) == index),
        "transcript speaker ordinals must be contiguous from zero"
    );
    let speakers = speakers
        .into_iter()
        .map(|speaker| {
            let profile = match extraction {
                ExtractionOutcome::Scored(scored) => scored
                    .speakers
                    .iter()
                    .find(|profile| profile.speaker_ordinal == speaker.speaker_ordinal),
                ExtractionOutcome::Unscorable { .. } => None,
            };
            ParsedSpeaker {
                local_label: speaker.local_label,
                primary_language: profile.map(|value| value.primary_language.as_ref().to_owned()),
                feature_row: profile.map(|value| value.features),
            }
        })
        .collect();
    let mut parsed = ParsedChunk {
        utterances: transcript.utterances,
        notes: transcript.notes,
        clip_valid,
        clip_validity_reason,
        speakers,
    };
    validate_parsed_chunk(&mut parsed, chunk_duration_seconds)?;
    Ok(parsed)
}

pub(crate) fn validate_parsed_chunk(
    parsed: &mut ParsedChunk,
    chunk_duration_seconds: f64,
) -> anyhow::Result<()> {
    ensure!(
        chunk_duration_seconds.is_finite() && chunk_duration_seconds > 0.0,
        "chunk duration must be finite and positive"
    );
    ensure!(
        parsed.notes.iter().all(|note| !note.trim().is_empty()),
        "chunk notes must not contain empty entries"
    );
    match (parsed.clip_valid, parsed.clip_validity_reason.as_deref()) {
        (true, None) => {}
        (false, Some(reason)) if !reason.trim().is_empty() => {}
        (true, Some(_)) => anyhow::bail!("a valid clip must not carry an invalidity reason"),
        (false, _) => anyhow::bail!("an invalid clip requires a brief reason"),
    }

    let mut labels = HashSet::new();
    for speaker in &parsed.speakers {
        ensure!(
            !speaker.local_label.trim().is_empty(),
            "speaker labels must not be empty"
        );
        ensure!(
            labels.insert(speaker.local_label.clone()),
            "duplicate speaker label {:?}",
            speaker.local_label
        );
        ensure!(
            speaker.primary_language.is_some() == speaker.feature_row.is_some(),
            "speaker language and feature row must be present together"
        );
        if let Some(language) = speaker.primary_language.as_deref() {
            validate_iso_639_3(language)
                .with_context(|| format!("speaker {} primary language", speaker.local_label))?;
        }
    }

    let mut referenced = HashSet::new();
    for (index, utterance) in parsed.utterances.iter().enumerate() {
        ensure!(
            labels.contains(&utterance.speaker),
            "utterance {index} references unknown speaker {:?}",
            utterance.speaker
        );
        referenced.insert(utterance.speaker.clone());
        validate_iso_639_3(&utterance.language)
            .with_context(|| format!("utterance {index} language"))?;
        ensure!(
            !utterance.original_text.trim().is_empty(),
            "utterance {index} original text must not be empty"
        );
        if utterance.language == "eng" {
            ensure!(
                utterance.english_translation.is_empty(),
                "English utterance {index} must use an empty translation"
            );
        } else {
            ensure!(
                !utterance.english_translation.trim().is_empty(),
                "non-English utterance {index} requires a complete English translation"
            );
        }
        if let Some(corrected) = utterance.corrected_natural_text.as_deref() {
            ensure!(
                !corrected.trim().is_empty(),
                "utterance {index} corrected text must not be empty"
            );
        }
        ensure!(
            utterance
                .coaching
                .iter()
                .chain(&utterance.annotations)
                .all(|entry| !entry.trim().is_empty()),
            "utterance {index} notes must not contain empty entries"
        );
    }

    ensure!(
        parsed
            .speakers
            .iter()
            .all(|speaker| referenced.contains(&speaker.local_label)),
        "every speaker row must be referenced by at least one utterance"
    );
    if parsed.clip_valid {
        ensure!(
            !parsed.speakers.is_empty()
                && parsed
                    .speakers
                    .iter()
                    .all(|speaker| speaker.feature_row.is_some()),
            "a valid clip must contain complete speaker rows"
        );
    }
    Ok(())
}

pub(crate) fn classify_speakers(
    context: &ClassificationContext,
    chunk_index: usize,
    parsed: &ParsedChunk,
) -> anyhow::Result<(Vec<CorrectionObservation>, bool)> {
    let mut observations = Vec::with_capacity(parsed.speakers.len());
    for (ordinal, speaker) in parsed.speakers.iter().enumerate() {
        let (primary_language, feature_row) = match (
            speaker.primary_language.as_deref(),
            speaker.feature_row.as_ref(),
        ) {
            (Some(language), Some(row)) => (language, row),
            (None, None) => continue,
            _ => anyhow::bail!("speaker profile is only partially present"),
        };
        let speaker_ordinal = u32::try_from(ordinal)
            .context("chunk has more speakers than the key schema supports")?;
        let cohort = cohort(primary_language);
        let probe_key = ObservationKey {
            object_id: probe_object_id(context.recording_id, chunk_index),
            piece_index: speaker_ordinal,
        };
        let outcome = context
            .classifier
            .identify(
                probe_key.clone(),
                cohort,
                *feature_row,
                READ_ONLY_IDENTIFY_THRESHOLD,
            )
            .with_context(|| {
                format!(
                    "read-only identity scoring failed for chunk {chunk_index} speaker {}",
                    speaker.local_label
                )
            })?;
        if outcome.speaker_id.is_some() {
            context
                .classifier
                .delete(probe_key)
                .context("removing an unexpectedly accepted read-only probe")?;
        }
        let candidate = outcome.evidence.as_ref().map(candidate_mapping);
        let identified_full_name = candidate.as_ref().map(|value| value.full_name.clone());
        observations.push(CorrectionObservation {
            local_label: speaker.local_label.clone(),
            speaker_ordinal,
            observation_key: ObservationKey {
                object_id: training_object_id(context.recording_id, chunk_index),
                piece_index: speaker_ordinal,
            },
            candidate,
            identified_full_name,
            confirmed_full_name: None,
        });
    }
    let clean = chunk_is_clean(parsed.clip_valid, &observations);
    Ok((observations, clean))
}

pub(crate) fn unclassified_observations(
    recording_id: Uuid,
    chunk_index: usize,
    parsed: &ParsedChunk,
) -> anyhow::Result<Vec<CorrectionObservation>> {
    parsed
        .speakers
        .iter()
        .enumerate()
        .filter(|(_, speaker)| speaker.feature_row.is_some())
        .map(|(ordinal, speaker)| {
            let speaker_ordinal = u32::try_from(ordinal)
                .context("chunk has more speakers than the key schema supports")?;
            Ok(CorrectionObservation {
                local_label: speaker.local_label.clone(),
                speaker_ordinal,
                observation_key: ObservationKey {
                    object_id: training_object_id(recording_id, chunk_index),
                    piece_index: speaker_ordinal,
                },
                candidate: None,
                identified_full_name: None,
                confirmed_full_name: None,
            })
        })
        .collect()
}

pub(crate) fn chunk_is_clean(clip_valid: bool, observations: &[CorrectionObservation]) -> bool {
    if !clip_valid || observations.is_empty() {
        return false;
    }
    let mut names = HashSet::new();
    observations.iter().all(|observation| {
        observation.candidate.as_ref().is_some_and(|candidate| {
            candidate.confidence > 0.0
                && candidate.cost < candidate.background_population_cost
                && candidate
                    .runner_up_cost
                    .is_some_and(|cost| cost > candidate.background_population_cost)
                && names.insert(candidate.full_name.as_str())
        })
    })
}

pub(crate) fn build_packet(
    context: &ClassificationContext,
    chunks: Vec<CorrectionChunk>,
) -> anyhow::Result<CorrectionPacket> {
    ensure!(!chunks.is_empty(), "correction packet has no chunks");
    let chunk_count = chunks.len();
    ensure!(
        chunks.iter().enumerate().all(|(index, chunk)| {
            chunk.chunk_index == index
                && chunk.chunk_count == chunk_count
                && chunk.audio_end_ms > chunk.audio_start_ms
                && chunk.observations.len()
                    == chunk
                        .parsed
                        .speakers
                        .iter()
                        .filter(|speaker| speaker.feature_row.is_some())
                        .count()
        }),
        "correction packet chunks are not one complete chronological plan"
    );
    let clean = chunks.iter().all(|chunk| chunk.clean);
    Ok(CorrectionPacket {
        recording_id: context.recording_id,
        user_id: context.user_id.clone(),
        sha256: context.sha256.clone(),
        original_filename: context.original_filename.clone(),
        size_bytes: context.size_bytes,
        recorded_at: context.recorded_at,
        clean,
        chunk_count,
        chunks,
        confirmation_state: ConfirmationState::Unconfirmed,
    })
}

pub(crate) fn train_clean_packet(
    classifier: &SpeechClassifier,
    packet: &mut CorrectionPacket,
) -> anyhow::Result<()> {
    if !packet.clean {
        ensure!(
            packet.confirmation_state == ConfirmationState::Unconfirmed,
            "unclean packet unexpectedly claims retained training"
        );
        return Ok(());
    }

    let mut added = Vec::new();
    for chunk in &packet.chunks {
        for observation in &chunk.observations {
            let speaker = speaker_for_observation(chunk, observation)?;
            let (primary_language, feature_row) = classifier_row(speaker)?;
            let full_name = observation
                .identified_full_name
                .as_deref()
                .context("clean observation omitted its identified full name")?;
            match classifier.train(
                observation.observation_key.clone(),
                cohort(primary_language),
                feature_row,
                full_name.to_owned(),
            ) {
                Ok(TrainOutcome::Added) => added.push(observation.observation_key.clone()),
                Ok(TrainOutcome::Unchanged | TrainOutcome::Corrected) => {}
                Err(error) => {
                    let rollback_errors = rollback_added(classifier, &added);
                    if rollback_errors.is_empty() {
                        anyhow::bail!("automatic identity training failed: {error}");
                    }
                    anyhow::bail!(
                        "automatic identity training failed: {error}; rollback also failed: {}",
                        rollback_errors.join("; ")
                    );
                }
            }
        }
    }
    packet.confirmation_state = ConfirmationState::AutomaticallyTrained;
    Ok(())
}

pub(crate) fn validate_confirmation_coverage(
    packet: &CorrectionPacket,
    confirmation: &RecordingConfirmation,
) -> Result<(), String> {
    if confirmation.recording_id != packet.recording_id {
        return Err("Confirmation recording ID does not match the packet.".into());
    }

    let known = packet
        .chunks
        .iter()
        .flat_map(|chunk| &chunk.observations)
        .map(|observation| key_tuple(&observation.observation_key))
        .collect::<BTreeSet<_>>();
    if known.is_empty() {
        return Err("The correction packet contains no speaker observations.".into());
    }

    let mut supplied = BTreeSet::new();
    for observation in &confirmation.observations {
        if observation.confirmed_full_name.trim().is_empty()
            || observation.confirmed_full_name.chars().count() > 512
        {
            return Err("Confirmed full names must contain between 1 and 512 characters.".into());
        }
        if !supplied.insert(key_tuple(&observation.observation_key)) {
            return Err("Confirmation contains a duplicate observation key.".into());
        }
    }
    if supplied != known {
        return Err(
            "Confirmation must cover every known observation exactly once, with no extras.".into(),
        );
    }
    Ok(())
}

pub(crate) fn apply_confirmations(
    classifier: &SpeechClassifier,
    packet: &mut CorrectionPacket,
    confirmation: &RecordingConfirmation,
    legacy_observation_keys: &HashSet<(String, u32)>,
) -> anyhow::Result<()> {
    validate_confirmation_coverage(packet, confirmation).map_err(anyhow::Error::msg)?;
    let assignments = confirmation
        .observations
        .iter()
        .map(|entry| {
            (
                key_tuple(&entry.observation_key),
                entry.confirmed_full_name.trim().to_owned(),
            )
        })
        .collect::<HashMap<_, _>>();

    #[derive(Clone)]
    struct Target {
        chunk_position: usize,
        observation_position: usize,
        key: ObservationKey,
        cohort: Cohort,
        row: FeatureRow,
        new_name: String,
        old_name: Option<String>,
    }

    let mut targets = Vec::new();
    let mut legacy_assignments = Vec::new();
    for (chunk_position, chunk) in packet.chunks.iter().enumerate() {
        for (observation_position, observation) in chunk.observations.iter().enumerate() {
            let new_name = assignments
                .get(&key_tuple(&observation.observation_key))
                .context("validated confirmation assignment disappeared")?
                .clone();
            if legacy_observation_keys.contains(&key_tuple(&observation.observation_key)) {
                legacy_assignments.push((chunk_position, observation_position, new_name));
                continue;
            }
            let speaker = speaker_for_observation(chunk, observation)?;
            let (primary_language, feature_row) = classifier_row(speaker)?;
            targets.push(Target {
                chunk_position,
                observation_position,
                key: observation.observation_key.clone(),
                cohort: cohort(primary_language),
                row: feature_row,
                new_name,
                old_name: retained_name(packet.confirmation_state, observation),
            });
        }
    }

    let mut applied = Vec::<(Target, TrainOutcome)>::new();
    for target in targets {
        match classifier.train(
            target.key.clone(),
            target.cohort.clone(),
            target.row,
            target.new_name.clone(),
        ) {
            Ok(outcome) => applied.push((target, outcome)),
            Err(error) => {
                let mut rollback_errors = Vec::new();
                for (previous, outcome) in applied.iter().rev() {
                    let rollback = if let Some(old_name) = &previous.old_name {
                        classifier
                            .train(
                                previous.key.clone(),
                                previous.cohort.clone(),
                                previous.row,
                                old_name.clone(),
                            )
                            .map(|_| ())
                    } else if *outcome == TrainOutcome::Added {
                        classifier.delete(previous.key.clone()).map(|_| ())
                    } else {
                        Ok(())
                    };
                    if let Err(rollback_error) = rollback {
                        rollback_errors.push(rollback_error.to_string());
                    }
                }
                if rollback_errors.is_empty() {
                    anyhow::bail!("applying identity confirmations failed: {error}");
                }
                anyhow::bail!(
                    "applying identity confirmations failed: {error}; rollback also failed: {}",
                    rollback_errors.join("; ")
                );
            }
        }
    }

    for (target, _) in applied {
        packet.chunks[target.chunk_position].observations[target.observation_position]
            .confirmed_full_name = Some(target.new_name);
    }
    for (chunk_position, observation_position, name) in legacy_assignments {
        packet.chunks[chunk_position].observations[observation_position].confirmed_full_name =
            Some(name);
    }
    packet.confirmation_state = ConfirmationState::Confirmed;
    Ok(())
}

pub(crate) fn restore_packet_training(
    classifier: &SpeechClassifier,
    packet: &CorrectionPacket,
    legacy_observation_keys: &HashSet<(String, u32)>,
) -> Vec<String> {
    let mut errors = Vec::new();
    for chunk in packet.chunks.iter().rev() {
        for observation in chunk.observations.iter().rev() {
            if legacy_observation_keys.contains(&key_tuple(&observation.observation_key)) {
                continue;
            }
            let result = match retained_name(packet.confirmation_state, observation) {
                Some(name) => speaker_for_observation(chunk, observation).and_then(|speaker| {
                    let (primary_language, feature_row) = classifier_row(speaker)?;
                    classifier
                        .train(
                            observation.observation_key.clone(),
                            cohort(primary_language),
                            feature_row,
                            name,
                        )
                        .map(|_| ())
                        .map_err(anyhow::Error::from)
                }),
                None => classifier
                    .delete(observation.observation_key.clone())
                    .map(|_| ())
                    .map_err(anyhow::Error::from),
            };
            if let Err(error) = result {
                errors.push(error.to_string());
            }
        }
    }
    errors
}

pub(crate) fn training_object_id(recording_id: Uuid, chunk_index: usize) -> String {
    format!("kcode-audio-ingress/recording/{recording_id}/chunk/{chunk_index}")
}

fn probe_object_id(recording_id: Uuid, chunk_index: usize) -> String {
    format!("kcode-audio-ingress/probe/{recording_id}/chunk/{chunk_index}")
}

fn cohort(primary_language: &str) -> Cohort {
    Cohort {
        provider: CLASSIFIER_PROVIDER.into(),
        model: CLASSIFIER_MODEL.into(),
        prompt_version: CLASSIFIER_PROMPT_VERSION.into(),
        schema_version: CLASSIFIER_SCHEMA_VERSION.into(),
        primary_language: primary_language.into(),
    }
}

fn candidate_mapping(evidence: &IdentifyEvidence) -> CandidateMapping {
    CandidateMapping {
        full_name: evidence.best.speaker_id.clone(),
        cost: evidence.best.cost,
        confidence: evidence.confidence_score,
        runner_up_full_name: evidence
            .runner_up
            .as_ref()
            .map(|candidate| candidate.speaker_id.clone()),
        runner_up_cost: evidence.runner_up.as_ref().map(|candidate| candidate.cost),
        background_population_cost: evidence.background_population_cost,
    }
}

fn speaker_for_observation<'a>(
    chunk: &'a CorrectionChunk,
    observation: &CorrectionObservation,
) -> anyhow::Result<&'a ParsedSpeaker> {
    let speaker = chunk
        .parsed
        .speakers
        .get(observation.speaker_ordinal as usize)
        .context("observation ordinal is outside the parsed speaker rows")?;
    ensure!(
        speaker.local_label == observation.local_label,
        "observation label does not match its parsed speaker row"
    );
    Ok(speaker)
}

fn classifier_row(speaker: &ParsedSpeaker) -> anyhow::Result<(&str, FeatureRow)> {
    Ok((
        speaker
            .primary_language
            .as_deref()
            .context("speaker observation omitted its primary language")?,
        speaker
            .feature_row
            .context("speaker observation omitted its feature row")?,
    ))
}

fn retained_name(state: ConfirmationState, observation: &CorrectionObservation) -> Option<String> {
    match state {
        ConfirmationState::Unconfirmed => None,
        ConfirmationState::AutomaticallyTrained => observation.identified_full_name.clone(),
        ConfirmationState::Confirmed => observation.confirmed_full_name.clone(),
    }
}

fn rollback_added(classifier: &SpeechClassifier, keys: &[ObservationKey]) -> Vec<String> {
    let mut errors = Vec::new();
    for key in keys.iter().rev() {
        if let Err(error) = classifier.delete(key.clone()) {
            errors.push(error.to_string());
        }
    }
    errors
}

fn key_tuple(key: &ObservationKey) -> (String, u32) {
    (key.object_id.clone(), key.piece_index)
}

fn validate_iso_639_3(value: &str) -> anyhow::Result<()> {
    ensure!(
        value.len() == 3 && value.bytes().all(|byte| byte.is_ascii_lowercase()),
        "must be a lowercase ISO 639-3 code"
    );
    ensure!(
        !matches!(value, "mis" | "mul" | "und" | "zxx"),
        "must identify one primary spoken language"
    );
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use kcode_speaker_extract::{AdditionalSpeaker, CompleteSpeaker, ScoredClip};
    use kcode_speaker_system::DeleteOutcome;
    use serde_json::{Value, json};
    use std::{
        fs,
        path::{Path, PathBuf},
        sync::atomic::{AtomicU64, Ordering},
    };

    static NEXT_PATH: AtomicU64 = AtomicU64::new(0);

    fn database_path(label: &str) -> PathBuf {
        std::env::temp_dir().join(format!(
            "kcode-audio-ingress-identity-{}-{label}-{}.sqlite3",
            std::process::id(),
            NEXT_PATH.fetch_add(1, Ordering::Relaxed)
        ))
    }

    fn remove_database(path: &Path) {
        for suffix in ["", "-wal", "-shm"] {
            let mut value = path.as_os_str().to_os_string();
            value.push(suffix);
            let _ = fs::remove_file(PathBuf::from(value));
        }
    }

    fn row() -> FeatureRow {
        FeatureRow::new(std::array::from_fn(|index| {
            u8::try_from(index * 3 + 10).unwrap()
        }))
        .unwrap()
    }

    fn extraction() -> ExtractionOutcome {
        ExtractionOutcome::Scored(ScoredClip {
            speakers: vec![CompleteSpeaker {
                speaker_ordinal: 0,
                primary_language: kcode_speaker_extract::Key::parse("eng").unwrap(),
                closest_dialect: "General American English".into(),
                usable_speech_ms: 1_000,
                features: row(),
            }],
            additional_speakers: Vec::new(),
        })
    }

    fn parsed() -> ParsedChunk {
        ParsedChunk {
            utterances: vec![ParsedUtterance {
                speaker: "Speaker A".into(),
                language: "eng".into(),
                original_text: "Hello.".into(),
                english_translation: String::new(),
                corrected_natural_text: None,
                coaching: Vec::new(),
                annotations: Vec::new(),
            }],
            notes: vec!["Clear recording.".into()],
            clip_valid: true,
            clip_validity_reason: None,
            speakers: vec![ParsedSpeaker {
                local_label: "Speaker A".into(),
                primary_language: Some("eng".into()),
                feature_row: Some(row()),
            }],
        }
    }

    fn observation(name: &str, confidence: f64, ordinal: u32) -> CorrectionObservation {
        CorrectionObservation {
            local_label: format!("Speaker {}", char::from(b'A' + ordinal as u8)),
            speaker_ordinal: ordinal,
            observation_key: ObservationKey {
                object_id: "recording".into(),
                piece_index: ordinal,
            },
            candidate: Some(CandidateMapping {
                full_name: name.into(),
                cost: 1.0,
                confidence,
                runner_up_full_name: Some("Runner Up".into()),
                runner_up_cost: Some(4.0),
                background_population_cost: 3.0,
            }),
            identified_full_name: Some(name.into()),
            confirmed_full_name: None,
        }
    }

    fn packet(clean: bool) -> CorrectionPacket {
        CorrectionPacket {
            recording_id: Uuid::nil(),
            user_id: "user".into(),
            sha256: "0".repeat(64),
            original_filename: "audio.wav".into(),
            size_bytes: 44,
            recorded_at: DateTime::parse_from_rfc3339("2026-01-01T00:00:00Z")
                .unwrap()
                .with_timezone(&Utc),
            clean,
            chunk_count: 1,
            chunks: vec![CorrectionChunk {
                chunk_index: 0,
                chunk_count: 1,
                audio_start_ms: 0,
                audio_end_ms: 1_000,
                raw_gemini_response: "raw".into(),
                parsed: parsed(),
                observations: vec![observation("David Example", 2.0, 0)],
                clean,
            }],
            confirmation_state: ConfirmationState::Unconfirmed,
        }
    }

    #[test]
    fn parser_merges_frozen_features_and_requires_matching_speakers() {
        let valid = json!({
            "utterances": [{
                "speaker":"Speaker A",
                "language":"eng",
                "original_text":"Hello.",
                "english_translation":"",
                "corrected_natural_text":null,
                "coaching":[],
                "annotations":[]
            }],
            "notes":["Clear recording."],
            "speakers":[{"local_label":"Speaker A", "speaker_ordinal":0}]
        });
        let restored = parse_and_validate_chunk(&valid.to_string(), &extraction(), 2.0).unwrap();
        assert_eq!(restored, parsed());

        let mut unknown = valid.clone();
        unknown["utterances"][0]["speaker"] = Value::String("Speaker Z".into());
        assert!(parse_and_validate_chunk(&unknown.to_string(), &extraction(), 2.0).is_err());

        let mut wrong_ordinal = valid.clone();
        wrong_ordinal["speakers"][0]["speaker_ordinal"] = json!(1);
        assert!(parse_and_validate_chunk(&wrong_ordinal.to_string(), &extraction(), 2.0).is_err());

        let incomplete = ExtractionOutcome::Scored(ScoredClip {
            speakers: Vec::new(),
            additional_speakers: vec![AdditionalSpeaker {
                speaker_ordinal: 0,
                description: "Too little speech.".into(),
            }],
        });
        let restored = parse_and_validate_chunk(&valid.to_string(), &incomplete, 2.0).unwrap();
        assert!(
            !restored.clip_valid
                && restored.speakers[0].primary_language.is_none()
                && restored.speakers[0].feature_row.is_none()
        );
    }

    #[test]
    fn deterministic_keys_are_unique_for_multi_speaker_chunks() {
        let recording = Uuid::new_v4();
        let first = training_object_id(recording, 0);
        let second = training_object_id(recording, 1);
        assert_ne!(first, second);
        let keys = [
            ObservationKey {
                object_id: first.clone(),
                piece_index: 0,
            },
            ObservationKey {
                object_id: first,
                piece_index: 1,
            },
            ObservationKey {
                object_id: second,
                piece_index: 0,
            },
        ];
        assert_eq!(
            keys.iter().map(key_tuple).collect::<BTreeSet<_>>().len(),
            keys.len()
        );
    }

    #[test]
    fn clean_gate_requires_background_bracketing_and_unique_candidates() {
        let bracketed = observation("David Example", 1.0, 0);
        assert!(chunk_is_clean(true, std::slice::from_ref(&bracketed)));
        assert!(!chunk_is_clean(true, &[]));

        let mut missing_candidate = bracketed.clone();
        missing_candidate.candidate = None;
        assert!(!chunk_is_clean(true, &[missing_candidate]));

        let zero_confidence = observation("David Example", 0.0, 0);
        assert!(!chunk_is_clean(true, &[zero_confidence]));
        let negative_confidence = observation("David Example", -1.0, 0);
        assert!(!chunk_is_clean(true, &[negative_confidence]));

        let mut best_equal = bracketed.clone();
        best_equal.candidate.as_mut().unwrap().cost = 3.0;
        assert!(!chunk_is_clean(true, &[best_equal]));
        let mut best_greater = bracketed.clone();
        best_greater.candidate.as_mut().unwrap().cost = 4.0;
        assert!(!chunk_is_clean(true, &[best_greater]));

        let mut runner_up_absent = bracketed.clone();
        let candidate = runner_up_absent.candidate.as_mut().unwrap();
        candidate.runner_up_full_name = None;
        candidate.runner_up_cost = None;
        assert!(!chunk_is_clean(true, &[runner_up_absent]));

        let mut runner_up_equal = bracketed.clone();
        runner_up_equal.candidate.as_mut().unwrap().runner_up_cost = Some(3.0);
        assert!(!chunk_is_clean(true, &[runner_up_equal]));
        let mut runner_up_below = bracketed.clone();
        runner_up_below.candidate.as_mut().unwrap().runner_up_cost = Some(2.0);
        assert!(!chunk_is_clean(true, &[runner_up_below]));

        assert!(!chunk_is_clean(
            true,
            &[bracketed.clone(), observation("David Example", 2.0, 1),]
        ));
        assert!(!chunk_is_clean(false, &[bracketed]));
    }

    #[test]
    fn recording_training_gate_trains_only_clean_packets() {
        let path = database_path("training-gate");
        let classifier = SpeechClassifier::open(&path).unwrap();
        let mut unclean = packet(false);
        train_clean_packet(&classifier, &mut unclean).unwrap();
        assert_eq!(
            classifier
                .delete(unclean.chunks[0].observations[0].observation_key.clone())
                .unwrap(),
            DeleteOutcome::NotFound
        );

        let mut clean = packet(true);
        train_clean_packet(&classifier, &mut clean).unwrap();
        assert_eq!(
            clean.confirmation_state,
            ConfirmationState::AutomaticallyTrained
        );
        assert_eq!(
            classifier
                .delete(clean.chunks[0].observations[0].observation_key.clone())
                .unwrap(),
            DeleteOutcome::Deleted
        );
        drop(classifier);
        remove_database(&path);
    }

    #[test]
    fn confirmation_requires_exact_coverage() {
        let packet = packet(false);
        let key = packet.chunks[0].observations[0].observation_key.clone();
        let exact = RecordingConfirmation {
            recording_id: packet.recording_id,
            observations: vec![ObservationConfirmation {
                observation_key: key.clone(),
                confirmed_full_name: "David Example".into(),
            }],
        };
        assert!(validate_confirmation_coverage(&packet, &exact).is_ok());

        let duplicate = RecordingConfirmation {
            recording_id: packet.recording_id,
            observations: vec![
                ObservationConfirmation {
                    observation_key: key.clone(),
                    confirmed_full_name: "David Example".into(),
                },
                ObservationConfirmation {
                    observation_key: key,
                    confirmed_full_name: "David Example".into(),
                },
            ],
        };
        assert!(validate_confirmation_coverage(&packet, &duplicate).is_err());

        let empty = RecordingConfirmation {
            recording_id: packet.recording_id,
            observations: Vec::new(),
        };
        assert!(validate_confirmation_coverage(&packet, &empty).is_err());
    }
}