1use serde::{Deserialize, Serialize};
2use std::fmt::{Display, Formatter};
3use std::path::{Component, Path, PathBuf};
4
5pub use kcode_k1_transaction_id::TxId;
6pub use kcode_speaker_v3_analysis::{ExecutedAnalysis, FeatureVector24, LocalSpeakerLabel};
7
8const EVENT_VERSION: u8 = 3;
9const QUEUE_TAG: u8 = 1;
10const PROCESSED_TAG: u8 = 2;
11const FAILED_TAG: u8 = 3;
12const DISCARDED_TAG: u8 = 4;
13const CONFIRMED_TAG: u8 = 5;
14const FRAGMENT_VERSION: u8 = 1;
15const STAGED_KIND: u8 = 1;
16const FINAL_KIND: u8 = 2;
17const HEADER_LEN: usize = 16;
18const ANALYSIS_SLOT_OFFSET: usize = 16;
19const CONFIRMATION_SLOT_OFFSET: usize = 32;
20const BODY_OFFSET: usize = 48;
21const BASE64_ALPHABET: &[u8; 64] =
22 b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_";
23
24#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
25pub enum AnalysisStageV1 {
26 GeminiTranscript,
27 TerraLabels,
28 GeminiFeatures,
29 TerraStructuring,
30}
31
32#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
33pub struct QueueV2 {
34 #[serde(with = "txid_serde")]
35 pub audio_object_id: TxId,
36}
37
38#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
39pub struct ProcessedV1 {
40 #[serde(with = "txid_serde")]
41 pub queue_id: TxId,
42 #[serde(with = "txid_serde")]
43 pub audio_object_id: TxId,
44 pub analysis: ExecutedAnalysis,
45}
46
47#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
48pub struct FailedV1 {
49 #[serde(with = "txid_serde")]
50 pub queue_id: TxId,
51 #[serde(with = "txid_serde")]
52 pub audio_object_id: TxId,
53 pub final_stage: AnalysisStageV1,
54 pub final_error: String,
55}
56
57#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
58pub struct DiscardedV1 {
59 #[serde(with = "txid_serde")]
60 pub failed_queue_id: TxId,
61 #[serde(with = "txid_serde")]
62 pub audio_object_id: TxId,
63}
64
65#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
66pub struct ConfirmedSpeakerV1 {
67 pub speaker: LocalSpeakerLabel,
68 pub person_id: String,
69}
70
71#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
72pub struct ConfirmedV1 {
73 #[serde(with = "txid_serde")]
74 pub queue_id: TxId,
75 #[serde(with = "txid_serde")]
76 pub audio_object_id: TxId,
77 #[serde(with = "txid_serde")]
78 pub analysis_txid: TxId,
79 pub speakers: Vec<ConfirmedSpeakerV1>,
80}
81
82#[allow(clippy::large_enum_variant)]
83#[derive(Debug, Clone, PartialEq)]
84pub enum AudioClassificationEventV2 {
85 Queue(QueueV2),
86 Processed(ProcessedV1),
87 Failed(FailedV1),
88 Discarded(DiscardedV1),
89 Confirmed(ConfirmedV1),
90}
91
92#[derive(Debug, Clone, PartialEq)]
93pub struct StagedSpeakerV1 {
94 pub speaker: LocalSpeakerLabel,
95 pub language: String,
96 pub features: FeatureVector24,
97 pub usable_for_training: bool,
98}
99
100#[derive(Debug, Clone, PartialEq)]
101pub struct StagedFragmentV1 {
102 pub analysis_txid: TxId,
103 pub transcript: String,
104 pub speakers: Vec<StagedSpeakerV1>,
105}
106
107#[derive(Debug, Clone, PartialEq)]
108pub struct FinalSpeakerV1 {
109 pub speaker: LocalSpeakerLabel,
110 pub person_id: String,
111 pub language: String,
112 pub features: FeatureVector24,
113 pub usable_for_training: bool,
114}
115
116#[derive(Debug, Clone, PartialEq)]
117pub struct FinalFragmentV1 {
118 pub analysis_txid: TxId,
119 pub confirmation_txid: TxId,
120 pub transcript: String,
121 pub speakers: Vec<FinalSpeakerV1>,
122}
123
124#[derive(Debug, Clone, PartialEq, Eq)]
125pub enum FormatError {
126 Truncated,
127 LengthOverflow,
128 UnsupportedVersion(u8),
129 UnknownEventTag(u8),
130 InvalidEventBody,
131 InvalidFragmentKind(u8),
132 NonZeroReserved,
133 NonZeroPadding,
134 NonZeroStagedConfirmation,
135 InvalidTxIdSlotLength(usize),
136 InvalidUtf8,
137 InvalidSpeakerLabel,
138 InvalidFeatureBody,
139 InvalidBoolean(u8),
140 TrailingBytes,
141 InvalidPath,
142}
143
144impl Display for FormatError {
145 fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
146 match self {
147 Self::Truncated => formatter.write_str("truncated input"),
148 Self::LengthOverflow => formatter.write_str("encoded length overflow"),
149 Self::UnsupportedVersion(version) => {
150 write!(formatter, "unsupported format version {version}")
151 }
152 Self::UnknownEventTag(tag) => write!(formatter, "unknown event tag {tag}"),
153 Self::InvalidEventBody => formatter.write_str("invalid event body"),
154 Self::InvalidFragmentKind(kind) => write!(formatter, "invalid fragment kind {kind}"),
155 Self::NonZeroReserved => formatter.write_str("nonzero reserved bytes"),
156 Self::NonZeroPadding => formatter.write_str("nonzero transaction ID slot padding"),
157 Self::NonZeroStagedConfirmation => {
158 formatter.write_str("nonzero staged confirmation slot")
159 }
160 Self::InvalidTxIdSlotLength(length) => {
161 write!(formatter, "invalid transaction ID slot length {length}")
162 }
163 Self::InvalidUtf8 => formatter.write_str("invalid UTF-8"),
164 Self::InvalidSpeakerLabel => formatter.write_str("invalid speaker label"),
165 Self::InvalidFeatureBody => formatter.write_str("invalid feature body"),
166 Self::InvalidBoolean(value) => write!(formatter, "invalid boolean byte {value}"),
167 Self::TrailingBytes => formatter.write_str("trailing bytes"),
168 Self::InvalidPath => formatter.write_str("invalid transaction ID path"),
169 }
170 }
171}
172impl std::error::Error for FormatError {}
173
174#[derive(Debug, Clone, Copy, PartialEq, Eq)]
175pub struct TxIdSlot {
176 txid: TxId,
177}
178impl TxIdSlot {
179 pub const LEN: usize = 16;
180 pub const PADDING_LEN: usize = 4;
181 pub const fn new(txid: TxId) -> Self {
182 Self { txid }
183 }
184 pub const fn txid(self) -> TxId {
185 self.txid
186 }
187 pub fn encode(self) -> [u8; Self::LEN] {
188 let mut encoded = [0_u8; Self::LEN];
189 encoded[..12].copy_from_slice(self.txid.as_bytes());
190 encoded
191 }
192 pub fn decode(bytes: &[u8]) -> Result<Self, FormatError> {
193 if bytes.len() != Self::LEN {
194 return Err(FormatError::InvalidTxIdSlotLength(bytes.len()));
195 }
196 if bytes[12..].iter().any(|byte| *byte != 0) {
197 return Err(FormatError::NonZeroPadding);
198 }
199 let mut txid_bytes = [0_u8; 12];
200 txid_bytes.copy_from_slice(&bytes[..12]);
201 Ok(Self::new(TxId::from_bytes(txid_bytes)))
202 }
203}
204
205pub fn encode_event(event: &AudioClassificationEventV2) -> Result<Vec<u8>, FormatError> {
206 let (tag, body) = match event {
207 AudioClassificationEventV2::Queue(value) => (QUEUE_TAG, postcard::to_allocvec(value)),
208 AudioClassificationEventV2::Processed(value) => {
209 (PROCESSED_TAG, postcard::to_allocvec(value))
210 }
211 AudioClassificationEventV2::Failed(value) => (FAILED_TAG, postcard::to_allocvec(value)),
212 AudioClassificationEventV2::Discarded(value) => {
213 (DISCARDED_TAG, postcard::to_allocvec(value))
214 }
215 AudioClassificationEventV2::Confirmed(value) => {
216 (CONFIRMED_TAG, postcard::to_allocvec(value))
217 }
218 };
219 let body = body.map_err(|_| FormatError::InvalidEventBody)?;
220 let mut encoded = Vec::with_capacity(2 + body.len());
221 encoded.push(EVENT_VERSION);
222 encoded.push(tag);
223 encoded.extend_from_slice(&body);
224 Ok(encoded)
225}
226pub fn decode_event(bytes: &[u8]) -> Result<AudioClassificationEventV2, FormatError> {
227 if bytes.len() < 2 {
228 return Err(FormatError::Truncated);
229 }
230 if bytes[0] != EVENT_VERSION {
231 return Err(FormatError::UnsupportedVersion(bytes[0]));
232 }
233 match bytes[1] {
234 QUEUE_TAG => decode_event_body(&bytes[2..]).map(AudioClassificationEventV2::Queue),
235 PROCESSED_TAG => decode_event_body(&bytes[2..]).map(AudioClassificationEventV2::Processed),
236 FAILED_TAG => decode_event_body(&bytes[2..]).map(AudioClassificationEventV2::Failed),
237 DISCARDED_TAG => decode_event_body(&bytes[2..]).map(AudioClassificationEventV2::Discarded),
238 CONFIRMED_TAG => decode_event_body(&bytes[2..]).map(AudioClassificationEventV2::Confirmed),
239 tag => Err(FormatError::UnknownEventTag(tag)),
240 }
241}
242
243pub fn encode_staged_fragment(value: &StagedFragmentV1) -> Result<Vec<u8>, FormatError> {
244 let mut encoded = encode_fragment_prefix(STAGED_KIND, value.analysis_txid, None);
245 append_string(&mut encoded, &value.transcript)?;
246 append_length(&mut encoded, value.speakers.len())?;
247 for speaker in &value.speakers {
248 append_string(&mut encoded, &speaker.speaker.to_string())?;
249 append_string(&mut encoded, &speaker.language)?;
250 append_features(&mut encoded, &speaker.features)?;
251 encoded.push(u8::from(speaker.usable_for_training));
252 }
253 Ok(encoded)
254}
255pub fn decode_staged_fragment(bytes: &[u8]) -> Result<StagedFragmentV1, FormatError> {
256 let (analysis_txid, _) = decode_fragment_header(bytes, STAGED_KIND, false)?;
257 let mut decoder = BodyDecoder::new(&bytes[BODY_OFFSET..]);
258 let transcript = decoder.read_string()?;
259 let speaker_count = decoder.read_count()?;
260 let mut speakers = Vec::new();
261 for _ in 0..speaker_count {
262 speakers.push(StagedSpeakerV1 {
263 speaker: decoder.read_speaker_label()?,
264 language: decoder.read_string()?,
265 features: decoder.read_features()?,
266 usable_for_training: decoder.read_boolean()?,
267 });
268 }
269 decoder.finish()?;
270 Ok(StagedFragmentV1 {
271 analysis_txid,
272 transcript,
273 speakers,
274 })
275}
276pub fn encode_final_fragment(value: &FinalFragmentV1) -> Result<Vec<u8>, FormatError> {
277 let mut encoded = encode_fragment_prefix(
278 FINAL_KIND,
279 value.analysis_txid,
280 Some(value.confirmation_txid),
281 );
282 append_string(&mut encoded, &value.transcript)?;
283 append_length(&mut encoded, value.speakers.len())?;
284 for speaker in &value.speakers {
285 append_string(&mut encoded, &speaker.speaker.to_string())?;
286 append_string(&mut encoded, &speaker.person_id)?;
287 append_string(&mut encoded, &speaker.language)?;
288 append_features(&mut encoded, &speaker.features)?;
289 encoded.push(u8::from(speaker.usable_for_training));
290 }
291 Ok(encoded)
292}
293pub fn decode_final_fragment(bytes: &[u8]) -> Result<FinalFragmentV1, FormatError> {
294 let (analysis_txid, confirmation_txid) = decode_fragment_header(bytes, FINAL_KIND, true)?;
295 let mut decoder = BodyDecoder::new(&bytes[BODY_OFFSET..]);
296 let transcript = decoder.read_string()?;
297 let speaker_count = decoder.read_count()?;
298 let mut speakers = Vec::new();
299 for _ in 0..speaker_count {
300 speakers.push(FinalSpeakerV1 {
301 speaker: decoder.read_speaker_label()?,
302 person_id: decoder.read_string()?,
303 language: decoder.read_string()?,
304 features: decoder.read_features()?,
305 usable_for_training: decoder.read_boolean()?,
306 });
307 }
308 decoder.finish()?;
309 Ok(FinalFragmentV1 {
310 analysis_txid,
311 confirmation_txid: confirmation_txid.expect("final header has a confirmation slot"),
312 transcript,
313 speakers,
314 })
315}
316
317pub fn txid_path(txid: TxId) -> PathBuf {
318 let encoded = encode_txid_base64(txid);
319 let first = String::from_utf8(encoded[..1].to_vec()).expect("base64 is ASCII");
320 let remaining = String::from_utf8(encoded[1..].to_vec()).expect("base64 is ASCII");
321 PathBuf::from(first).join(format!("{remaining}.dat"))
322}
323pub fn txid_from_path(path: impl AsRef<Path>) -> Result<TxId, FormatError> {
324 let mut components = path.as_ref().components();
325 let first = normal_utf8_component(components.next())?;
326 let filename = normal_utf8_component(components.next())?;
327 if components.next().is_some() || first.len() != 1 {
328 return Err(FormatError::InvalidPath);
329 }
330 let remaining = filename
331 .strip_suffix(".dat")
332 .ok_or(FormatError::InvalidPath)?;
333 if remaining.len() != 15 {
334 return Err(FormatError::InvalidPath);
335 }
336 let mut encoded = [0_u8; 16];
337 encoded[0] = first.as_bytes()[0];
338 encoded[1..].copy_from_slice(remaining.as_bytes());
339 decode_txid_base64(encoded)
340}
341
342fn decode_event_body<T>(bytes: &[u8]) -> Result<T, FormatError>
343where
344 T: for<'de> Deserialize<'de>,
345{
346 let (value, remaining) =
347 postcard::take_from_bytes(bytes).map_err(|_| FormatError::InvalidEventBody)?;
348 if !remaining.is_empty() {
349 return Err(FormatError::TrailingBytes);
350 }
351 Ok(value)
352}
353fn encode_fragment_prefix(kind: u8, analysis: TxId, confirmation: Option<TxId>) -> Vec<u8> {
354 let mut encoded = vec![0_u8; BODY_OFFSET];
355 encoded[0] = FRAGMENT_VERSION;
356 encoded[1] = kind;
357 encoded[ANALYSIS_SLOT_OFFSET..ANALYSIS_SLOT_OFFSET + TxIdSlot::LEN]
358 .copy_from_slice(&TxIdSlot::new(analysis).encode());
359 if let Some(confirmation) = confirmation {
360 encoded[CONFIRMATION_SLOT_OFFSET..CONFIRMATION_SLOT_OFFSET + TxIdSlot::LEN]
361 .copy_from_slice(&TxIdSlot::new(confirmation).encode());
362 }
363 encoded
364}
365fn decode_fragment_header(
366 bytes: &[u8],
367 expected_kind: u8,
368 needs_confirmation: bool,
369) -> Result<(TxId, Option<TxId>), FormatError> {
370 if bytes.len() < BODY_OFFSET {
371 return Err(FormatError::Truncated);
372 }
373 if bytes[0] != FRAGMENT_VERSION {
374 return Err(FormatError::UnsupportedVersion(bytes[0]));
375 }
376 if bytes[1] != expected_kind {
377 return Err(FormatError::InvalidFragmentKind(bytes[1]));
378 }
379 if bytes[2..HEADER_LEN].iter().any(|byte| *byte != 0) {
380 return Err(FormatError::NonZeroReserved);
381 }
382 let analysis =
383 TxIdSlot::decode(&bytes[ANALYSIS_SLOT_OFFSET..ANALYSIS_SLOT_OFFSET + TxIdSlot::LEN])?
384 .txid();
385 let confirmation_bytes =
386 &bytes[CONFIRMATION_SLOT_OFFSET..CONFIRMATION_SLOT_OFFSET + TxIdSlot::LEN];
387 let confirmation = if needs_confirmation {
388 Some(TxIdSlot::decode(confirmation_bytes)?.txid())
389 } else {
390 if confirmation_bytes.iter().any(|byte| *byte != 0) {
391 return Err(FormatError::NonZeroStagedConfirmation);
392 }
393 None
394 };
395 Ok((analysis, confirmation))
396}
397fn append_length(encoded: &mut Vec<u8>, length: usize) -> Result<(), FormatError> {
398 encoded.extend_from_slice(
399 &u64::try_from(length)
400 .map_err(|_| FormatError::LengthOverflow)?
401 .to_le_bytes(),
402 );
403 Ok(())
404}
405fn append_bytes(encoded: &mut Vec<u8>, bytes: &[u8]) -> Result<(), FormatError> {
406 append_length(encoded, bytes.len())?;
407 encoded.extend_from_slice(bytes);
408 Ok(())
409}
410fn append_string(encoded: &mut Vec<u8>, value: &str) -> Result<(), FormatError> {
411 append_bytes(encoded, value.as_bytes())
412}
413fn append_features(encoded: &mut Vec<u8>, features: &FeatureVector24) -> Result<(), FormatError> {
414 append_bytes(
415 encoded,
416 &postcard::to_allocvec(features).map_err(|_| FormatError::InvalidFeatureBody)?,
417 )
418}
419
420struct BodyDecoder<'a> {
421 bytes: &'a [u8],
422 position: usize,
423}
424impl<'a> BodyDecoder<'a> {
425 fn new(bytes: &'a [u8]) -> Self {
426 Self { bytes, position: 0 }
427 }
428 fn take(&mut self, length: usize) -> Result<&'a [u8], FormatError> {
429 let end = self
430 .position
431 .checked_add(length)
432 .ok_or(FormatError::LengthOverflow)?;
433 let value = self
434 .bytes
435 .get(self.position..end)
436 .ok_or(FormatError::Truncated)?;
437 self.position = end;
438 Ok(value)
439 }
440 fn read_u64(&mut self) -> Result<u64, FormatError> {
441 let bytes = self.take(8)?;
442 let mut value = [0_u8; 8];
443 value.copy_from_slice(bytes);
444 Ok(u64::from_le_bytes(value))
445 }
446 fn read_length(&mut self) -> Result<usize, FormatError> {
447 usize::try_from(self.read_u64()?).map_err(|_| FormatError::LengthOverflow)
448 }
449 fn read_count(&mut self) -> Result<usize, FormatError> {
450 self.read_length()
451 }
452 fn read_bytes(&mut self) -> Result<&'a [u8], FormatError> {
453 let length = self.read_length()?;
454 self.take(length)
455 }
456 fn read_string(&mut self) -> Result<String, FormatError> {
457 Ok(std::str::from_utf8(self.read_bytes()?)
458 .map_err(|_| FormatError::InvalidUtf8)?
459 .to_owned())
460 }
461 fn read_speaker_label(&mut self) -> Result<LocalSpeakerLabel, FormatError> {
462 self.read_string()?
463 .parse()
464 .map_err(|_| FormatError::InvalidSpeakerLabel)
465 }
466 fn read_features(&mut self) -> Result<FeatureVector24, FormatError> {
467 let bytes = self.read_bytes()?;
468 let (features, remaining) =
469 postcard::take_from_bytes(bytes).map_err(|_| FormatError::InvalidFeatureBody)?;
470 if !remaining.is_empty() {
471 return Err(FormatError::InvalidFeatureBody);
472 }
473 Ok(features)
474 }
475 fn read_boolean(&mut self) -> Result<bool, FormatError> {
476 match self.take(1)?[0] {
477 0 => Ok(false),
478 1 => Ok(true),
479 value => Err(FormatError::InvalidBoolean(value)),
480 }
481 }
482 fn finish(self) -> Result<(), FormatError> {
483 if self.position == self.bytes.len() {
484 Ok(())
485 } else {
486 Err(FormatError::TrailingBytes)
487 }
488 }
489}
490fn normal_utf8_component(component: Option<Component<'_>>) -> Result<&str, FormatError> {
491 match component {
492 Some(Component::Normal(value)) => value.to_str().ok_or(FormatError::InvalidPath),
493 _ => Err(FormatError::InvalidPath),
494 }
495}
496fn encode_txid_base64(txid: TxId) -> [u8; 16] {
497 let bytes = txid.into_bytes();
498 let mut encoded = [0_u8; 16];
499 for group in 0..4 {
500 let input = group * 3;
501 let output = group * 4;
502 encoded[output] = BASE64_ALPHABET[(bytes[input] >> 2) as usize];
503 encoded[output + 1] =
504 BASE64_ALPHABET[(((bytes[input] & 3) << 4) | (bytes[input + 1] >> 4)) as usize];
505 encoded[output + 2] =
506 BASE64_ALPHABET[(((bytes[input + 1] & 15) << 2) | (bytes[input + 2] >> 6)) as usize];
507 encoded[output + 3] = BASE64_ALPHABET[(bytes[input + 2] & 63) as usize];
508 }
509 encoded
510}
511fn decode_txid_base64(encoded: [u8; 16]) -> Result<TxId, FormatError> {
512 let mut bytes = [0_u8; 12];
513 for group in 0..4 {
514 let input = group * 4;
515 let output = group * 3;
516 let first = decode_base64_character(encoded[input])?;
517 let second = decode_base64_character(encoded[input + 1])?;
518 let third = decode_base64_character(encoded[input + 2])?;
519 let fourth = decode_base64_character(encoded[input + 3])?;
520 bytes[output] = (first << 2) | (second >> 4);
521 bytes[output + 1] = (second << 4) | (third >> 2);
522 bytes[output + 2] = (third << 6) | fourth;
523 }
524 Ok(TxId::from_bytes(bytes))
525}
526fn decode_base64_character(value: u8) -> Result<u8, FormatError> {
527 match value {
528 b'A'..=b'Z' => Ok(value - b'A'),
529 b'a'..=b'z' => Ok(value - b'a' + 26),
530 b'0'..=b'9' => Ok(value - b'0' + 52),
531 b'-' => Ok(62),
532 b'_' => Ok(63),
533 _ => Err(FormatError::InvalidPath),
534 }
535}
536mod txid_serde {
537 use super::TxId;
538 use serde::{Deserialize, Deserializer, Serialize, Serializer};
539 pub fn serialize<S>(value: &TxId, serializer: S) -> Result<S::Ok, S::Error>
540 where
541 S: Serializer,
542 {
543 value.into_bytes().serialize(serializer)
544 }
545 pub fn deserialize<'de, D>(deserializer: D) -> Result<TxId, D::Error>
546 where
547 D: Deserializer<'de>,
548 {
549 Ok(TxId::from_bytes(<[u8; 12]>::deserialize(deserializer)?))
550 }
551}
552
553#[cfg(test)]
554mod tests {
555 use super::*;
556 use kcode_speaker_v3_analysis::{
557 AnalysisEnvelope, GeminiCohort, OggAudioMetadata, StructuredAnalysis, StructuredSpeaker,
558 StructurerProvenance,
559 };
560 fn txid(seed: u8) -> TxId {
561 let mut bytes = [0_u8; 12];
562 for (index, byte) in bytes.iter_mut().enumerate() {
563 *byte = seed.wrapping_add(index as u8);
564 }
565 TxId::from_bytes(bytes)
566 }
567 fn features(seed: f64) -> FeatureVector24 {
568 FeatureVector24 {
569 median_f0_hz: Some(seed),
570 ..FeatureVector24::default()
571 }
572 }
573 fn executed_analysis() -> ExecutedAnalysis {
574 let mut ogg = vec![0_u8; 29];
575 ogg[..4].copy_from_slice(b"OggS");
576 ogg[26] = 1;
577 ogg[27] = 1;
578 let audio = OggAudioMetadata::from_bytes(&ogg, 1_250, Some("sample.ogg".to_owned()))
579 .expect("valid Ogg metadata");
580 let analysis = StructuredAnalysis {
581 transcript: "Speaker 1".to_owned(),
582 speakers: vec![StructuredSpeaker {
583 speaker: LocalSpeakerLabel::new(1).expect("valid label"),
584 language: "English".to_owned(),
585 features: features(182.5),
586 features_usable_for_training: true,
587 }],
588 };
589 ExecutedAnalysis {
590 envelope: AnalysisEnvelope {
591 audio,
592 analysis,
593 gemini: GeminiCohort::new("gemini-3.1-pro"),
594 structurer: StructurerProvenance::new("terra-structurer"),
595 },
596 label_extractor: StructurerProvenance::new("terra-labeler"),
597 }
598 }
599 fn staged_fragment() -> StagedFragmentV1 {
600 StagedFragmentV1 {
601 analysis_txid: txid(20),
602 transcript: "hello".to_owned(),
603 speakers: vec![StagedSpeakerV1 {
604 speaker: LocalSpeakerLabel::new(1).expect("valid label"),
605 language: "English".to_owned(),
606 features: features(201.25),
607 usable_for_training: true,
608 }],
609 }
610 }
611 fn final_fragment() -> FinalFragmentV1 {
612 FinalFragmentV1 {
613 analysis_txid: txid(30),
614 confirmation_txid: txid(40),
615 transcript: "hello".to_owned(),
616 speakers: vec![FinalSpeakerV1 {
617 speaker: LocalSpeakerLabel::new(1).expect("valid label"),
618 person_id: "person-1".to_owned(),
619 language: "English".to_owned(),
620 features: features(201.25),
621 usable_for_training: false,
622 }],
623 }
624 }
625 #[test]
626 fn txid_slot_roundtrips_and_rejects_nonzero_padding() {
627 let id = txid(3);
628 let encoded = TxIdSlot::new(id).encode();
629 assert_eq!(encoded.len(), 16);
630 assert_eq!(TxIdSlot::decode(&encoded).expect("slot").txid(), id);
631 let mut corrupt = encoded;
632 corrupt[12] = 1;
633 assert_eq!(TxIdSlot::decode(&corrupt), Err(FormatError::NonZeroPadding));
634 }
635 #[test]
636 fn events_use_v3_stable_tags_and_queue_has_only_audio_id() {
637 let events = vec![
638 AudioClassificationEventV2::Queue(QueueV2 {
639 audio_object_id: txid(1),
640 }),
641 AudioClassificationEventV2::Processed(ProcessedV1 {
642 queue_id: txid(3),
643 audio_object_id: txid(4),
644 analysis: executed_analysis(),
645 }),
646 AudioClassificationEventV2::Failed(FailedV1 {
647 queue_id: txid(5),
648 audio_object_id: txid(6),
649 final_stage: AnalysisStageV1::GeminiFeatures,
650 final_error: "failure".to_owned(),
651 }),
652 AudioClassificationEventV2::Discarded(DiscardedV1 {
653 failed_queue_id: txid(7),
654 audio_object_id: txid(8),
655 }),
656 AudioClassificationEventV2::Confirmed(ConfirmedV1 {
657 queue_id: txid(9),
658 audio_object_id: txid(10),
659 analysis_txid: txid(11),
660 speakers: vec![],
661 }),
662 ];
663 for (index, event) in events.into_iter().enumerate() {
664 let encoded = encode_event(&event).expect("encode event");
665 assert_eq!(encoded[0], 3);
666 assert_eq!(encoded[1], index as u8 + 1);
667 assert_eq!(decode_event(&encoded), Ok(event));
668 }
669 }
670 #[test]
671 fn event_v2_and_invalid_bodies_are_rejected() {
672 let event = AudioClassificationEventV2::Queue(QueueV2 {
673 audio_object_id: txid(1),
674 });
675 let encoded = encode_event(&event).expect("encode");
676 let mut v2 = encoded.clone();
677 v2[0] = 2;
678 assert_eq!(decode_event(&v2), Err(FormatError::UnsupportedVersion(2)));
679 assert_eq!(
680 decode_event(&encoded[..2]),
681 Err(FormatError::InvalidEventBody)
682 );
683 let mut trailing = encoded;
684 trailing.push(0);
685 assert_eq!(decode_event(&trailing), Err(FormatError::TrailingBytes));
686 }
687 #[test]
688 fn fragments_roundtrip_without_format_changes() {
689 let staged = staged_fragment();
690 let staged_bytes = encode_staged_fragment(&staged).expect("encode staged");
691 assert_eq!(
692 &staged_bytes[16..32],
693 &TxIdSlot::new(staged.analysis_txid).encode()
694 );
695 assert_eq!(&staged_bytes[32..48], &[0; 16]);
696 assert_eq!(decode_staged_fragment(&staged_bytes), Ok(staged));
697 let final_value = final_fragment();
698 let final_bytes = encode_final_fragment(&final_value).expect("encode final");
699 assert_eq!(decode_final_fragment(&final_bytes), Ok(final_value));
700 }
701 #[test]
702 fn fragment_corruption_is_rejected() {
703 let mut bytes = encode_staged_fragment(&staged_fragment()).expect("encode");
704 bytes[2] = 1;
705 assert_eq!(
706 decode_staged_fragment(&bytes),
707 Err(FormatError::NonZeroReserved)
708 );
709 }
710 #[test]
711 fn paths_roundtrip() {
712 let id = txid(99);
713 assert_eq!(txid_from_path(txid_path(id)), Ok(id));
714 assert_eq!(
715 txid_from_path("A/invalid.dat"),
716 Err(FormatError::InvalidPath)
717 );
718 }
719}