1#![forbid(unsafe_code)]
2
3pub use kcode_speaker_dataset::{
4 Dataset, DatasetError, FoldConfig, OpenSetFold, RepeatabilityGroup,
5};
6pub use kcode_speaker_eval::{
7 AggregateMetrics, CandidateEvaluation, EvalError, EvaluationPlan, EvaluationResult,
8};
9pub use kcode_speaker_extract::{ExtractError, ExtractionContract};
10pub use kcode_speaker_model::{
11 ALL_24, CandidateScore, Decision, Identification, ModelConfig, ModelError, ModelSnapshot,
12};
13pub use kcode_speaker_store::{AttemptSnapshot, SampleState, StoreError};
14pub use kcode_speaker_types::{
15 FEATURE_COUNT, FeatureMask, FeatureVector, Key, LabeledSample, ObjectId, RecordingKind,
16 SegmentRef,
17};
18
19use kcode_speaker_store::{
20 AttemptRecord, AttemptSelection, Event, EventEnvelope, InventorySnapshot, Query, Request,
21 Response, ResponseKind, SampleStateChange, SegmentRegistration, Store, StoredAdditionalSpeaker,
22 StoredAttemptOutcome, StoredSpeaker,
23};
24use std::{collections::BTreeSet, error::Error, fmt, path::Path};
25
26const _: [(); 24] = [(); FEATURE_COUNT];
27
28#[derive(Clone, Debug, Eq, PartialEq)]
29pub struct SegmentBinding {
30 pub event_id: Key,
31 pub clip_object: ObjectId,
32}
33
34#[derive(Clone, Debug, Eq, PartialEq)]
35pub struct SourceRegistration {
36 pub source_object: ObjectId,
37 pub source_duration_ms: u64,
38 pub group_id: Key,
39 pub recording_kind: RecordingKind,
40 pub segments: Vec<SegmentBinding>,
41}
42
43#[derive(Clone, Debug, Eq, PartialEq)]
44pub struct NormalizedAttempt<'a> {
45 pub event_id: Key,
46 pub attempt_id: Key,
47 pub cohort_id: Key,
48 pub source_object: ObjectId,
49 pub clip_object: ObjectId,
50 pub segment_ordinal: u16,
51 pub provider_result_object: ObjectId,
52 pub recording_quality: Option<u8>,
53 pub normalized_response: &'a str,
54}
55
56#[derive(Clone, Debug, Eq, PartialEq)]
57pub struct AttemptSelectionRequest {
58 pub event_id: Key,
59 pub attempt_id: Key,
60 pub cohort_id: Key,
61 pub clip_object: ObjectId,
62 pub reason: String,
63}
64
65#[derive(Clone, Debug, Eq, PartialEq)]
66pub struct SampleStateRequest {
67 pub event_id: Key,
68 pub sample_id: Key,
69 pub state: SampleState,
70 pub reason: String,
71}
72
73#[derive(Clone, Copy, Debug, Eq, PartialEq)]
74pub struct CommitReceipt {
75 pub revision: u64,
76 pub applied: u64,
77}
78
79#[derive(Debug)]
80pub enum SystemError {
81 Boundary(&'static str),
82 ProjectionRevisionChanged {
83 active_revision: u64,
84 repeatability_revision: u64,
85 },
86 UnexpectedStoreResponse(&'static str),
87 Extract(ExtractError),
88 Store(StoreError),
89 Dataset(DatasetError),
90 Model(ModelError),
91 Eval(EvalError),
92}
93
94impl fmt::Display for SystemError {
95 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
96 match self {
97 Self::Boundary(message) => write!(formatter, "boundary: {message}"),
98 Self::ProjectionRevisionChanged {
99 active_revision,
100 repeatability_revision,
101 } => write!(
102 formatter,
103 "store projection revision changed between reads ({active_revision} then {repeatability_revision})"
104 ),
105 Self::UnexpectedStoreResponse(message) => {
106 write!(formatter, "unexpected store response: {message}")
107 }
108 Self::Extract(error) => error.fmt(formatter),
109 Self::Store(error) => error.fmt(formatter),
110 Self::Dataset(error) => error.fmt(formatter),
111 Self::Model(error) => error.fmt(formatter),
112 Self::Eval(error) => error.fmt(formatter),
113 }
114 }
115}
116
117impl Error for SystemError {
118 fn source(&self) -> Option<&(dyn Error + 'static)> {
119 match self {
120 Self::Extract(error) => Some(error),
121 Self::Store(error) => Some(error),
122 Self::Dataset(error) => Some(error),
123 Self::Model(error) => Some(error),
124 Self::Eval(error) => Some(error),
125 Self::Boundary(_)
126 | Self::ProjectionRevisionChanged { .. }
127 | Self::UnexpectedStoreResponse(_) => None,
128 }
129 }
130}
131
132impl From<ExtractError> for SystemError {
133 fn from(error: ExtractError) -> Self {
134 Self::Extract(error)
135 }
136}
137
138impl From<StoreError> for SystemError {
139 fn from(error: StoreError) -> Self {
140 Self::Store(error)
141 }
142}
143
144impl From<DatasetError> for SystemError {
145 fn from(error: DatasetError) -> Self {
146 Self::Dataset(error)
147 }
148}
149
150impl From<ModelError> for SystemError {
151 fn from(error: ModelError) -> Self {
152 Self::Model(error)
153 }
154}
155
156impl From<EvalError> for SystemError {
157 fn from(error: EvalError) -> Self {
158 Self::Eval(error)
159 }
160}
161
162pub struct SpeakerSystem {
163 store: Store,
164}
165
166pub fn cohort_id() -> &'static Key {
167 &extraction_contract().normalized_schema_key
168}
169
170pub fn extraction_contract() -> &'static ExtractionContract {
171 kcode_speaker_extract::contract()
172}
173
174pub fn normalization_prompt(raw_analysis: &str) -> Result<String, SystemError> {
175 kcode_speaker_extract::normalization_prompt(raw_analysis).map_err(SystemError::from)
176}
177
178pub fn open(path: impl AsRef<Path>) -> Result<SpeakerSystem, SystemError> {
179 Ok(SpeakerSystem {
180 store: kcode_speaker_store::open(path)?,
181 })
182}
183
184impl SpeakerSystem {
185 pub fn register_source(
186 &self,
187 request: SourceRegistration,
188 ) -> Result<CommitReceipt, SystemError> {
189 let SourceRegistration {
190 source_object,
191 source_duration_ms,
192 group_id,
193 recording_kind,
194 segments,
195 } = request;
196 let plan = kcode_speaker_extract::plan_segments(source_duration_ms)?;
197
198 if segments.len() != plan.segments.len() {
199 return Err(SystemError::Boundary(
200 "clip bindings must exactly match the planned segment count",
201 ));
202 }
203
204 let mut event_ids = BTreeSet::new();
205 let mut clip_objects = BTreeSet::new();
206 for binding in &segments {
207 if !event_ids.insert(binding.event_id.clone()) {
208 return Err(SystemError::Boundary(
209 "source registration event IDs must be unique",
210 ));
211 }
212 if !clip_objects.insert(binding.clip_object.clone()) {
213 return Err(SystemError::Boundary(
214 "source registration clip object IDs must be unique",
215 ));
216 }
217 }
218
219 let segment_count = u16::try_from(plan.segments.len()).map_err(|_| {
220 SystemError::Boundary("planned segment count exceeds the shared u16 boundary")
221 })?;
222 let events = segments
223 .into_iter()
224 .zip(plan.segments)
225 .map(|(binding, planned)| EventEnvelope {
226 event_id: binding.event_id,
227 event: Event::RegisterSegment(SegmentRegistration {
228 segment: SegmentRef {
229 source_object: source_object.clone(),
230 clip_object: binding.clip_object,
231 ordinal: planned.ordinal,
232 segment_count,
233 start_ms: planned.start_ms,
234 end_ms: planned.end_ms,
235 policy: plan.policy.clone(),
236 },
237 group_id: group_id.clone(),
238 recording_kind,
239 }),
240 })
241 .collect();
242
243 self.commit(events)
244 }
245
246 pub fn record_normalized_attempt(
247 &self,
248 request: NormalizedAttempt<'_>,
249 ) -> Result<CommitReceipt, SystemError> {
250 let NormalizedAttempt {
251 event_id,
252 attempt_id,
253 cohort_id: requested_cohort,
254 source_object,
255 clip_object,
256 segment_ordinal,
257 provider_result_object,
258 recording_quality,
259 normalized_response,
260 } = request;
261 require_frozen_cohort(&requested_cohort)?;
262
263 let registration =
264 self.registered_segment(&source_object, &clip_object, segment_ordinal)?;
265 let clip_duration_ms = registration.segment.end_ms - registration.segment.start_ms;
266 let parsed = kcode_speaker_extract::parse(normalized_response, clip_duration_ms)?;
267
268 let outcome = match parsed {
269 kcode_speaker_extract::ExtractionOutcome::Scored(scored) => {
270 let recording_quality = recording_quality.ok_or(SystemError::Boundary(
271 "scored outcomes require caller-supplied recording quality",
272 ))?;
273 if recording_quality > 100 {
274 return Err(SystemError::Boundary(
275 "recording quality must be at most 100",
276 ));
277 }
278
279 StoredAttemptOutcome::Scored {
280 recording_quality,
281 speakers: scored
282 .speakers
283 .into_iter()
284 .map(|speaker| StoredSpeaker {
285 speaker_ordinal: speaker.speaker_ordinal,
286 primary_language: speaker.primary_language,
287 closest_dialect: speaker.closest_dialect,
288 usable_speech_ms: speaker.usable_speech_ms,
289 features: speaker.features,
290 })
291 .collect(),
292 additional_speakers: scored
293 .additional_speakers
294 .into_iter()
295 .map(|speaker| StoredAdditionalSpeaker {
296 speaker_ordinal: speaker.speaker_ordinal,
297 description: speaker.description,
298 })
299 .collect(),
300 }
301 }
302 kcode_speaker_extract::ExtractionOutcome::Unscorable {
303 reason,
304 additional_speakers,
305 } => {
306 if recording_quality.is_some() {
307 return Err(SystemError::Boundary(
308 "unscorable outcomes must not discard recording quality",
309 ));
310 }
311
312 StoredAttemptOutcome::Unscorable {
313 reason,
314 additional_speakers: additional_speakers
315 .into_iter()
316 .map(|speaker| StoredAdditionalSpeaker {
317 speaker_ordinal: speaker.speaker_ordinal,
318 description: speaker.description,
319 })
320 .collect(),
321 }
322 }
323 };
324
325 self.commit(vec![EventEnvelope {
326 event_id,
327 event: Event::RecordAttempt(AttemptRecord {
328 attempt_id,
329 clip_object,
330 extraction_key: requested_cohort,
331 provider_result_object: Some(provider_result_object),
332 outcome,
333 }),
334 }])
335 }
336
337 pub fn select_attempt(
338 &self,
339 request: AttemptSelectionRequest,
340 ) -> Result<CommitReceipt, SystemError> {
341 let AttemptSelectionRequest {
342 event_id,
343 attempt_id,
344 cohort_id: requested_cohort,
345 clip_object,
346 reason,
347 } = request;
348 require_frozen_cohort(&requested_cohort)?;
349
350 let snapshot = self
351 .attempt(&attempt_id)?
352 .ok_or(SystemError::Boundary("selected attempt does not exist"))?;
353 if snapshot.attempt.clip_object != clip_object
354 || snapshot.attempt.extraction_key != requested_cohort
355 || !matches!(
356 &snapshot.attempt.outcome,
357 StoredAttemptOutcome::Scored { .. }
358 )
359 {
360 return Err(SystemError::Boundary(
361 "selection must match a scored attempt, clip, and cohort",
362 ));
363 }
364
365 self.commit(vec![EventEnvelope {
366 event_id,
367 event: Event::SelectAttempt(AttemptSelection {
368 clip_object,
369 extraction_key: requested_cohort,
370 attempt_id,
371 reason,
372 }),
373 }])
374 }
375
376 pub fn change_sample_state(
377 &self,
378 request: SampleStateRequest,
379 ) -> Result<CommitReceipt, SystemError> {
380 let SampleStateRequest {
381 event_id,
382 sample_id,
383 state,
384 reason,
385 } = request;
386
387 let response = self.store.execute(Request::Read(Query::Attempts {
388 extraction_key: Some(cohort_id().clone()),
389 source_object: None,
390 }))?;
391 let ResponseKind::Attempts(attempts) = response.result else {
392 return Err(SystemError::UnexpectedStoreResponse(
393 "attempt-list query returned another result kind",
394 ));
395 };
396 if !attempts
397 .iter()
398 .any(|attempt| attempt.sample_ids.contains(&sample_id))
399 {
400 return Err(SystemError::Boundary(
401 "sample does not belong to a complete speaker in the frozen cohort",
402 ));
403 }
404
405 self.commit(vec![EventEnvelope {
406 event_id,
407 event: Event::SetSampleState(SampleStateChange {
408 sample_id,
409 state,
410 reason,
411 }),
412 }])
413 }
414
415 pub fn attempt(&self, attempt_id: &Key) -> Result<Option<AttemptSnapshot>, SystemError> {
416 let response = self.store.execute(Request::Read(Query::Attempt {
417 attempt_id: attempt_id.clone(),
418 }))?;
419 match response.result {
420 ResponseKind::Attempt(snapshot) => Ok(snapshot),
421 _ => Err(SystemError::UnexpectedStoreResponse(
422 "attempt query returned another result kind",
423 )),
424 }
425 }
426
427 pub fn dataset(&self, requested_cohort: &Key) -> Result<Dataset, SystemError> {
428 require_frozen_cohort(requested_cohort)?;
429
430 let active_response = self.store.execute(Request::Read(Query::TrainingRows {
431 cohort_id: requested_cohort.clone(),
432 }))?;
433 let active_revision = active_response.revision;
434 let ResponseKind::TrainingRows(active_rows) = active_response.result else {
435 return Err(SystemError::UnexpectedStoreResponse(
436 "training-row query returned another result kind",
437 ));
438 };
439
440 let repeatability_response =
441 self.store.execute(Request::Read(Query::RepeatabilityRows {
442 cohort_id: requested_cohort.clone(),
443 }))?;
444 let repeatability_revision = repeatability_response.revision;
445 if active_revision != repeatability_revision {
446 return Err(SystemError::ProjectionRevisionChanged {
447 active_revision,
448 repeatability_revision,
449 });
450 }
451 let ResponseKind::RepeatabilityRows(repeatability_rows) = repeatability_response.result
452 else {
453 return Err(SystemError::UnexpectedStoreResponse(
454 "repeatability-row query returned another result kind",
455 ));
456 };
457
458 kcode_speaker_dataset::build(active_rows, repeatability_rows).map_err(SystemError::from)
459 }
460
461 fn registered_segment(
462 &self,
463 source_object: &ObjectId,
464 clip_object: &ObjectId,
465 segment_ordinal: u16,
466 ) -> Result<SegmentRegistration, SystemError> {
467 let response = self.store.execute(Request::Read(Query::Inventory {
468 extraction_key: None,
469 }))?;
470 let ResponseKind::Inventory(inventory) = response.result else {
471 return Err(SystemError::UnexpectedStoreResponse(
472 "inventory query returned another result kind",
473 ));
474 };
475 validate_registered_segment(&inventory, source_object, clip_object, segment_ordinal)
476 }
477
478 fn commit(&self, events: Vec<EventEnvelope>) -> Result<CommitReceipt, SystemError> {
479 let Response { revision, result } = self.store.execute(Request::Commit(events))?;
480 match result {
481 ResponseKind::Commit { applied } => Ok(CommitReceipt { revision, applied }),
482 _ => Err(SystemError::UnexpectedStoreResponse(
483 "commit returned another result kind",
484 )),
485 }
486 }
487}
488
489pub fn open_set_folds(
490 dataset: &Dataset,
491 config: FoldConfig,
492) -> Result<Vec<OpenSetFold>, SystemError> {
493 kcode_speaker_dataset::open_set_folds(dataset, config).map_err(SystemError::from)
494}
495
496pub fn repeatability_groups(dataset: &Dataset) -> Vec<RepeatabilityGroup> {
497 kcode_speaker_dataset::repeatability_groups(dataset)
498}
499
500pub fn fit(
501 dataset: &Dataset,
502 requested_cohort: &Key,
503 config: ModelConfig,
504) -> Result<ModelSnapshot, SystemError> {
505 require_frozen_cohort(requested_cohort)?;
506 let samples = kcode_speaker_dataset::active_rows(dataset);
507 if samples
508 .iter()
509 .any(|sample| sample.cohort_id != *requested_cohort)
510 {
511 return Err(SystemError::Boundary(
512 "dataset rows must all match the requested frozen cohort",
513 ));
514 }
515
516 kcode_speaker_model::fit(kcode_speaker_model::FitInput {
517 cohort_id: requested_cohort,
518 samples,
519 config,
520 })
521 .map_err(SystemError::from)
522}
523
524pub fn identify(
525 model: &ModelSnapshot,
526 requested_cohort: &Key,
527 features: &FeatureVector,
528) -> Result<Identification, SystemError> {
529 require_frozen_cohort(requested_cohort)?;
530 kcode_speaker_model::identify(model, requested_cohort, features).map_err(SystemError::from)
531}
532
533pub fn snapshot_bytes(model: &ModelSnapshot) -> Result<Vec<u8>, SystemError> {
534 kcode_speaker_model::encode(model).map_err(SystemError::from)
535}
536
537pub fn snapshot_from_bytes(bytes: &[u8]) -> Result<ModelSnapshot, SystemError> {
538 kcode_speaker_model::decode(bytes).map_err(SystemError::from)
539}
540
541pub fn evaluate(plan: EvaluationPlan<'_>) -> Result<EvaluationResult, SystemError> {
542 let rows = kcode_speaker_dataset::active_rows(plan.dataset);
543 let dataset_cohort = rows
544 .first()
545 .ok_or(SystemError::Boundary(
546 "evaluation dataset has no active rows",
547 ))?
548 .cohort_id
549 .clone();
550 require_frozen_cohort(&dataset_cohort)?;
551 if rows.iter().any(|row| row.cohort_id != dataset_cohort) {
552 return Err(SystemError::Boundary(
553 "evaluation dataset rows must share the frozen cohort",
554 ));
555 }
556
557 kcode_speaker_eval::evaluate(plan).map_err(SystemError::from)
558}
559
560fn require_frozen_cohort(requested: &Key) -> Result<(), SystemError> {
561 if requested != cohort_id() {
562 return Err(SystemError::Boundary(
563 "cohort ID must equal the frozen normalized extraction schema key",
564 ));
565 }
566 Ok(())
567}
568
569fn validate_registered_segment(
570 inventory: &InventorySnapshot,
571 source_object: &ObjectId,
572 clip_object: &ObjectId,
573 segment_ordinal: u16,
574) -> Result<SegmentRegistration, SystemError> {
575 let target = inventory
576 .registrations
577 .iter()
578 .find(|registration| registration.segment.clip_object == *clip_object)
579 .cloned()
580 .ok_or(SystemError::Boundary("clip object is not registered"))?;
581
582 if target.segment.source_object != *source_object || target.segment.ordinal != segment_ordinal {
583 return Err(SystemError::Boundary(
584 "source object, clip object, and segment ordinal do not match",
585 ));
586 }
587
588 let source_registrations = inventory
589 .registrations
590 .iter()
591 .filter(|registration| registration.segment.source_object == *source_object)
592 .collect::<Vec<_>>();
593 if source_registrations.len() != usize::from(target.segment.segment_count) {
594 return Err(SystemError::Boundary(
595 "registered source does not contain its complete segment plan",
596 ));
597 }
598
599 let source_duration_ms = source_registrations
600 .iter()
601 .map(|registration| registration.segment.end_ms)
602 .max()
603 .ok_or(SystemError::Boundary(
604 "registered source has no segment ranges",
605 ))?;
606 let plan = kcode_speaker_extract::plan_segments(source_duration_ms)?;
607 if plan.segments.len() != source_registrations.len() {
608 return Err(SystemError::Boundary(
609 "registered source segment count differs from deterministic planning",
610 ));
611 }
612
613 for planned in &plan.segments {
614 let registration = source_registrations
615 .iter()
616 .find(|registration| registration.segment.ordinal == planned.ordinal)
617 .ok_or(SystemError::Boundary(
618 "registered source is missing a planned segment ordinal",
619 ))?;
620 if registration.segment.segment_count != target.segment.segment_count
621 || registration.segment.policy != plan.policy
622 || registration.segment.start_ms != planned.start_ms
623 || registration.segment.end_ms != planned.end_ms
624 {
625 return Err(SystemError::Boundary(
626 "registered source ranges differ from deterministic planning",
627 ));
628 }
629 }
630
631 Ok(target)
632}
633
634#[cfg(test)]
635mod tests {
636 use super::*;
637 use kcode_speaker_store::StoredAttemptOutcome;
638 use std::{
639 fs,
640 path::PathBuf,
641 sync::atomic::{AtomicU64, Ordering},
642 };
643
644 static NEXT_PATH: AtomicU64 = AtomicU64::new(0);
645
646 fn key(value: &str) -> Key {
647 Key::parse(value).unwrap()
648 }
649
650 fn object(value: &str) -> ObjectId {
651 ObjectId::parse(value).unwrap()
652 }
653
654 fn path() -> PathBuf {
655 std::env::temp_dir().join(format!(
656 "kcode-speaker-system-{}-{}.db",
657 std::process::id(),
658 NEXT_PATH.fetch_add(1, Ordering::Relaxed)
659 ))
660 }
661
662 fn feature_values(seed: u8) -> String {
663 (0..FEATURE_COUNT)
664 .map(|index| (usize::from(seed) + index).to_string())
665 .collect::<Vec<_>>()
666 .join(",")
667 }
668
669 fn scored_response(seed: u8, has_additional_speaker: bool) -> String {
670 let features = feature_values(seed);
671 let additional = if has_additional_speaker {
672 r#"[{"speakerOrdinal":1,"description":"Brief background interjection."}]"#
673 } else {
674 "[]"
675 };
676 format!(
677 r#"{{"status":"scored","speakers":[{{"speakerOrdinal":0,"primaryLanguage":"en-US","closestDialect":"General American English","usableSpeechMs":20000,"features":[{features}]}}],"additionalSpeakers":{additional}}}"#
678 )
679 }
680
681 fn register(system: &SpeakerSystem, source: &str, clip: &str, event_id: &str, group: &str) {
682 system
683 .register_source(SourceRegistration {
684 source_object: object(source),
685 source_duration_ms: 100_000,
686 group_id: key(group),
687 recording_kind: RecordingKind::VoiceNote,
688 segments: vec![SegmentBinding {
689 event_id: key(event_id),
690 clip_object: object(clip),
691 }],
692 })
693 .unwrap();
694 }
695
696 fn record(
697 system: &SpeakerSystem,
698 source: &str,
699 clip: &str,
700 event_id: &str,
701 attempt_id: &str,
702 result_object: &str,
703 response: &str,
704 ) {
705 system
706 .record_normalized_attempt(NormalizedAttempt {
707 event_id: key(event_id),
708 attempt_id: key(attempt_id),
709 cohort_id: cohort_id().clone(),
710 source_object: object(source),
711 clip_object: object(clip),
712 segment_ordinal: 0,
713 provider_result_object: object(result_object),
714 recording_quality: Some(87),
715 normalized_response: response,
716 })
717 .unwrap();
718 }
719
720 fn select_and_confirm(
721 system: &SpeakerSystem,
722 attempt_id: &str,
723 clip: &str,
724 select_event: &str,
725 state_event: &str,
726 ) -> Key {
727 let snapshot = system.attempt(&key(attempt_id)).unwrap().unwrap();
728 assert_eq!(snapshot.sample_ids.len(), 1);
729 let sample_id = snapshot.sample_ids[0].clone();
730
731 system
732 .select_attempt(AttemptSelectionRequest {
733 event_id: key(select_event),
734 attempt_id: key(attempt_id),
735 cohort_id: cohort_id().clone(),
736 clip_object: object(clip),
737 reason: "reviewed complete profile".into(),
738 })
739 .unwrap();
740 system
741 .change_sample_state(SampleStateRequest {
742 event_id: key(state_event),
743 sample_id: sample_id.clone(),
744 state: SampleState::Confirmed {
745 speaker_id: key("speaker/alice"),
746 },
747 reason: "confirmed caller label".into(),
748 })
749 .unwrap();
750
751 sample_id
752 }
753
754 #[test]
755 fn exact_extraction_store_dataset_and_model_contracts_compose() {
756 let direct_contract = kcode_speaker_extract::contract();
757 assert!(std::ptr::eq(extraction_contract(), direct_contract));
758 assert_eq!(
759 extraction_contract().schema_key.as_ref(),
760 "gemini-speaker-24-freeform/1"
761 );
762 assert_eq!(
763 extraction_contract().normalized_schema_key.as_ref(),
764 "gemini-speaker-24-normalized/1"
765 );
766 assert_eq!(extraction_contract().response_mime, "text/plain");
767 assert_eq!(extraction_contract().normalized_mime, "application/json");
768 assert!(
769 extraction_contract()
770 .prompt
771 .starts_with("Analyze the attached audio directly and separate")
772 );
773 assert!(
774 extraction_contract()
775 .prompt
776 .contains("24. sibilant_sharpness")
777 );
778 assert!(!extraction_contract().prompt.contains("recording_quality"));
779
780 let raw_analysis = "Speaker 1: complete profile. Speaker 2: too brief.";
781 assert_eq!(
782 normalization_prompt(raw_analysis).unwrap(),
783 kcode_speaker_extract::normalization_prompt(raw_analysis).unwrap()
784 );
785
786 let database = path();
787 let system = open(&database).unwrap();
788 register(
789 &system,
790 "SOURCE01",
791 "CLIP0001",
792 "event/register/1",
793 "group/source/1",
794 );
795 register(
796 &system,
797 "SOURCE02",
798 "CLIP0002",
799 "event/register/2",
800 "group/source/2",
801 );
802
803 let first_response = scored_response(10, true);
804 let second_response = scored_response(60, false);
805 record(
806 &system,
807 "SOURCE01",
808 "CLIP0001",
809 "event/attempt/1",
810 "attempt/1",
811 "RESULT01",
812 &first_response,
813 );
814 record(
815 &system,
816 "SOURCE02",
817 "CLIP0002",
818 "event/attempt/2",
819 "attempt/2",
820 "RESULT02",
821 &second_response,
822 );
823
824 let first = system.attempt(&key("attempt/1")).unwrap().unwrap();
825 assert_eq!(first.sample_ids.len(), 1);
826 let StoredAttemptOutcome::Scored {
827 recording_quality,
828 speakers,
829 additional_speakers,
830 } = &first.attempt.outcome
831 else {
832 panic!("expected scored stored outcome");
833 };
834 assert_eq!(*recording_quality, 87);
835 assert_eq!(speakers.len(), 1);
836 assert_eq!(speakers[0].speaker_ordinal, 0);
837 assert_eq!(speakers[0].features.as_ref().len(), FEATURE_COUNT);
838 assert_eq!(additional_speakers.len(), 1);
839 assert_eq!(additional_speakers[0].speaker_ordinal, 1);
840
841 let first_sample = select_and_confirm(
842 &system,
843 "attempt/1",
844 "CLIP0001",
845 "event/select/1",
846 "event/state/1",
847 );
848 let second_sample = select_and_confirm(
849 &system,
850 "attempt/2",
851 "CLIP0002",
852 "event/select/2",
853 "event/state/2",
854 );
855 assert_ne!(first_sample, second_sample);
856
857 let dataset = system.dataset(cohort_id()).unwrap();
858 let rows = kcode_speaker_dataset::active_rows(&dataset);
859 assert_eq!(rows.len(), 2);
860 assert!(
861 rows.iter()
862 .all(|row| row.speaker_id.as_ref() == "speaker/alice")
863 );
864 assert!(
865 rows.iter()
866 .all(|row| row.features.as_ref().len() == FEATURE_COUNT)
867 );
868
869 let model = fit(
870 &dataset,
871 cohort_id(),
872 ModelConfig {
873 mask: ALL_24,
874 components: 1,
875 relevance: 2.0,
876 variance_floor: 0.01,
877 absolute_threshold: -1.0e9,
878 margin_threshold: -1.0e9,
879 },
880 )
881 .unwrap();
882 let before = identify(&model, cohort_id(), &rows[0].features).unwrap();
883 assert!(matches!(
884 before.decision,
885 Decision::Known { ref speaker_id }
886 if speaker_id.as_ref() == "speaker/alice"
887 ));
888
889 let bytes = snapshot_bytes(&model).unwrap();
890 let decoded = snapshot_from_bytes(&bytes).unwrap();
891 assert_eq!(bytes, snapshot_bytes(&decoded).unwrap());
892 assert_eq!(
893 before,
894 identify(&decoded, cohort_id(), &rows[0].features).unwrap()
895 );
896
897 drop(system);
898 fs::remove_file(database).unwrap();
899 }
900
901 #[test]
902 fn incompatible_boundary_inputs_fail_closed() {
903 let database = path();
904 let system = open(&database).unwrap();
905 register(
906 &system,
907 "SOURCE11",
908 "CLIP0011",
909 "event/register/11",
910 "group/source/11",
911 );
912 register(
913 &system,
914 "SOURCE12",
915 "CLIP0012",
916 "event/register/12",
917 "group/source/12",
918 );
919
920 let response = scored_response(20, false);
921 record(
922 &system,
923 "SOURCE11",
924 "CLIP0011",
925 "event/attempt/original",
926 "attempt/original",
927 "RESULT11",
928 &response,
929 );
930
931 let wrong_cohort = system.record_normalized_attempt(NormalizedAttempt {
932 event_id: key("event/wrong/cohort"),
933 attempt_id: key("attempt/wrong/cohort"),
934 cohort_id: key("different/cohort"),
935 source_object: object("SOURCE11"),
936 clip_object: object("CLIP0011"),
937 segment_ordinal: 0,
938 provider_result_object: object("RESULT12"),
939 recording_quality: Some(80),
940 normalized_response: &response,
941 });
942 assert!(matches!(wrong_cohort, Err(SystemError::Boundary(_))));
943 assert!(
944 system
945 .attempt(&key("attempt/wrong/cohort"))
946 .unwrap()
947 .is_none()
948 );
949
950 let wrong_ordinal = system.record_normalized_attempt(NormalizedAttempt {
951 event_id: key("event/wrong/ordinal"),
952 attempt_id: key("attempt/wrong/ordinal"),
953 cohort_id: cohort_id().clone(),
954 source_object: object("SOURCE11"),
955 clip_object: object("CLIP0011"),
956 segment_ordinal: 1,
957 provider_result_object: object("RESULT13"),
958 recording_quality: Some(80),
959 normalized_response: &response,
960 });
961 assert!(matches!(wrong_ordinal, Err(SystemError::Boundary(_))));
962
963 let wrong_source = system.record_normalized_attempt(NormalizedAttempt {
964 event_id: key("event/wrong/source"),
965 attempt_id: key("attempt/wrong/source"),
966 cohort_id: cohort_id().clone(),
967 source_object: object("SOURCE12"),
968 clip_object: object("CLIP0011"),
969 segment_ordinal: 0,
970 provider_result_object: object("RESULT14"),
971 recording_quality: Some(80),
972 normalized_response: &response,
973 });
974 assert!(matches!(wrong_source, Err(SystemError::Boundary(_))));
975
976 let wrong_speaker_ordinal =
977 response.replace("\"speakerOrdinal\":0", "\"speakerOrdinal\":1");
978 let ordinal_result = system.record_normalized_attempt(NormalizedAttempt {
979 event_id: key("event/wrong/speaker"),
980 attempt_id: key("attempt/wrong/speaker"),
981 cohort_id: cohort_id().clone(),
982 source_object: object("SOURCE11"),
983 clip_object: object("CLIP0011"),
984 segment_ordinal: 0,
985 provider_result_object: object("RESULT15"),
986 recording_quality: Some(80),
987 normalized_response: &wrong_speaker_ordinal,
988 });
989 assert!(matches!(ordinal_result, Err(SystemError::Extract(_))));
990
991 let short_features = (0..FEATURE_COUNT - 1)
992 .map(|_| "50")
993 .collect::<Vec<_>>()
994 .join(",");
995 let wrong_feature_count = format!(
996 r#"{{"status":"scored","speakers":[{{"speakerOrdinal":0,"primaryLanguage":"en-US","closestDialect":"General American English","usableSpeechMs":20000,"features":[{short_features}]}}],"additionalSpeakers":[]}}"#
997 );
998 let feature_result = system.record_normalized_attempt(NormalizedAttempt {
999 event_id: key("event/wrong/features"),
1000 attempt_id: key("attempt/wrong/features"),
1001 cohort_id: cohort_id().clone(),
1002 source_object: object("SOURCE11"),
1003 clip_object: object("CLIP0011"),
1004 segment_ordinal: 0,
1005 provider_result_object: object("RESULT16"),
1006 recording_quality: Some(80),
1007 normalized_response: &wrong_feature_count,
1008 });
1009 assert!(matches!(feature_result, Err(SystemError::Extract(_))));
1010
1011 let duplicate_attempt = system.record_normalized_attempt(NormalizedAttempt {
1012 event_id: key("event/duplicate/attempt"),
1013 attempt_id: key("attempt/original"),
1014 cohort_id: cohort_id().clone(),
1015 source_object: object("SOURCE12"),
1016 clip_object: object("CLIP0012"),
1017 segment_ordinal: 0,
1018 provider_result_object: object("RESULT17"),
1019 recording_quality: Some(80),
1020 normalized_response: &response,
1021 });
1022 assert!(matches!(
1023 duplicate_attempt,
1024 Err(SystemError::Store(StoreError::Conflict(_)))
1025 ));
1026
1027 let mismatched_selection = system.select_attempt(AttemptSelectionRequest {
1028 event_id: key("event/wrong/selection"),
1029 attempt_id: key("attempt/original"),
1030 cohort_id: cohort_id().clone(),
1031 clip_object: object("CLIP0012"),
1032 reason: "wrong clip must not select".into(),
1033 });
1034 assert!(matches!(
1035 mismatched_selection,
1036 Err(SystemError::Boundary(_))
1037 ));
1038
1039 let original = system.attempt(&key("attempt/original")).unwrap().unwrap();
1040 assert_eq!(original.attempt.clip_object, object("CLIP0011"));
1041 assert!(!original.selected);
1042
1043 drop(system);
1044 fs::remove_file(database).unwrap();
1045 }
1046}