use serde::{Deserialize, Serialize};
use std::{collections::BTreeSet, error::Error, fmt};
pub use kcode_speaker_v3_llm_protocol::{
GEMINI_FEATURE_PROMPT_ONE, GEMINI_FEATURE_PROMPT_ONE_REVISION, GEMINI_FEATURE_PROMPT_REVISIONS,
GEMINI_FEATURE_PROMPT_THREE, GEMINI_FEATURE_PROMPT_THREE_REVISION, GEMINI_FEATURE_PROMPT_TWO,
GEMINI_FEATURE_PROMPT_TWO_REVISION, GEMINI_TRANSCRIPT_PROMPT,
GEMINI_TRANSCRIPT_PROMPT_REVISION, GPT_STRUCTURING_PROMPT, GPT_STRUCTURING_PROMPT_REVISION,
};
pub use kcode_speaker_v3_schema::{
FEATURE_NAMES, FEATURE_SCHEMA_REVISION, FeatureVector24, LocalSpeakerLabel,
MAX_AUDIO_DURATION_MS, OGG_MEDIA_TYPE, OggAudioMetadata, StructuredAnalysis, StructuredSpeaker,
ValidationError, VocalGenderPresentation,
};
#[cfg(any(feature = "providers", test))]
use kcode_speaker_v3_llm_protocol::SpeakerFeatureEvidence;
#[cfg(any(feature = "providers", test))]
use std::{future::Future, pin::Pin};
const GEMINI_MODEL_ID: &str = "gemini-3.1-pro-preview";
const TERRA_MODEL_ID: &str = "gpt-5.6-terra";
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct GeminiCohort {
pub model_id: String,
pub transcript_prompt_revision: String,
pub feature_prompt_revisions: [String; 3],
pub feature_schema_revision: String,
}
impl GeminiCohort {
pub fn new(model_id: impl Into<String>) -> Self {
Self {
model_id: model_id.into(),
transcript_prompt_revision: GEMINI_TRANSCRIPT_PROMPT_REVISION.into(),
feature_prompt_revisions: GEMINI_FEATURE_PROMPT_REVISIONS.map(str::to_owned),
feature_schema_revision: FEATURE_SCHEMA_REVISION.into(),
}
}
pub fn validate(&self) -> Result<(), ValidationError> {
validate_text(&self.model_id, "gemini_model_id")?;
validate_text(
&self.transcript_prompt_revision,
"transcript_prompt_revision",
)?;
for revision in &self.feature_prompt_revisions {
validate_text(revision, "feature_prompt_revision")?;
}
validate_text(&self.feature_schema_revision, "feature_schema_revision")
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct StructurerProvenance {
pub model_id: String,
pub prompt_revision: String,
}
impl StructurerProvenance {
pub fn new(model_id: impl Into<String>) -> Self {
Self {
model_id: model_id.into(),
prompt_revision: GPT_STRUCTURING_PROMPT_REVISION.into(),
}
}
pub fn validate(&self) -> Result<(), ValidationError> {
validate_text(&self.model_id, "structurer_model_id")?;
validate_text(&self.prompt_revision, "structurer_prompt_revision")
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct AnalysisEnvelope {
pub audio: OggAudioMetadata,
pub analysis: StructuredAnalysis,
pub gemini: GeminiCohort,
pub structurer: StructurerProvenance,
}
impl AnalysisEnvelope {
pub fn validate(&self) -> Result<(), ValidationError> {
self.audio.validate()?;
self.analysis.validate()?;
self.gemini.validate()?;
self.structurer.validate()
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AnalysisError {
Input(String),
GeminiTranscript(String),
TerraLabels(String),
GeminiCache(String),
GeminiFeature {
speaker: LocalSpeakerLabel,
packet: u8,
message: String,
},
TerraStructuring(String),
TranscriptMismatch,
SpeakerSetMismatch,
}
impl fmt::Display for AnalysisError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Input(message) => write!(formatter, "invalid input: {message}"),
Self::GeminiTranscript(message) => {
write!(formatter, "Gemini transcript failed: {message}")
}
Self::TerraLabels(message) => {
write!(
formatter,
"Terra speaker-label extraction failed: {message}"
)
}
Self::GeminiCache(message) => {
write!(formatter, "Gemini feature cache creation failed: {message}")
}
Self::GeminiFeature {
speaker,
packet,
message,
} => write!(
formatter,
"Gemini feature call failed for {speaker}, packet {packet}: {message}"
),
Self::TerraStructuring(message) => {
write!(formatter, "Terra final structuring failed: {message}")
}
Self::TranscriptMismatch => {
formatter.write_str("Terra returned a different transcript")
}
Self::SpeakerSetMismatch => {
formatter.write_str("Terra returned a different speaker set")
}
}
}
}
impl Error for AnalysisError {}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ExecutedAnalysis {
pub envelope: AnalysisEnvelope,
pub label_extractor: StructurerProvenance,
}
#[cfg(feature = "providers")]
pub struct Analyzer {
operations: ProviderOperations,
}
#[cfg(feature = "providers")]
impl Analyzer {
pub fn new(
gemini: kcode_gemini_3_1_pro::Gemini31Pro,
terra: kcode_codex_terra::CodexTerra,
) -> Self {
Self {
operations: ProviderOperations {
gemini: kcode_speaker_v3_gemini_analysis::GeminiAnalysis::new(gemini),
terra: kcode_speaker_v3_terra_analysis::TerraAnalysis::new(terra),
},
}
}
pub async fn analyze_ogg(
&self,
bytes: &[u8],
duration_ms: u64,
filename: Option<String>,
) -> Result<ExecutedAnalysis, AnalysisError> {
execute(&self.operations, bytes, duration_ms, filename).await
}
}
fn validate_text(value: &str, field: &'static str) -> Result<(), ValidationError> {
(!value.trim().is_empty())
.then_some(())
.ok_or(ValidationError::Blank(field))
}
#[cfg(any(feature = "providers", test))]
type AnalysisFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
#[cfg(any(feature = "providers", test))]
trait AnalysisOperations: Sync {
fn transcript<'a>(
&'a self,
audio: &'a [u8],
) -> AnalysisFuture<'a, Result<String, AnalysisError>>;
fn speaker_labels<'a>(
&'a self,
transcript: &'a str,
) -> AnalysisFuture<'a, Result<Vec<LocalSpeakerLabel>, AnalysisError>>;
fn feature_evidence<'a>(
&'a self,
audio: &'a [u8],
transcript: &'a str,
labels: &'a [LocalSpeakerLabel],
) -> AnalysisFuture<'a, Result<Vec<SpeakerFeatureEvidence>, AnalysisError>>;
fn structured_analysis<'a>(
&'a self,
transcript: &'a str,
evidence: Vec<SpeakerFeatureEvidence>,
) -> AnalysisFuture<'a, Result<StructuredAnalysis, AnalysisError>>;
}
#[cfg(any(feature = "providers", test))]
struct ProviderOperations {
gemini: kcode_speaker_v3_gemini_analysis::GeminiAnalysis,
terra: kcode_speaker_v3_terra_analysis::TerraAnalysis,
}
#[cfg(any(feature = "providers", test))]
impl AnalysisOperations for ProviderOperations {
fn transcript<'a>(
&'a self,
audio: &'a [u8],
) -> AnalysisFuture<'a, Result<String, AnalysisError>> {
Box::pin(async move {
self.gemini.transcript(audio).await.map_err(|error| match error {
kcode_speaker_v3_gemini_analysis::GeminiTranscriptError::Provider(message)
| kcode_speaker_v3_gemini_analysis::GeminiTranscriptError::Protocol(message) => {
AnalysisError::GeminiTranscript(message)
}
})
})
}
fn speaker_labels<'a>(
&'a self,
transcript: &'a str,
) -> AnalysisFuture<'a, Result<Vec<LocalSpeakerLabel>, AnalysisError>> {
Box::pin(async move {
self.terra
.speaker_labels(transcript)
.await
.map_err(|error| match error {
kcode_speaker_v3_terra_analysis::TerraAnalysisError::Protocol(message)
| kcode_speaker_v3_terra_analysis::TerraAnalysisError::Provider(message) => {
AnalysisError::TerraLabels(message)
}
})
})
}
fn feature_evidence<'a>(
&'a self,
audio: &'a [u8],
transcript: &'a str,
labels: &'a [LocalSpeakerLabel],
) -> AnalysisFuture<'a, Result<Vec<SpeakerFeatureEvidence>, AnalysisError>> {
Box::pin(async move {
self.gemini
.feature_evidence(audio, transcript, labels)
.await
.map_err(|error| match error {
kcode_speaker_v3_gemini_analysis::GeminiFeatureError::Cache(message) => {
AnalysisError::GeminiCache(message)
}
kcode_speaker_v3_gemini_analysis::GeminiFeatureError::Feature {
speaker,
packet,
message,
} => AnalysisError::GeminiFeature {
speaker,
packet,
message,
},
})
})
}
fn structured_analysis<'a>(
&'a self,
transcript: &'a str,
evidence: Vec<SpeakerFeatureEvidence>,
) -> AnalysisFuture<'a, Result<StructuredAnalysis, AnalysisError>> {
Box::pin(async move {
self.terra
.structured_analysis(transcript, evidence)
.await
.map_err(|error| match error {
kcode_speaker_v3_terra_analysis::TerraAnalysisError::Protocol(message)
| kcode_speaker_v3_terra_analysis::TerraAnalysisError::Provider(message) => {
AnalysisError::TerraStructuring(message)
}
})
})
}
}
#[cfg(any(feature = "providers", test))]
async fn execute<O: AnalysisOperations>(
operations: &O,
bytes: &[u8],
duration_ms: u64,
filename: Option<String>,
) -> Result<ExecutedAnalysis, AnalysisError> {
let audio = OggAudioMetadata::from_bytes(bytes, duration_ms, filename)
.map_err(|error| AnalysisError::Input(error.to_string()))?;
let transcript = operations.transcript(bytes).await?;
let labels = operations.speaker_labels(&transcript).await?;
let evidence = operations
.feature_evidence(bytes, &transcript, &labels)
.await?;
let analysis = operations
.structured_analysis(&transcript, evidence)
.await?;
if analysis.transcript != transcript {
return Err(AnalysisError::TranscriptMismatch);
}
let expected_speakers = labels.iter().copied().collect::<BTreeSet<_>>();
let returned_speakers = analysis
.speakers
.iter()
.map(|speaker| speaker.speaker)
.collect::<BTreeSet<_>>();
if expected_speakers != returned_speakers {
return Err(AnalysisError::SpeakerSetMismatch);
}
let envelope = AnalysisEnvelope {
audio,
analysis,
gemini: GeminiCohort::new(GEMINI_MODEL_ID),
structurer: StructurerProvenance::new(TERRA_MODEL_ID),
};
envelope
.validate()
.map_err(|error| AnalysisError::TerraStructuring(error.to_string()))?;
let label_extractor = StructurerProvenance {
model_id: TERRA_MODEL_ID.into(),
prompt_revision: kcode_speaker_v3_llm_protocol::TERRA_SPEAKER_LABELS_PROMPT_REVISION.into(),
};
label_extractor
.validate()
.map_err(|error| AnalysisError::TerraLabels(error.to_string()))?;
Ok(ExecutedAnalysis {
envelope,
label_extractor,
})
}
#[cfg(test)]
mod tests {
use super::*;
use futures::{executor::block_on, future::poll_fn, join};
use std::{
sync::{
Arc, Mutex,
atomic::{AtomicBool, AtomicUsize, Ordering},
},
task::Poll,
time::Instant,
};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum FailureStage {
Transcript,
Labels,
Features,
Final,
}
#[derive(Clone)]
struct FakeConfig {
transcript: String,
labels: Vec<LocalSpeakerLabel>,
analysis: StructuredAnalysis,
failure: Option<FailureStage>,
wait_for: Option<Arc<AtomicBool>>,
mark_complete: Option<Arc<AtomicBool>>,
}
struct FakeState {
config: FakeConfig,
transcript_calls: AtomicUsize,
label_calls: AtomicUsize,
feature_calls: AtomicUsize,
final_calls: AtomicUsize,
calls: Mutex<Vec<&'static str>>,
}
#[derive(Clone)]
struct FakeOperations {
state: Arc<FakeState>,
}
impl FakeOperations {
fn successful(speaker_count: u32) -> Self {
let transcript = "[high] Speaker 1: exact transcript".to_owned();
Self::from_config(FakeConfig {
labels: (1..=speaker_count).map(label).collect(),
analysis: structured_analysis(&transcript, speaker_count),
transcript,
failure: None,
wait_for: None,
mark_complete: None,
})
}
fn from_config(config: FakeConfig) -> Self {
Self {
state: Arc::new(FakeState {
config,
transcript_calls: AtomicUsize::new(0),
label_calls: AtomicUsize::new(0),
feature_calls: AtomicUsize::new(0),
final_calls: AtomicUsize::new(0),
calls: Mutex::new(Vec::new()),
}),
}
}
fn with_config(&self, update: impl FnOnce(&mut FakeConfig)) -> Self {
let mut config = self.state.config.clone();
update(&mut config);
Self::from_config(config)
}
}
impl AnalysisOperations for FakeOperations {
fn transcript<'a>(
&'a self,
_audio: &'a [u8],
) -> AnalysisFuture<'a, Result<String, AnalysisError>> {
Box::pin(async move {
self.state.transcript_calls.fetch_add(1, Ordering::SeqCst);
self.state.calls.lock().unwrap().push("transcript");
if let Some(wait_for) = &self.state.config.wait_for {
poll_fn(|context| {
if wait_for.load(Ordering::SeqCst) {
Poll::Ready(())
} else {
context.waker().wake_by_ref();
Poll::Pending
}
})
.await;
}
if self.state.config.failure == Some(FailureStage::Transcript) {
return Err(AnalysisError::GeminiTranscript("transcript".into()));
}
Ok(self.state.config.transcript.clone())
})
}
fn speaker_labels<'a>(
&'a self,
_transcript: &'a str,
) -> AnalysisFuture<'a, Result<Vec<LocalSpeakerLabel>, AnalysisError>> {
Box::pin(async move {
self.state.label_calls.fetch_add(1, Ordering::SeqCst);
self.state.calls.lock().unwrap().push("labels");
if self.state.config.failure == Some(FailureStage::Labels) {
return Err(AnalysisError::TerraLabels("labels".into()));
}
Ok(self.state.config.labels.clone())
})
}
fn feature_evidence<'a>(
&'a self,
_audio: &'a [u8],
_transcript: &'a str,
labels: &'a [LocalSpeakerLabel],
) -> AnalysisFuture<'a, Result<Vec<SpeakerFeatureEvidence>, AnalysisError>> {
Box::pin(async move {
self.state.feature_calls.fetch_add(1, Ordering::SeqCst);
self.state.calls.lock().unwrap().push("features");
if self.state.config.failure == Some(FailureStage::Features) {
return Err(AnalysisError::GeminiFeature {
speaker: label(1),
packet: 2,
message: "features".into(),
});
}
labels
.iter()
.copied()
.map(|speaker| {
SpeakerFeatureEvidence::new(
speaker,
format!("{speaker} packet 1"),
format!("{speaker} packet 2"),
format!("{speaker} packet 3"),
)
.map_err(|error| AnalysisError::GeminiFeature {
speaker,
packet: 1,
message: error.to_string(),
})
})
.collect()
})
}
fn structured_analysis<'a>(
&'a self,
_transcript: &'a str,
_evidence: Vec<SpeakerFeatureEvidence>,
) -> AnalysisFuture<'a, Result<StructuredAnalysis, AnalysisError>> {
Box::pin(async move {
self.state.final_calls.fetch_add(1, Ordering::SeqCst);
self.state.calls.lock().unwrap().push("final");
if self.state.config.failure == Some(FailureStage::Final) {
return Err(AnalysisError::TerraStructuring("final".into()));
}
if let Some(mark_complete) = &self.state.config.mark_complete {
mark_complete.store(true, Ordering::SeqCst);
}
Ok(self.state.config.analysis.clone())
})
}
}
fn label(number: u32) -> LocalSpeakerLabel {
LocalSpeakerLabel::new(number).unwrap()
}
fn structured_analysis(transcript: &str, speaker_count: u32) -> StructuredAnalysis {
StructuredAnalysis {
transcript: transcript.into(),
speakers: (1..=speaker_count)
.map(|number| StructuredSpeaker {
speaker: label(number),
language: "English".into(),
features: FeatureVector24::default(),
features_usable_for_training: false,
})
.collect(),
}
}
fn ogg() -> Vec<u8> {
let mut bytes = vec![0; 28];
bytes[..4].copy_from_slice(b"OggS");
bytes[4] = 0;
bytes[26] = 1;
bytes[27] = 0;
bytes
}
#[test]
fn zero_one_and_many_speaker_workflows_keep_exact_stage_order() {
for speaker_count in [0, 1, 40] {
let operations = FakeOperations::successful(speaker_count);
let result =
block_on(execute(&operations, &ogg(), 1, Some("voice.ogg".into()))).unwrap();
assert_eq!(
result.envelope.analysis.speakers.len(),
speaker_count as usize
);
assert_eq!(
*operations.state.calls.lock().unwrap(),
["transcript", "labels", "features", "final"]
);
assert_eq!(
result.label_extractor,
StructurerProvenance {
model_id: TERRA_MODEL_ID.into(),
prompt_revision:
kcode_speaker_v3_llm_protocol::TERRA_SPEAKER_LABELS_PROMPT_REVISION.into(),
}
);
}
}
#[test]
fn input_and_each_provider_stage_fail_without_retry() {
let input = FakeOperations::successful(1);
assert!(matches!(
block_on(execute(&input, b"bad", 1, None)),
Err(AnalysisError::Input(_))
));
assert!(input.state.calls.lock().unwrap().is_empty());
for (stage, expected) in [
(FailureStage::Transcript, vec!["transcript"]),
(FailureStage::Labels, vec!["transcript", "labels"]),
(
FailureStage::Features,
vec!["transcript", "labels", "features"],
),
(
FailureStage::Final,
vec!["transcript", "labels", "features", "final"],
),
] {
let operations =
FakeOperations::successful(1).with_config(|config| config.failure = Some(stage));
assert!(block_on(execute(&operations, &ogg(), 1, None)).is_err());
assert_eq!(*operations.state.calls.lock().unwrap(), expected);
}
}
#[test]
fn cross_stage_transcript_and_speaker_mismatches_are_rejected() {
let transcript = FakeOperations::successful(1).with_config(|config| {
config.analysis = structured_analysis("different", 1);
});
assert_eq!(
block_on(execute(&transcript, &ogg(), 1, None)),
Err(AnalysisError::TranscriptMismatch)
);
let speakers = FakeOperations::successful(1).with_config(|config| {
config.analysis = structured_analysis(&config.transcript, 2);
});
assert_eq!(
block_on(execute(&speakers, &ogg(), 1, None)),
Err(AnalysisError::SpeakerSetMismatch)
);
}
#[test]
fn a_blocked_analysis_does_not_block_an_unrelated_analysis() {
let completed = Arc::new(AtomicBool::new(false));
let fast = FakeOperations::successful(0).with_config(|config| {
config.mark_complete = Some(completed.clone());
});
let slow = FakeOperations::successful(0).with_config(|config| {
config.wait_for = Some(completed.clone());
});
let slow_audio = ogg();
let fast_audio = ogg();
let (slow_result, fast_result) = block_on(async {
join!(
execute(&slow, &slow_audio, 1, None),
execute(&fast, &fast_audio, 1, None)
)
});
slow_result.unwrap();
fast_result.unwrap();
assert!(completed.load(Ordering::SeqCst));
}
#[test]
fn provenance_preserves_the_previous_public_contract() {
let cohort = GeminiCohort::new("gemini-model");
assert_eq!(
cohort.feature_prompt_revisions,
GEMINI_FEATURE_PROMPT_REVISIONS.map(str::to_owned)
);
cohort.validate().unwrap();
StructurerProvenance::new("gpt-5.6").validate().unwrap();
assert_eq!(
GeminiCohort::new(" ").validate(),
Err(ValidationError::Blank("gemini_model_id"))
);
}
#[test]
fn reference_scale_local_orchestration_completes_within_envelope() {
let started = Instant::now();
let operations = FakeOperations::successful(1000);
let result = block_on(execute(&operations, &ogg(), 1, None)).unwrap();
assert_eq!(result.envelope.analysis.speakers.len(), 1000);
assert!(started.elapsed().as_secs() < 10);
}
#[test]
fn concrete_provider_operations_compile() {
fn require_operations<O: AnalysisOperations>() {}
require_operations::<ProviderOperations>();
}
}