kcode-audio-speaker-review 0.1.0

//! Human speaker-review policy and classifier application for audio packets.

#![deny(missing_docs)]
#![forbid(unsafe_code)]

use std::collections::{HashMap, HashSet};

use anyhow::{Context, ensure};
use chrono::{DateTime, Utc};
use kcode_speaker_system::{Cohort, SpeechClassifier};
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-transcript-speaker-24-freeform/2";
/// Exact classifier feature-schema cohort component.
pub const CLASSIFIER_SCHEMA_VERSION: &str = "gemini-speaker-24-normalized/1";

/// One typed speaker row normalized from a raw provider response.
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ParsedSpeaker {
    /// Exact chunk-local label, such as `Speaker 1`.
    pub local_label: String,
    /// Primary language identifier, absent with no complete profile.
    pub primary_language: Option<String>,
    /// Validated 24-value row, absent with no complete profile.
    pub feature_row: Option<FeatureRow>,
}

/// Complete normalized speaker structure for one raw provider result.
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
pub struct ParsedChunk {
    /// Whether every substantive speaker supplied a complete feature profile.
    pub clip_valid: bool,
    /// Incompleteness reason, present exactly when `clip_valid` is false.
    pub clip_validity_reason: Option<String>,
    /// One row for every substantive 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 log-likelihood score relative to background at zero.
    pub score: f64,
    /// Runner-up log-likelihood score relative to background at zero.
    pub runner_up_score: Option<f64>,
}

/// Human resolution for one chunk-local speaker.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum SpeakerResolution {
    /// A known or newly entered speaker name that may be trained.
    Known {
        /// Exact human-approved full name.
        full_name: String,
    },
    /// A deliberately unidentified speaker that must not be trained.
    Unknown,
}

/// 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.
    pub speaker_ordinal: u32,
    /// Deterministic persisted training and correction key.
    pub observation_key: ObservationKey,
    /// Best available read-only classifier evidence.
    pub candidate: Option<CandidateMapping>,
    /// Human-approved resolution, absent until this chunk is signed off.
    pub resolution: Option<SpeakerResolution>,
}

/// 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 provider response without normalization.
    pub raw_gemini_response: String,
    /// Normalized feature structure from the recording-wide parsing pass.
    pub parsed: ParsedChunk,
    /// Read-only classifier mappings for every substantive speaker.
    pub observations: Vec<CorrectionObservation>,
    /// Whether a human approved every resolution in this chunk.
    pub signed_off: 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 {
    /// One or more chunks still await human signoff.
    Unconfirmed,
    /// Legacy state retained only so old packets remain decodable.
    AutomaticallyTrained,
    /// Every chunk has explicit human signoff.
    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>,
    /// Total chronological chunk count.
    pub chunk_count: usize,
    /// Every raw response, feature row, mapping, key, resolution, and interval.
    pub chunks: Vec<CorrectionChunk>,
    /// Current durable confirmation state.
    pub confirmation_state: ConfirmationState,
}

/// One observation-level human resolution.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct ObservationConfirmation {
    /// Exact deterministic observation key from the correction packet.
    pub observation_key: ObservationKey,
    /// Caller-confirmed known or unknown resolution.
    pub resolution: SpeakerResolution,
}

/// Exact speaker resolutions and signoff for one review chunk.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct ChunkConfirmation {
    /// Recording receiving the chunk signoff.
    pub recording_id: Uuid,
    /// Exact zero-based chunk index receiving signoff.
    pub chunk_index: usize,
    /// One resolution for every chunk-local observation, with no extras.
    pub observations: Vec<ObservationConfirmation>,
}

/// Durable resolution for a finalized recording with an obsolete unsigned packet.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum LegacyReviewDisposition {
    /// Archive the obsolete result and queue retained audio for current analysis.
    Reprocess,
    /// Preserve the accepted result and stop exposing its obsolete review.
    Complete,
}

/// Validates exact, complete observation coverage for one chunk signoff.
pub fn validate_confirmation_coverage(
    packet: &CorrectionPacket,
    confirmation: &ChunkConfirmation,
) -> Result<(), String> {
    if confirmation.recording_id != packet.recording_id {
        return Err("Confirmation recording ID does not match the packet.".into());
    }
    let chunk = packet
        .chunks
        .get(confirmation.chunk_index)
        .filter(|chunk| chunk.chunk_index == confirmation.chunk_index)
        .ok_or_else(|| "Confirmation chunk does not exist.".to_owned())?;
    let known = chunk
        .observations
        .iter()
        .map(|observation| observation.observation_key.clone())
        .collect::<HashSet<_>>();
    let mut supplied = HashSet::new();
    for observation in &confirmation.observations {
        if let SpeakerResolution::Known { full_name } = &observation.resolution
            && (full_name.trim().is_empty() || full_name.chars().count() > 512)
        {
            return Err("Known speaker names must contain between 1 and 512 characters.".into());
        }
        if !supplied.insert(observation.observation_key.clone()) {
            return Err("Confirmation contains a duplicate observation key.".into());
        }
    }
    if supplied != known {
        return Err(
            "Chunk signoff must resolve every speaker exactly once, with no extras.".into(),
        );
    }
    Ok(())
}

/// Applies one exact chunk signoff to the classifier and in-memory packet.
pub fn apply_confirmation(
    classifier: &SpeechClassifier,
    packet: &mut CorrectionPacket,
    confirmation: &ChunkConfirmation,
    legacy_keys: &HashSet<ObservationKey>,
) -> anyhow::Result<()> {
    validate_confirmation_coverage(packet, confirmation).map_err(anyhow::Error::msg)?;
    let chunk_index = confirmation.chunk_index;
    if packet.chunks[chunk_index].signed_off {
        ensure!(
            confirmation_matches(packet, confirmation),
            "signed-off chunk cannot be changed"
        );
        return Ok(());
    }

    let assignments = confirmation
        .observations
        .iter()
        .map(|entry| (entry.observation_key.clone(), normalized(&entry.resolution)))
        .collect::<HashMap<_, _>>();
    let mut applied: Vec<ObservationKey> = Vec::new();
    for position in 0..packet.chunks[chunk_index].observations.len() {
        let observation = &packet.chunks[chunk_index].observations[position];
        let resolution = assignments
            .get(&observation.observation_key)
            .context("validated confirmation assignment disappeared")?
            .clone();
        if !legacy_keys.contains(&observation.observation_key) {
            let speaker = &packet.chunks[chunk_index].parsed.speakers[position];
            let result = match (
                &resolution,
                speaker.primary_language.as_deref(),
                speaker.feature_row,
            ) {
                (SpeakerResolution::Known { full_name }, Some(language), Some(row)) => classifier
                    .train(
                        observation.observation_key.clone(),
                        cohort(language),
                        row,
                        full_name.clone(),
                    )
                    .map(|_| ()),
                _ => classifier
                    .delete(observation.observation_key.clone())
                    .map(|_| ()),
            };
            if let Err(error) = result {
                let rollback = applied
                    .iter()
                    .rev()
                    .filter_map(|key| classifier.delete(key.clone()).err().map(|e| e.to_string()))
                    .collect::<Vec<_>>();
                if rollback.is_empty() {
                    anyhow::bail!("applying identity confirmations failed: {error}");
                }
                anyhow::bail!(
                    "applying identity confirmations failed: {error}; rollback also failed: {}",
                    rollback.join("; ")
                );
            }
            applied.push(observation.observation_key.clone());
        }
        packet.chunks[chunk_index].observations[position].resolution = Some(resolution);
    }
    packet.chunks[chunk_index].signed_off = true;
    if packet.chunks.iter().all(|chunk| chunk.signed_off) {
        packet.confirmation_state = ConfirmationState::Confirmed;
    }
    Ok(())
}

/// Best-effort restores classifier rows to one packet's authoritative state.
pub fn restore_training(
    classifier: &SpeechClassifier,
    packet: &CorrectionPacket,
    legacy_keys: &HashSet<ObservationKey>,
) -> Vec<String> {
    let mut errors = Vec::new();
    for chunk in packet.chunks.iter().rev() {
        for observation in chunk.observations.iter().rev() {
            if legacy_keys.contains(&observation.observation_key) {
                continue;
            }
            let speaker = chunk
                .parsed
                .speakers
                .get(observation.speaker_ordinal as usize);
            let result = match (observation.resolution.as_ref(), speaker) {
                (
                    Some(SpeakerResolution::Known { full_name }),
                    Some(ParsedSpeaker {
                        primary_language: Some(language),
                        feature_row: Some(row),
                        ..
                    }),
                ) => classifier
                    .train(
                        observation.observation_key.clone(),
                        cohort(language),
                        *row,
                        full_name.clone(),
                    )
                    .map(|_| ()),
                _ => classifier
                    .delete(observation.observation_key.clone())
                    .map(|_| ()),
            };
            if let Err(error) = result {
                errors.push(error.to_string());
            }
        }
    }
    errors
}

/// Returns the deterministic training object identity for one review chunk.
pub fn training_object_id(recording_id: Uuid, chunk_index: usize) -> String {
    format!("kcode-audio-ingress/recording/{recording_id}/chunk/{chunk_index}")
}

/// Chooses the durable resolution for an obsolete finalized review packet.
pub fn legacy_review_disposition(
    recording_complete: bool,
    confirmation_state: Option<ConfirmationState>,
    has_ingress: bool,
) -> Option<LegacyReviewDisposition> {
    if !recording_complete || confirmation_state == Some(ConfirmationState::Confirmed) {
        return None;
    }
    confirmation_state.map(|_| {
        if has_ingress {
            LegacyReviewDisposition::Complete
        } else {
            LegacyReviewDisposition::Reprocess
        }
    })
}

/// Reports whether a signed chunk already has the exact submitted mapping.
pub fn confirmation_matches(packet: &CorrectionPacket, confirmation: &ChunkConfirmation) -> bool {
    let Some(chunk) = packet
        .chunks
        .get(confirmation.chunk_index)
        .filter(|chunk| chunk.chunk_index == confirmation.chunk_index && chunk.signed_off)
    else {
        return false;
    };
    chunk.observations.len() == confirmation.observations.len()
        && confirmation.observations.iter().all(|entry| {
            chunk
                .observations
                .iter()
                .find(|stored| stored.observation_key == entry.observation_key)
                .and_then(|stored| stored.resolution.as_ref())
                == Some(&entry.resolution)
        })
}

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 normalized(resolution: &SpeakerResolution) -> SpeakerResolution {
    match resolution {
        SpeakerResolution::Known { full_name } => SpeakerResolution::Known {
            full_name: full_name.trim().to_owned(),
        },
        SpeakerResolution::Unknown => SpeakerResolution::Unknown,
    }
}

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

    fn packet(id: Uuid) -> CorrectionPacket {
        let chunks = (0..2)
            .map(|index| CorrectionChunk {
                chunk_index: index,
                chunk_count: 2,
                audio_start_ms: index as u64 * 1_000,
                audio_end_ms: (index as u64 + 1) * 1_000,
                raw_gemini_response: "Speaker 1: hello".into(),
                parsed: ParsedChunk {
                    clip_valid: true,
                    clip_validity_reason: None,
                    speakers: vec![ParsedSpeaker {
                        local_label: "Speaker 1".into(),
                        primary_language: Some("eng".into()),
                        feature_row: Some(FeatureRow::new([50; 24]).unwrap()),
                    }],
                },
                observations: vec![CorrectionObservation {
                    local_label: "Speaker 1".into(),
                    speaker_ordinal: 0,
                    observation_key: ObservationKey {
                        object_id: training_object_id(id, index),
                        piece_index: 0,
                    },
                    candidate: None,
                    resolution: None,
                }],
                signed_off: false,
            })
            .collect();
        CorrectionPacket {
            recording_id: id,
            user_id: "user".into(),
            sha256: "a".repeat(64),
            original_filename: "voice.wav".into(),
            size_bytes: 1,
            recorded_at: Utc::now(),
            chunk_count: 2,
            chunks,
            confirmation_state: ConfirmationState::Unconfirmed,
        }
    }

    #[test]
    fn signoff_trains_known_skips_unknown_and_rejects_changes() {
        let id = Uuid::new_v4();
        let path = std::env::temp_dir().join(format!("audio-review-{id}.sqlite3"));
        let classifier = SpeechClassifier::open(&path).unwrap();
        let mut review_packet = packet(id);
        let legacy = HashSet::new();
        for (index, resolution) in [
            SpeakerResolution::Known {
                full_name: "David Example".into(),
            },
            SpeakerResolution::Unknown,
        ]
        .into_iter()
        .enumerate()
        {
            let confirmation = ChunkConfirmation {
                recording_id: id,
                chunk_index: index,
                observations: vec![ObservationConfirmation {
                    observation_key: review_packet.chunks[index].observations[0]
                        .observation_key
                        .clone(),
                    resolution,
                }],
            };
            apply_confirmation(&classifier, &mut review_packet, &confirmation, &legacy).unwrap();
            apply_confirmation(&classifier, &mut review_packet, &confirmation, &legacy).unwrap();
        }
        assert_eq!(
            review_packet.confirmation_state,
            ConfirmationState::Confirmed
        );
        assert_eq!(classifier.known_speakers().unwrap(), vec!["David Example"]);
        let changed = packet(id).chunks[0].observations[0].observation_key.clone();
        let conflict = ChunkConfirmation {
            recording_id: id,
            chunk_index: 0,
            observations: vec![ObservationConfirmation {
                observation_key: changed,
                resolution: SpeakerResolution::Unknown,
            }],
        };
        assert!(apply_confirmation(&classifier, &mut review_packet, &conflict, &legacy).is_err());
        drop(classifier);
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn legacy_policy_only_selects_unresolved_finalized_packets() {
        use ConfirmationState::{AutomaticallyTrained, Unconfirmed};
        use LegacyReviewDisposition::{Complete, Reprocess};

        assert_eq!(
            legacy_review_disposition(true, Some(Unconfirmed), false),
            Some(Reprocess)
        );
        assert_eq!(
            legacy_review_disposition(true, Some(AutomaticallyTrained), true),
            Some(Complete)
        );
        assert_eq!(legacy_review_disposition(false, None, false), None);
    }
}