kcode_audio_speaker_review/
lib.rs

1//! Human speaker-review policy and classifier application for audio packets.
2
3#![deny(missing_docs)]
4#![forbid(unsafe_code)]
5
6use std::collections::{HashMap, HashSet};
7
8use anyhow::{Context, ensure};
9use chrono::{DateTime, Utc};
10use kcode_speaker_system::{Cohort, SpeechClassifier};
11pub use kcode_speaker_system::{FeatureRow, ObservationKey};
12use serde::{Deserialize, Serialize};
13use uuid::Uuid;
14
15/// Exact classifier provider cohort component.
16pub const CLASSIFIER_PROVIDER: &str = "google";
17/// Exact classifier model cohort component.
18pub const CLASSIFIER_MODEL: &str = "gemini-3.1-pro-preview";
19/// Exact classifier prompt-version cohort component.
20pub const CLASSIFIER_PROMPT_VERSION: &str = "gemini-transcript-speaker-24-freeform/2";
21/// Exact classifier feature-schema cohort component.
22pub const CLASSIFIER_SCHEMA_VERSION: &str = "gemini-speaker-24-normalized/1";
23
24/// One typed speaker row normalized from a raw provider response.
25#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
26#[serde(deny_unknown_fields)]
27pub struct ParsedSpeaker {
28    /// Exact chunk-local label, such as `Speaker 1`.
29    pub local_label: String,
30    /// Primary language identifier, absent with no complete profile.
31    pub primary_language: Option<String>,
32    /// Validated 24-value row, absent with no complete profile.
33    pub feature_row: Option<FeatureRow>,
34}
35
36/// Complete normalized speaker structure for one raw provider result.
37#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
38pub struct ParsedChunk {
39    /// Whether every substantive speaker supplied a complete feature profile.
40    pub clip_valid: bool,
41    /// Incompleteness reason, present exactly when `clip_valid` is false.
42    pub clip_validity_reason: Option<String>,
43    /// One row for every substantive chunk-local speaker.
44    pub speakers: Vec<ParsedSpeaker>,
45}
46
47/// Classifier evidence retained for one chunk-local speaker.
48#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
49pub struct CandidateMapping {
50    /// Best candidate's caller-owned full name.
51    pub full_name: String,
52    /// Best-candidate log-likelihood score relative to background at zero.
53    pub score: f64,
54    /// Runner-up log-likelihood score relative to background at zero.
55    pub runner_up_score: Option<f64>,
56}
57
58/// Human resolution for one chunk-local speaker.
59#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
60#[serde(tag = "kind", rename_all = "snake_case")]
61pub enum SpeakerResolution {
62    /// A known or newly entered speaker name that may be trained.
63    Known {
64        /// Exact human-approved full name.
65        full_name: String,
66    },
67    /// A deliberately unidentified speaker that must not be trained.
68    Unknown,
69}
70
71/// One deterministic classifier observation in a correction packet.
72#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
73pub struct CorrectionObservation {
74    /// Chunk-local speaker label.
75    pub local_label: String,
76    /// Stable zero-based ordinal.
77    pub speaker_ordinal: u32,
78    /// Deterministic persisted training and correction key.
79    pub observation_key: ObservationKey,
80    /// Best available read-only classifier evidence.
81    pub candidate: Option<CandidateMapping>,
82    /// Human-approved resolution, absent until this chunk is signed off.
83    pub resolution: Option<SpeakerResolution>,
84}
85
86/// One complete chunk in a recording-level correction packet.
87#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
88pub struct CorrectionChunk {
89    /// Zero-based chronological chunk index.
90    pub chunk_index: usize,
91    /// Total recording chunk count.
92    pub chunk_count: usize,
93    /// Source-audio start in milliseconds.
94    pub audio_start_ms: u64,
95    /// Source-audio end in milliseconds.
96    pub audio_end_ms: u64,
97    /// Complete raw provider response without normalization.
98    pub raw_gemini_response: String,
99    /// Normalized feature structure from the recording-wide parsing pass.
100    pub parsed: ParsedChunk,
101    /// Read-only classifier mappings for every substantive speaker.
102    pub observations: Vec<CorrectionObservation>,
103    /// Whether a human approved every resolution in this chunk.
104    pub signed_off: bool,
105}
106
107/// Durable identity-confirmation lifecycle for a correction packet.
108#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
109#[serde(rename_all = "snake_case")]
110pub enum ConfirmationState {
111    /// One or more chunks still await human signoff.
112    Unconfirmed,
113    /// Legacy state retained only so old packets remain decodable.
114    AutomaticallyTrained,
115    /// Every chunk has explicit human signoff.
116    Confirmed,
117}
118
119/// Complete transport-neutral correction packet for one recording.
120#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
121pub struct CorrectionPacket {
122    /// Stable recording UUID.
123    pub recording_id: Uuid,
124    /// Stable application user identifier associated with provider usage.
125    pub user_id: String,
126    /// Lowercase SHA-256 identity of the retained original bytes.
127    pub sha256: String,
128    /// Sanitized original filename.
129    pub original_filename: String,
130    /// Original retained file size in bytes.
131    pub size_bytes: u64,
132    /// Instant at which the recording began.
133    pub recorded_at: DateTime<Utc>,
134    /// Total chronological chunk count.
135    pub chunk_count: usize,
136    /// Every raw response, feature row, mapping, key, resolution, and interval.
137    pub chunks: Vec<CorrectionChunk>,
138    /// Current durable confirmation state.
139    pub confirmation_state: ConfirmationState,
140}
141
142/// One observation-level human resolution.
143#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
144pub struct ObservationConfirmation {
145    /// Exact deterministic observation key from the correction packet.
146    pub observation_key: ObservationKey,
147    /// Caller-confirmed known or unknown resolution.
148    pub resolution: SpeakerResolution,
149}
150
151/// Exact speaker resolutions and signoff for one review chunk.
152#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
153pub struct ChunkConfirmation {
154    /// Recording receiving the chunk signoff.
155    pub recording_id: Uuid,
156    /// Exact zero-based chunk index receiving signoff.
157    pub chunk_index: usize,
158    /// One resolution for every chunk-local observation, with no extras.
159    pub observations: Vec<ObservationConfirmation>,
160}
161
162/// Durable resolution for a finalized recording with an obsolete unsigned packet.
163#[derive(Clone, Copy, Debug, Eq, PartialEq)]
164pub enum LegacyReviewDisposition {
165    /// Archive the obsolete result and queue retained audio for current analysis.
166    Reprocess,
167    /// Preserve the accepted result and stop exposing its obsolete review.
168    Complete,
169}
170
171/// Validates exact, complete observation coverage for one chunk signoff.
172pub fn validate_confirmation_coverage(
173    packet: &CorrectionPacket,
174    confirmation: &ChunkConfirmation,
175) -> Result<(), String> {
176    if confirmation.recording_id != packet.recording_id {
177        return Err("Confirmation recording ID does not match the packet.".into());
178    }
179    let chunk = packet
180        .chunks
181        .get(confirmation.chunk_index)
182        .filter(|chunk| chunk.chunk_index == confirmation.chunk_index)
183        .ok_or_else(|| "Confirmation chunk does not exist.".to_owned())?;
184    let known = chunk
185        .observations
186        .iter()
187        .map(|observation| observation.observation_key.clone())
188        .collect::<HashSet<_>>();
189    let mut supplied = HashSet::new();
190    for observation in &confirmation.observations {
191        if let SpeakerResolution::Known { full_name } = &observation.resolution
192            && (full_name.trim().is_empty() || full_name.chars().count() > 512)
193        {
194            return Err("Known speaker names must contain between 1 and 512 characters.".into());
195        }
196        if !supplied.insert(observation.observation_key.clone()) {
197            return Err("Confirmation contains a duplicate observation key.".into());
198        }
199    }
200    if supplied != known {
201        return Err(
202            "Chunk signoff must resolve every speaker exactly once, with no extras.".into(),
203        );
204    }
205    Ok(())
206}
207
208/// Applies one exact chunk signoff to the classifier and in-memory packet.
209pub fn apply_confirmation(
210    classifier: &SpeechClassifier,
211    packet: &mut CorrectionPacket,
212    confirmation: &ChunkConfirmation,
213    legacy_keys: &HashSet<ObservationKey>,
214) -> anyhow::Result<()> {
215    validate_confirmation_coverage(packet, confirmation).map_err(anyhow::Error::msg)?;
216    let chunk_index = confirmation.chunk_index;
217    if packet.chunks[chunk_index].signed_off {
218        ensure!(
219            confirmation_matches(packet, confirmation),
220            "signed-off chunk cannot be changed"
221        );
222        return Ok(());
223    }
224
225    let assignments = confirmation
226        .observations
227        .iter()
228        .map(|entry| (entry.observation_key.clone(), normalized(&entry.resolution)))
229        .collect::<HashMap<_, _>>();
230    let mut applied: Vec<ObservationKey> = Vec::new();
231    for position in 0..packet.chunks[chunk_index].observations.len() {
232        let observation = &packet.chunks[chunk_index].observations[position];
233        let resolution = assignments
234            .get(&observation.observation_key)
235            .context("validated confirmation assignment disappeared")?
236            .clone();
237        if !legacy_keys.contains(&observation.observation_key) {
238            let speaker = &packet.chunks[chunk_index].parsed.speakers[position];
239            let result = match (
240                &resolution,
241                speaker.primary_language.as_deref(),
242                speaker.feature_row,
243            ) {
244                (SpeakerResolution::Known { full_name }, Some(language), Some(row)) => classifier
245                    .train(
246                        observation.observation_key.clone(),
247                        cohort(language),
248                        row,
249                        full_name.clone(),
250                    )
251                    .map(|_| ()),
252                _ => classifier
253                    .delete(observation.observation_key.clone())
254                    .map(|_| ()),
255            };
256            if let Err(error) = result {
257                let rollback = applied
258                    .iter()
259                    .rev()
260                    .filter_map(|key| classifier.delete(key.clone()).err().map(|e| e.to_string()))
261                    .collect::<Vec<_>>();
262                if rollback.is_empty() {
263                    anyhow::bail!("applying identity confirmations failed: {error}");
264                }
265                anyhow::bail!(
266                    "applying identity confirmations failed: {error}; rollback also failed: {}",
267                    rollback.join("; ")
268                );
269            }
270            applied.push(observation.observation_key.clone());
271        }
272        packet.chunks[chunk_index].observations[position].resolution = Some(resolution);
273    }
274    packet.chunks[chunk_index].signed_off = true;
275    if packet.chunks.iter().all(|chunk| chunk.signed_off) {
276        packet.confirmation_state = ConfirmationState::Confirmed;
277    }
278    Ok(())
279}
280
281/// Best-effort restores classifier rows to one packet's authoritative state.
282pub fn restore_training(
283    classifier: &SpeechClassifier,
284    packet: &CorrectionPacket,
285    legacy_keys: &HashSet<ObservationKey>,
286) -> Vec<String> {
287    let mut errors = Vec::new();
288    for chunk in packet.chunks.iter().rev() {
289        for observation in chunk.observations.iter().rev() {
290            if legacy_keys.contains(&observation.observation_key) {
291                continue;
292            }
293            let speaker = chunk
294                .parsed
295                .speakers
296                .get(observation.speaker_ordinal as usize);
297            let result = match (observation.resolution.as_ref(), speaker) {
298                (
299                    Some(SpeakerResolution::Known { full_name }),
300                    Some(ParsedSpeaker {
301                        primary_language: Some(language),
302                        feature_row: Some(row),
303                        ..
304                    }),
305                ) => classifier
306                    .train(
307                        observation.observation_key.clone(),
308                        cohort(language),
309                        *row,
310                        full_name.clone(),
311                    )
312                    .map(|_| ()),
313                _ => classifier
314                    .delete(observation.observation_key.clone())
315                    .map(|_| ()),
316            };
317            if let Err(error) = result {
318                errors.push(error.to_string());
319            }
320        }
321    }
322    errors
323}
324
325/// Returns the deterministic training object identity for one review chunk.
326pub fn training_object_id(recording_id: Uuid, chunk_index: usize) -> String {
327    format!("kcode-audio-ingress/recording/{recording_id}/chunk/{chunk_index}")
328}
329
330/// Chooses the durable resolution for an obsolete finalized review packet.
331pub fn legacy_review_disposition(
332    recording_complete: bool,
333    confirmation_state: Option<ConfirmationState>,
334    has_ingress: bool,
335) -> Option<LegacyReviewDisposition> {
336    if !recording_complete || confirmation_state == Some(ConfirmationState::Confirmed) {
337        return None;
338    }
339    confirmation_state.map(|_| {
340        if has_ingress {
341            LegacyReviewDisposition::Complete
342        } else {
343            LegacyReviewDisposition::Reprocess
344        }
345    })
346}
347
348/// Reports whether a signed chunk already has the exact submitted mapping.
349pub fn confirmation_matches(packet: &CorrectionPacket, confirmation: &ChunkConfirmation) -> bool {
350    let Some(chunk) = packet
351        .chunks
352        .get(confirmation.chunk_index)
353        .filter(|chunk| chunk.chunk_index == confirmation.chunk_index && chunk.signed_off)
354    else {
355        return false;
356    };
357    chunk.observations.len() == confirmation.observations.len()
358        && confirmation.observations.iter().all(|entry| {
359            chunk
360                .observations
361                .iter()
362                .find(|stored| stored.observation_key == entry.observation_key)
363                .and_then(|stored| stored.resolution.as_ref())
364                == Some(&entry.resolution)
365        })
366}
367
368fn cohort(primary_language: &str) -> Cohort {
369    Cohort {
370        provider: CLASSIFIER_PROVIDER.into(),
371        model: CLASSIFIER_MODEL.into(),
372        prompt_version: CLASSIFIER_PROMPT_VERSION.into(),
373        schema_version: CLASSIFIER_SCHEMA_VERSION.into(),
374        primary_language: primary_language.into(),
375    }
376}
377
378fn normalized(resolution: &SpeakerResolution) -> SpeakerResolution {
379    match resolution {
380        SpeakerResolution::Known { full_name } => SpeakerResolution::Known {
381            full_name: full_name.trim().to_owned(),
382        },
383        SpeakerResolution::Unknown => SpeakerResolution::Unknown,
384    }
385}
386
387#[cfg(test)]
388mod tests {
389    use super::*;
390
391    fn packet(id: Uuid) -> CorrectionPacket {
392        let chunks = (0..2)
393            .map(|index| CorrectionChunk {
394                chunk_index: index,
395                chunk_count: 2,
396                audio_start_ms: index as u64 * 1_000,
397                audio_end_ms: (index as u64 + 1) * 1_000,
398                raw_gemini_response: "Speaker 1: hello".into(),
399                parsed: ParsedChunk {
400                    clip_valid: true,
401                    clip_validity_reason: None,
402                    speakers: vec![ParsedSpeaker {
403                        local_label: "Speaker 1".into(),
404                        primary_language: Some("eng".into()),
405                        feature_row: Some(FeatureRow::new([50; 24]).unwrap()),
406                    }],
407                },
408                observations: vec![CorrectionObservation {
409                    local_label: "Speaker 1".into(),
410                    speaker_ordinal: 0,
411                    observation_key: ObservationKey {
412                        object_id: training_object_id(id, index),
413                        piece_index: 0,
414                    },
415                    candidate: None,
416                    resolution: None,
417                }],
418                signed_off: false,
419            })
420            .collect();
421        CorrectionPacket {
422            recording_id: id,
423            user_id: "user".into(),
424            sha256: "a".repeat(64),
425            original_filename: "voice.wav".into(),
426            size_bytes: 1,
427            recorded_at: Utc::now(),
428            chunk_count: 2,
429            chunks,
430            confirmation_state: ConfirmationState::Unconfirmed,
431        }
432    }
433
434    #[test]
435    fn signoff_trains_known_skips_unknown_and_rejects_changes() {
436        let id = Uuid::new_v4();
437        let path = std::env::temp_dir().join(format!("audio-review-{id}.sqlite3"));
438        let classifier = SpeechClassifier::open(&path).unwrap();
439        let mut review_packet = packet(id);
440        let legacy = HashSet::new();
441        for (index, resolution) in [
442            SpeakerResolution::Known {
443                full_name: "David Example".into(),
444            },
445            SpeakerResolution::Unknown,
446        ]
447        .into_iter()
448        .enumerate()
449        {
450            let confirmation = ChunkConfirmation {
451                recording_id: id,
452                chunk_index: index,
453                observations: vec![ObservationConfirmation {
454                    observation_key: review_packet.chunks[index].observations[0]
455                        .observation_key
456                        .clone(),
457                    resolution,
458                }],
459            };
460            apply_confirmation(&classifier, &mut review_packet, &confirmation, &legacy).unwrap();
461            apply_confirmation(&classifier, &mut review_packet, &confirmation, &legacy).unwrap();
462        }
463        assert_eq!(
464            review_packet.confirmation_state,
465            ConfirmationState::Confirmed
466        );
467        assert_eq!(classifier.known_speakers().unwrap(), vec!["David Example"]);
468        let changed = packet(id).chunks[0].observations[0].observation_key.clone();
469        let conflict = ChunkConfirmation {
470            recording_id: id,
471            chunk_index: 0,
472            observations: vec![ObservationConfirmation {
473                observation_key: changed,
474                resolution: SpeakerResolution::Unknown,
475            }],
476        };
477        assert!(apply_confirmation(&classifier, &mut review_packet, &conflict, &legacy).is_err());
478        drop(classifier);
479        let _ = std::fs::remove_file(path);
480    }
481
482    #[test]
483    fn legacy_policy_only_selects_unresolved_finalized_packets() {
484        use ConfirmationState::{AutomaticallyTrained, Unconfirmed};
485        use LegacyReviewDisposition::{Complete, Reprocess};
486
487        assert_eq!(
488            legacy_review_disposition(true, Some(Unconfirmed), false),
489            Some(Reprocess)
490        );
491        assert_eq!(
492            legacy_review_disposition(true, Some(AutomaticallyTrained), true),
493            Some(Complete)
494        );
495        assert_eq!(legacy_review_disposition(false, None, false), None);
496    }
497}
kcode_audio_speaker_review/lib.rs

kcode_audio_speaker_review/
lib.rs
