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