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 = 2;
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 QueueV1 {
34 #[serde(with = "txid_serde")]
35 pub audio_object_id: TxId,
36 pub duration_ms: u64,
37 pub filename: Option<String>,
38}
39
40#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
41pub struct ProcessedV1 {
42 #[serde(with = "txid_serde")]
43 pub queue_id: TxId,
44 #[serde(with = "txid_serde")]
45 pub audio_object_id: TxId,
46 pub analysis: ExecutedAnalysis,
47}
48
49#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
50pub struct FailedV1 {
51 #[serde(with = "txid_serde")]
52 pub queue_id: TxId,
53 #[serde(with = "txid_serde")]
54 pub audio_object_id: TxId,
55 pub final_stage: AnalysisStageV1,
56 pub final_error: String,
57}
58
59#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
60pub struct DiscardedV1 {
61 #[serde(with = "txid_serde")]
62 pub failed_queue_id: TxId,
63 #[serde(with = "txid_serde")]
64 pub audio_object_id: TxId,
65}
66
67#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
68pub struct ConfirmedSpeakerV1 {
69 pub speaker: LocalSpeakerLabel,
70 pub person_id: String,
71}
72
73#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
74pub struct ConfirmedV1 {
75 #[serde(with = "txid_serde")]
76 pub queue_id: TxId,
77 #[serde(with = "txid_serde")]
78 pub audio_object_id: TxId,
79 #[serde(with = "txid_serde")]
80 pub analysis_txid: TxId,
81 pub speakers: Vec<ConfirmedSpeakerV1>,
82}
83
84#[allow(clippy::large_enum_variant)]
85#[derive(Debug, Clone, PartialEq)]
86pub enum AudioClassificationEventV1 {
87 Queue(QueueV1),
88 Processed(ProcessedV1),
89 Failed(FailedV1),
90 Discarded(DiscardedV1),
91 Confirmed(ConfirmedV1),
92}
93
94#[derive(Debug, Clone, PartialEq)]
95pub struct StagedSpeakerV1 {
96 pub speaker: LocalSpeakerLabel,
97 pub language: String,
98 pub features: FeatureVector24,
99 pub usable_for_training: bool,
100}
101
102#[derive(Debug, Clone, PartialEq)]
103pub struct StagedFragmentV1 {
104 pub analysis_txid: TxId,
105 pub transcript: String,
106 pub speakers: Vec<StagedSpeakerV1>,
107}
108
109#[derive(Debug, Clone, PartialEq)]
110pub struct FinalSpeakerV1 {
111 pub speaker: LocalSpeakerLabel,
112 pub person_id: String,
113 pub language: String,
114 pub features: FeatureVector24,
115 pub usable_for_training: bool,
116}
117
118#[derive(Debug, Clone, PartialEq)]
119pub struct FinalFragmentV1 {
120 pub analysis_txid: TxId,
121 pub confirmation_txid: TxId,
122 pub transcript: String,
123 pub speakers: Vec<FinalSpeakerV1>,
124}
125
126#[derive(Debug, Clone, PartialEq, Eq)]
127pub enum FormatError {
128 Truncated,
129 LengthOverflow,
130 UnsupportedVersion(u8),
131 UnknownEventTag(u8),
132 InvalidEventBody,
133 InvalidFragmentKind(u8),
134 NonZeroReserved,
135 NonZeroPadding,
136 NonZeroStagedConfirmation,
137 InvalidTxIdSlotLength(usize),
138 InvalidUtf8,
139 InvalidSpeakerLabel,
140 InvalidFeatureBody,
141 InvalidBoolean(u8),
142 TrailingBytes,
143 InvalidPath,
144}
145
146impl Display for FormatError {
147 fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
148 match self {
149 Self::Truncated => formatter.write_str("truncated input"),
150 Self::LengthOverflow => formatter.write_str("encoded length overflow"),
151 Self::UnsupportedVersion(version) => {
152 write!(formatter, "unsupported format version {version}")
153 }
154 Self::UnknownEventTag(tag) => write!(formatter, "unknown event tag {tag}"),
155 Self::InvalidEventBody => formatter.write_str("invalid event body"),
156 Self::InvalidFragmentKind(kind) => write!(formatter, "invalid fragment kind {kind}"),
157 Self::NonZeroReserved => formatter.write_str("nonzero reserved bytes"),
158 Self::NonZeroPadding => formatter.write_str("nonzero transaction ID slot padding"),
159 Self::NonZeroStagedConfirmation => {
160 formatter.write_str("nonzero staged confirmation slot")
161 }
162 Self::InvalidTxIdSlotLength(length) => {
163 write!(formatter, "invalid transaction ID slot length {length}")
164 }
165 Self::InvalidUtf8 => formatter.write_str("invalid UTF-8"),
166 Self::InvalidSpeakerLabel => formatter.write_str("invalid speaker label"),
167 Self::InvalidFeatureBody => formatter.write_str("invalid feature body"),
168 Self::InvalidBoolean(value) => write!(formatter, "invalid boolean byte {value}"),
169 Self::TrailingBytes => formatter.write_str("trailing bytes"),
170 Self::InvalidPath => formatter.write_str("invalid transaction ID path"),
171 }
172 }
173}
174
175impl std::error::Error for FormatError {}
176
177#[derive(Debug, Clone, Copy, PartialEq, Eq)]
178pub struct TxIdSlot {
179 txid: TxId,
180}
181
182impl TxIdSlot {
183 pub const LEN: usize = 16;
184 pub const PADDING_LEN: usize = 4;
185 pub const fn new(txid: TxId) -> Self {
186 Self { txid }
187 }
188 pub const fn txid(self) -> TxId {
189 self.txid
190 }
191 pub fn encode(self) -> [u8; Self::LEN] {
192 let mut encoded = [0_u8; Self::LEN];
193 encoded[..12].copy_from_slice(self.txid.as_bytes());
194 encoded
195 }
196 pub fn decode(bytes: &[u8]) -> Result<Self, FormatError> {
197 if bytes.len() != Self::LEN {
198 return Err(FormatError::InvalidTxIdSlotLength(bytes.len()));
199 }
200 if bytes[12..].iter().any(|byte| *byte != 0) {
201 return Err(FormatError::NonZeroPadding);
202 }
203 let mut txid_bytes = [0_u8; 12];
204 txid_bytes.copy_from_slice(&bytes[..12]);
205 Ok(Self::new(TxId::from_bytes(txid_bytes)))
206 }
207}
208
209pub fn encode_event(event: &AudioClassificationEventV1) -> Result<Vec<u8>, FormatError> {
210 let (tag, body) = match event {
211 AudioClassificationEventV1::Queue(value) => (QUEUE_TAG, postcard::to_allocvec(value)),
212 AudioClassificationEventV1::Processed(value) => {
213 (PROCESSED_TAG, postcard::to_allocvec(value))
214 }
215 AudioClassificationEventV1::Failed(value) => (FAILED_TAG, postcard::to_allocvec(value)),
216 AudioClassificationEventV1::Discarded(value) => {
217 (DISCARDED_TAG, postcard::to_allocvec(value))
218 }
219 AudioClassificationEventV1::Confirmed(value) => {
220 (CONFIRMED_TAG, postcard::to_allocvec(value))
221 }
222 };
223 let body = body.map_err(|_| FormatError::InvalidEventBody)?;
224 let mut encoded = Vec::with_capacity(2 + body.len());
225 encoded.push(EVENT_VERSION);
226 encoded.push(tag);
227 encoded.extend_from_slice(&body);
228 Ok(encoded)
229}
230
231pub fn decode_event(bytes: &[u8]) -> Result<AudioClassificationEventV1, FormatError> {
232 if bytes.len() < 2 {
233 return Err(FormatError::Truncated);
234 }
235 if bytes[0] != EVENT_VERSION {
236 return Err(FormatError::UnsupportedVersion(bytes[0]));
237 }
238 let body = &bytes[2..];
239 match bytes[1] {
240 QUEUE_TAG => decode_event_body(body).map(AudioClassificationEventV1::Queue),
241 PROCESSED_TAG => decode_event_body(body).map(AudioClassificationEventV1::Processed),
242 FAILED_TAG => decode_event_body(body).map(AudioClassificationEventV1::Failed),
243 DISCARDED_TAG => decode_event_body(body).map(AudioClassificationEventV1::Discarded),
244 CONFIRMED_TAG => decode_event_body(body).map(AudioClassificationEventV1::Confirmed),
245 tag => Err(FormatError::UnknownEventTag(tag)),
246 }
247}
248
249pub fn encode_staged_fragment(value: &StagedFragmentV1) -> Result<Vec<u8>, FormatError> {
250 let mut encoded = encode_fragment_prefix(STAGED_KIND, value.analysis_txid, None);
251 append_string(&mut encoded, &value.transcript)?;
252 append_length(&mut encoded, value.speakers.len())?;
253 for speaker in &value.speakers {
254 append_string(&mut encoded, &speaker.speaker.to_string())?;
255 append_string(&mut encoded, &speaker.language)?;
256 append_features(&mut encoded, &speaker.features)?;
257 encoded.push(u8::from(speaker.usable_for_training));
258 }
259 Ok(encoded)
260}
261
262pub fn decode_staged_fragment(bytes: &[u8]) -> Result<StagedFragmentV1, FormatError> {
263 let (analysis_txid, _) = decode_fragment_header(bytes, STAGED_KIND, false)?;
264 let mut decoder = BodyDecoder::new(&bytes[BODY_OFFSET..]);
265 let transcript = decoder.read_string()?;
266 let speaker_count = decoder.read_count()?;
267 let mut speakers = Vec::new();
268 for _ in 0..speaker_count {
269 speakers.push(StagedSpeakerV1 {
270 speaker: decoder.read_speaker_label()?,
271 language: decoder.read_string()?,
272 features: decoder.read_features()?,
273 usable_for_training: decoder.read_boolean()?,
274 });
275 }
276 decoder.finish()?;
277 Ok(StagedFragmentV1 {
278 analysis_txid,
279 transcript,
280 speakers,
281 })
282}
283
284pub fn encode_final_fragment(value: &FinalFragmentV1) -> Result<Vec<u8>, FormatError> {
285 let mut encoded = encode_fragment_prefix(
286 FINAL_KIND,
287 value.analysis_txid,
288 Some(value.confirmation_txid),
289 );
290 append_string(&mut encoded, &value.transcript)?;
291 append_length(&mut encoded, value.speakers.len())?;
292 for speaker in &value.speakers {
293 append_string(&mut encoded, &speaker.speaker.to_string())?;
294 append_string(&mut encoded, &speaker.person_id)?;
295 append_string(&mut encoded, &speaker.language)?;
296 append_features(&mut encoded, &speaker.features)?;
297 encoded.push(u8::from(speaker.usable_for_training));
298 }
299 Ok(encoded)
300}
301
302pub fn decode_final_fragment(bytes: &[u8]) -> Result<FinalFragmentV1, FormatError> {
303 let (analysis_txid, confirmation_txid) = decode_fragment_header(bytes, FINAL_KIND, true)?;
304 let mut decoder = BodyDecoder::new(&bytes[BODY_OFFSET..]);
305 let transcript = decoder.read_string()?;
306 let speaker_count = decoder.read_count()?;
307 let mut speakers = Vec::new();
308 for _ in 0..speaker_count {
309 speakers.push(FinalSpeakerV1 {
310 speaker: decoder.read_speaker_label()?,
311 person_id: decoder.read_string()?,
312 language: decoder.read_string()?,
313 features: decoder.read_features()?,
314 usable_for_training: decoder.read_boolean()?,
315 });
316 }
317 decoder.finish()?;
318 Ok(FinalFragmentV1 {
319 analysis_txid,
320 confirmation_txid: confirmation_txid.expect("final header has a confirmation slot"),
321 transcript,
322 speakers,
323 })
324}
325
326pub fn txid_path(txid: TxId) -> PathBuf {
327 let encoded = encode_txid_base64(txid);
328 let first = String::from_utf8(encoded[..1].to_vec()).expect("base64 is ASCII");
329 let remaining = String::from_utf8(encoded[1..].to_vec()).expect("base64 is ASCII");
330 PathBuf::from(first).join(format!("{remaining}.dat"))
331}
332
333pub fn txid_from_path(path: impl AsRef<Path>) -> Result<TxId, FormatError> {
334 let mut components = path.as_ref().components();
335 let first = normal_utf8_component(components.next())?;
336 let filename = normal_utf8_component(components.next())?;
337 if components.next().is_some() || first.len() != 1 {
338 return Err(FormatError::InvalidPath);
339 }
340 let remaining = filename
341 .strip_suffix(".dat")
342 .ok_or(FormatError::InvalidPath)?;
343 if remaining.len() != 15 {
344 return Err(FormatError::InvalidPath);
345 }
346 let mut encoded = [0_u8; 16];
347 encoded[0] = first.as_bytes()[0];
348 encoded[1..].copy_from_slice(remaining.as_bytes());
349 decode_txid_base64(encoded)
350}
351
352fn decode_event_body<T>(bytes: &[u8]) -> Result<T, FormatError>
353where
354 T: for<'de> Deserialize<'de>,
355{
356 let (value, remaining) =
357 postcard::take_from_bytes(bytes).map_err(|_| FormatError::InvalidEventBody)?;
358 if !remaining.is_empty() {
359 return Err(FormatError::TrailingBytes);
360 }
361 Ok(value)
362}
363
364fn encode_fragment_prefix(kind: u8, analysis: TxId, confirmation: Option<TxId>) -> Vec<u8> {
365 let mut encoded = vec![0_u8; BODY_OFFSET];
366 encoded[0] = FRAGMENT_VERSION;
367 encoded[1] = kind;
368 encoded[ANALYSIS_SLOT_OFFSET..ANALYSIS_SLOT_OFFSET + TxIdSlot::LEN]
369 .copy_from_slice(&TxIdSlot::new(analysis).encode());
370 if let Some(confirmation) = confirmation {
371 encoded[CONFIRMATION_SLOT_OFFSET..CONFIRMATION_SLOT_OFFSET + TxIdSlot::LEN]
372 .copy_from_slice(&TxIdSlot::new(confirmation).encode());
373 }
374 encoded
375}
376
377fn decode_fragment_header(
378 bytes: &[u8],
379 expected_kind: u8,
380 needs_confirmation: bool,
381) -> Result<(TxId, Option<TxId>), FormatError> {
382 if bytes.len() < BODY_OFFSET {
383 return Err(FormatError::Truncated);
384 }
385 if bytes[0] != FRAGMENT_VERSION {
386 return Err(FormatError::UnsupportedVersion(bytes[0]));
387 }
388 if bytes[1] != expected_kind {
389 return Err(FormatError::InvalidFragmentKind(bytes[1]));
390 }
391 if bytes[2..HEADER_LEN].iter().any(|byte| *byte != 0) {
392 return Err(FormatError::NonZeroReserved);
393 }
394 let analysis =
395 TxIdSlot::decode(&bytes[ANALYSIS_SLOT_OFFSET..ANALYSIS_SLOT_OFFSET + TxIdSlot::LEN])?
396 .txid();
397 let confirmation_bytes =
398 &bytes[CONFIRMATION_SLOT_OFFSET..CONFIRMATION_SLOT_OFFSET + TxIdSlot::LEN];
399 let confirmation = if needs_confirmation {
400 Some(TxIdSlot::decode(confirmation_bytes)?.txid())
401 } else {
402 if confirmation_bytes.iter().any(|byte| *byte != 0) {
403 return Err(FormatError::NonZeroStagedConfirmation);
404 }
405 None
406 };
407 Ok((analysis, confirmation))
408}
409
410fn append_length(encoded: &mut Vec<u8>, length: usize) -> Result<(), FormatError> {
411 encoded.extend_from_slice(
412 &u64::try_from(length)
413 .map_err(|_| FormatError::LengthOverflow)?
414 .to_le_bytes(),
415 );
416 Ok(())
417}
418fn append_bytes(encoded: &mut Vec<u8>, bytes: &[u8]) -> Result<(), FormatError> {
419 append_length(encoded, bytes.len())?;
420 encoded.extend_from_slice(bytes);
421 Ok(())
422}
423fn append_string(encoded: &mut Vec<u8>, value: &str) -> Result<(), FormatError> {
424 append_bytes(encoded, value.as_bytes())
425}
426fn append_features(encoded: &mut Vec<u8>, features: &FeatureVector24) -> Result<(), FormatError> {
427 append_bytes(
428 encoded,
429 &postcard::to_allocvec(features).map_err(|_| FormatError::InvalidFeatureBody)?,
430 )
431}
432
433struct BodyDecoder<'a> {
434 bytes: &'a [u8],
435 position: usize,
436}
437impl<'a> BodyDecoder<'a> {
438 fn new(bytes: &'a [u8]) -> Self {
439 Self { bytes, position: 0 }
440 }
441 fn take(&mut self, length: usize) -> Result<&'a [u8], FormatError> {
442 let end = self
443 .position
444 .checked_add(length)
445 .ok_or(FormatError::LengthOverflow)?;
446 let value = self
447 .bytes
448 .get(self.position..end)
449 .ok_or(FormatError::Truncated)?;
450 self.position = end;
451 Ok(value)
452 }
453 fn read_u64(&mut self) -> Result<u64, FormatError> {
454 let bytes = self.take(8)?;
455 let mut value = [0_u8; 8];
456 value.copy_from_slice(bytes);
457 Ok(u64::from_le_bytes(value))
458 }
459 fn read_length(&mut self) -> Result<usize, FormatError> {
460 usize::try_from(self.read_u64()?).map_err(|_| FormatError::LengthOverflow)
461 }
462 fn read_count(&mut self) -> Result<usize, FormatError> {
463 self.read_length()
464 }
465 fn read_bytes(&mut self) -> Result<&'a [u8], FormatError> {
466 let length = self.read_length()?;
467 self.take(length)
468 }
469 fn read_string(&mut self) -> Result<String, FormatError> {
470 Ok(std::str::from_utf8(self.read_bytes()?)
471 .map_err(|_| FormatError::InvalidUtf8)?
472 .to_owned())
473 }
474 fn read_speaker_label(&mut self) -> Result<LocalSpeakerLabel, FormatError> {
475 self.read_string()?
476 .parse()
477 .map_err(|_| FormatError::InvalidSpeakerLabel)
478 }
479 fn read_features(&mut self) -> Result<FeatureVector24, FormatError> {
480 let bytes = self.read_bytes()?;
481 let (features, remaining) =
482 postcard::take_from_bytes(bytes).map_err(|_| FormatError::InvalidFeatureBody)?;
483 if !remaining.is_empty() {
484 return Err(FormatError::InvalidFeatureBody);
485 }
486 Ok(features)
487 }
488 fn read_boolean(&mut self) -> Result<bool, FormatError> {
489 match self.take(1)?[0] {
490 0 => Ok(false),
491 1 => Ok(true),
492 value => Err(FormatError::InvalidBoolean(value)),
493 }
494 }
495 fn finish(self) -> Result<(), FormatError> {
496 if self.position == self.bytes.len() {
497 Ok(())
498 } else {
499 Err(FormatError::TrailingBytes)
500 }
501 }
502}
503
504fn normal_utf8_component(component: Option<Component<'_>>) -> Result<&str, FormatError> {
505 match component {
506 Some(Component::Normal(value)) => value.to_str().ok_or(FormatError::InvalidPath),
507 _ => Err(FormatError::InvalidPath),
508 }
509}
510
511fn encode_txid_base64(txid: TxId) -> [u8; 16] {
512 let bytes = txid.into_bytes();
513 let mut encoded = [0_u8; 16];
514 for group in 0..4 {
515 let input = group * 3;
516 let output = group * 4;
517 encoded[output] = BASE64_ALPHABET[(bytes[input] >> 2) as usize];
518 encoded[output + 1] =
519 BASE64_ALPHABET[(((bytes[input] & 3) << 4) | (bytes[input + 1] >> 4)) as usize];
520 encoded[output + 2] =
521 BASE64_ALPHABET[(((bytes[input + 1] & 15) << 2) | (bytes[input + 2] >> 6)) as usize];
522 encoded[output + 3] = BASE64_ALPHABET[(bytes[input + 2] & 63) as usize];
523 }
524 encoded
525}
526
527fn decode_txid_base64(encoded: [u8; 16]) -> Result<TxId, FormatError> {
528 let mut bytes = [0_u8; 12];
529 for group in 0..4 {
530 let input = group * 4;
531 let output = group * 3;
532 let first = decode_base64_character(encoded[input])?;
533 let second = decode_base64_character(encoded[input + 1])?;
534 let third = decode_base64_character(encoded[input + 2])?;
535 let fourth = decode_base64_character(encoded[input + 3])?;
536 bytes[output] = (first << 2) | (second >> 4);
537 bytes[output + 1] = (second << 4) | (third >> 2);
538 bytes[output + 2] = (third << 6) | fourth;
539 }
540 Ok(TxId::from_bytes(bytes))
541}
542
543fn decode_base64_character(value: u8) -> Result<u8, FormatError> {
544 match value {
545 b'A'..=b'Z' => Ok(value - b'A'),
546 b'a'..=b'z' => Ok(value - b'a' + 26),
547 b'0'..=b'9' => Ok(value - b'0' + 52),
548 b'-' => Ok(62),
549 b'_' => Ok(63),
550 _ => Err(FormatError::InvalidPath),
551 }
552}
553
554mod txid_serde {
555 use super::TxId;
556 use serde::{Deserialize, Deserializer, Serialize, Serializer};
557 pub fn serialize<S>(value: &TxId, serializer: S) -> Result<S::Ok, S::Error>
558 where
559 S: Serializer,
560 {
561 value.into_bytes().serialize(serializer)
562 }
563 pub fn deserialize<'de, D>(deserializer: D) -> Result<TxId, D::Error>
564 where
565 D: Deserializer<'de>,
566 {
567 Ok(TxId::from_bytes(<[u8; 12]>::deserialize(deserializer)?))
568 }
569}
570
571#[cfg(test)]
572mod tests {
573 use super::*;
574 use kcode_speaker_v3_analysis::{
575 AnalysisEnvelope, GeminiCohort, OggAudioMetadata, StructuredAnalysis, StructuredSpeaker,
576 StructurerProvenance,
577 };
578 fn txid(seed: u8) -> TxId {
579 let mut bytes = [0_u8; 12];
580 for (index, byte) in bytes.iter_mut().enumerate() {
581 *byte = seed.wrapping_add(index as u8);
582 }
583 TxId::from_bytes(bytes)
584 }
585 fn features(seed: f64, name: Option<&str>) -> FeatureVector24 {
586 FeatureVector24 {
587 median_f0_hz: Some(seed),
588 high_front_vowel_f1_hz: None,
589 dominant_rhotic_realization: name.map(str::to_owned),
590 hypernasality_0_to_4: Some(seed / 100.0),
591 ..FeatureVector24::default()
592 }
593 }
594 fn executed_analysis() -> ExecutedAnalysis {
595 let mut ogg = vec![0_u8; 29];
596 ogg[..4].copy_from_slice(b"OggS");
597 ogg[4] = 0;
598 ogg[26] = 1;
599 ogg[27] = 1;
600 ogg[28] = 0;
601 let audio = OggAudioMetadata::from_bytes(&ogg, 1_250, Some("sample.ogg".to_owned()))
602 .expect("valid Ogg metadata");
603 let analysis = StructuredAnalysis {
604 transcript: "Speaker 1: Héllo from 東京".to_owned(),
605 speakers: vec![StructuredSpeaker {
606 speaker: LocalSpeakerLabel::new(1).expect("valid label"),
607 language: "English / 日本語".to_owned(),
608 features: features(182.5, Some("approximant")),
609 features_usable_for_training: true,
610 }],
611 };
612 ExecutedAnalysis {
613 envelope: AnalysisEnvelope {
614 audio,
615 analysis,
616 gemini: GeminiCohort::new("gemini-3.1-pro"),
617 structurer: StructurerProvenance::new("terra-structurer"),
618 },
619 label_extractor: StructurerProvenance::new("terra-labeler"),
620 }
621 }
622 fn staged_fragment() -> StagedFragmentV1 {
623 StagedFragmentV1 {
624 analysis_txid: txid(20),
625 transcript: "Élodie: bonjour 🌍\n話者 2: こんにちは".to_owned(),
626 speakers: vec![
627 StagedSpeakerV1 {
628 speaker: LocalSpeakerLabel::new(1).expect("valid label"),
629 language: "français".to_owned(),
630 features: features(201.25, Some("uvulaire")),
631 usable_for_training: true,
632 },
633 StagedSpeakerV1 {
634 speaker: LocalSpeakerLabel::new(2).expect("valid label"),
635 language: "日本語".to_owned(),
636 features: FeatureVector24 {
637 low_vowel_f2_hz: Some(1_234.5),
638 dominant_lateral_realization: Some("明瞭".to_owned()),
639 ..FeatureVector24::default()
640 },
641 usable_for_training: false,
642 },
643 ],
644 }
645 }
646 fn final_fragment() -> FinalFragmentV1 {
647 FinalFragmentV1 {
648 analysis_txid: txid(30),
649 confirmation_txid: txid(40),
650 transcript: "Élodie: bonjour 🌍\n話者 2: こんにちは".to_owned(),
651 speakers: vec![
652 FinalSpeakerV1 {
653 speaker: LocalSpeakerLabel::new(1).expect("valid label"),
654 person_id: "person-éloïse".to_owned(),
655 language: "français".to_owned(),
656 features: features(201.25, Some("uvulaire")),
657 usable_for_training: false,
658 },
659 FinalSpeakerV1 {
660 speaker: LocalSpeakerLabel::new(2).expect("valid label"),
661 person_id: "人物-東京".to_owned(),
662 language: "日本語".to_owned(),
663 features: FeatureVector24::default(),
664 usable_for_training: true,
665 },
666 ],
667 }
668 }
669 fn read_test_length(bytes: &[u8], position: &mut usize) -> usize {
670 let end = *position + 8;
671 let mut value = [0_u8; 8];
672 value.copy_from_slice(&bytes[*position..end]);
673 *position = end;
674 usize::try_from(u64::from_le_bytes(value)).expect("test length fits")
675 }
676 fn first_staged_feature_and_boolean_offsets(bytes: &[u8]) -> (usize, usize) {
677 let mut position = BODY_OFFSET;
678 let transcript_length = read_test_length(bytes, &mut position);
679 position += transcript_length;
680 let speaker_count = read_test_length(bytes, &mut position);
681 assert!(speaker_count > 0);
682 let label_length = read_test_length(bytes, &mut position);
683 position += label_length;
684 let language_length = read_test_length(bytes, &mut position);
685 position += language_length;
686 let feature_length = read_test_length(bytes, &mut position);
687 let feature_offset = position;
688 position += feature_length;
689 (feature_offset, position)
690 }
691 #[test]
692 fn txid_slot_has_exact_bytes_and_padding() {
693 let id = txid(3);
694 let encoded = TxIdSlot::new(id).encode();
695 assert_eq!(encoded.len(), TxIdSlot::LEN);
696 assert_eq!(TxIdSlot::PADDING_LEN, 4);
697 assert_eq!(&encoded[..12], id.as_bytes());
698 assert_eq!(&encoded[12..], &[0_u8; 4]);
699 assert_eq!(TxIdSlot::decode(&encoded).expect("slot").txid(), id);
700 assert_eq!(
701 TxIdSlot::decode(&encoded[..15]),
702 Err(FormatError::InvalidTxIdSlotLength(15))
703 );
704 let mut corrupt = encoded;
705 corrupt[12] = 1;
706 assert_eq!(TxIdSlot::decode(&corrupt), Err(FormatError::NonZeroPadding));
707 }
708 #[test]
709 fn synthetic_slots_start_on_sixteen_byte_boundaries() {
710 let mut bytes = [0_u8; 64];
711 for (index, offset) in (0..64).step_by(TxIdSlot::LEN).enumerate() {
712 assert_eq!(offset % TxIdSlot::LEN, 0);
713 let slot = TxIdSlot::new(txid(index as u8)).encode();
714 bytes[offset..offset + TxIdSlot::LEN].copy_from_slice(&slot);
715 }
716 for (index, offset) in (0..64).step_by(TxIdSlot::LEN).enumerate() {
717 let slot =
718 TxIdSlot::decode(&bytes[offset..offset + TxIdSlot::LEN]).expect("aligned slot");
719 assert_eq!(slot.txid(), txid(index as u8));
720 }
721 }
722 #[test]
723 fn fragment_slots_are_at_fixed_offsets() {
724 let staged = staged_fragment();
725 let staged_bytes = encode_staged_fragment(&staged).expect("encode staged");
726 assert_eq!(
727 &staged_bytes[16..32],
728 &TxIdSlot::new(staged.analysis_txid).encode()
729 );
730 assert_eq!(&staged_bytes[32..48], &[0_u8; 16]);
731 let final_value = final_fragment();
732 let final_bytes = encode_final_fragment(&final_value).expect("encode final");
733 assert_eq!(
734 &final_bytes[16..32],
735 &TxIdSlot::new(final_value.analysis_txid).encode()
736 );
737 assert_eq!(
738 &final_bytes[32..48],
739 &TxIdSlot::new(final_value.confirmation_txid).encode()
740 );
741 }
742 #[test]
743 fn every_shard_character_roundtrips() {
744 for shard in 0..64 {
745 let mut bytes = [0_u8; 12];
746 bytes[0] = (shard as u8) << 2;
747 bytes[11] = shard as u8;
748 let id = TxId::from_bytes(bytes);
749 let path = txid_path(id);
750 let first = path
751 .components()
752 .next()
753 .expect("first component")
754 .as_os_str()
755 .to_str()
756 .expect("UTF-8");
757 assert_eq!(first.as_bytes(), &BASE64_ALPHABET[shard..shard + 1]);
758 assert_eq!(txid_from_path(&path), Ok(id));
759 }
760 }
761 #[test]
762 fn malformed_paths_are_rejected() {
763 for path in [
764 "",
765 "A",
766 "A/B.dat",
767 "AA/AAAAAAAAAAAAAAA.dat",
768 "A/AAAAAAAAAAAAAAA.bin",
769 "A/AAAAAAAAAAAAAA!.dat",
770 "/A/AAAAAAAAAAAAAAA.dat",
771 "A/../B.dat",
772 "A/AAAAAAAAAAAAAAA.dat/extra",
773 ] {
774 assert_eq!(txid_from_path(path), Err(FormatError::InvalidPath));
775 }
776 }
777 #[test]
778 fn every_event_variant_roundtrips_with_stable_tags() {
779 let events = vec![
780 AudioClassificationEventV1::Queue(QueueV1 {
781 audio_object_id: txid(1),
782 duration_ms: 98_765,
783 filename: Some("réunion.ogg".to_owned()),
784 }),
785 AudioClassificationEventV1::Processed(ProcessedV1 {
786 queue_id: txid(3),
787 audio_object_id: txid(4),
788 analysis: executed_analysis(),
789 }),
790 AudioClassificationEventV1::Failed(FailedV1 {
791 queue_id: txid(5),
792 audio_object_id: txid(6),
793 final_stage: AnalysisStageV1::GeminiFeatures,
794 final_error: "packet 2 unavailable".to_owned(),
795 }),
796 AudioClassificationEventV1::Discarded(DiscardedV1 {
797 failed_queue_id: txid(7),
798 audio_object_id: txid(8),
799 }),
800 AudioClassificationEventV1::Confirmed(ConfirmedV1 {
801 queue_id: txid(9),
802 audio_object_id: txid(10),
803 analysis_txid: txid(11),
804 speakers: vec![ConfirmedSpeakerV1 {
805 speaker: LocalSpeakerLabel::new(1).expect("valid label"),
806 person_id: "person-α".to_owned(),
807 }],
808 }),
809 ];
810 for (index, event) in events.into_iter().enumerate() {
811 let encoded = encode_event(&event).expect("encode event");
812 assert_eq!(encoded[0], 2);
813 assert_eq!(encoded[1], index as u8 + 1);
814 assert_eq!(decode_event(&encoded), Ok(event));
815 }
816 }
817 #[test]
818 fn every_analysis_stage_roundtrips() {
819 for stage in [
820 AnalysisStageV1::GeminiTranscript,
821 AnalysisStageV1::TerraLabels,
822 AnalysisStageV1::GeminiFeatures,
823 AnalysisStageV1::TerraStructuring,
824 ] {
825 let event = AudioClassificationEventV1::Failed(FailedV1 {
826 queue_id: txid(1),
827 audio_object_id: txid(2),
828 final_stage: stage,
829 final_error: "failure".to_owned(),
830 });
831 let encoded = encode_event(&event).expect("encode event");
832 assert_eq!(decode_event(&encoded), Ok(event));
833 }
834 }
835 #[test]
836 fn event_corruption_and_version_one_are_rejected() {
837 let event = AudioClassificationEventV1::Queue(QueueV1 {
838 audio_object_id: txid(1),
839 duration_ms: 500,
840 filename: None,
841 });
842 let encoded = encode_event(&event).expect("encode event");
843 assert_eq!(decode_event(&[]), Err(FormatError::Truncated));
844 let mut version_one = encoded.clone();
845 version_one[0] = 1;
846 assert_eq!(
847 decode_event(&version_one),
848 Err(FormatError::UnsupportedVersion(1))
849 );
850 let mut wrong_tag = encoded.clone();
851 wrong_tag[1] = 99;
852 assert_eq!(
853 decode_event(&wrong_tag),
854 Err(FormatError::UnknownEventTag(99))
855 );
856 assert_eq!(
857 decode_event(&encoded[..2]),
858 Err(FormatError::InvalidEventBody)
859 );
860 let mut trailing = encoded;
861 trailing.push(0);
862 assert_eq!(decode_event(&trailing), Err(FormatError::TrailingBytes));
863 }
864 #[test]
865 fn staged_and_final_fragments_roundtrip_utf8_speakers() {
866 let staged = staged_fragment();
867 let staged_bytes = encode_staged_fragment(&staged).expect("encode staged");
868 assert_eq!(decode_staged_fragment(&staged_bytes), Ok(staged));
869 let final_value = final_fragment();
870 let final_bytes = encode_final_fragment(&final_value).expect("encode final");
871 assert_eq!(decode_final_fragment(&final_bytes), Ok(final_value));
872 }
873 #[test]
874 fn fragment_header_and_slot_corruption_is_rejected() {
875 let staged = staged_fragment();
876 let encoded = encode_staged_fragment(&staged).expect("encode staged");
877 for cut in [0, 15, 47, encoded.len() - 1] {
878 assert!(decode_staged_fragment(&encoded[..cut]).is_err());
879 }
880 let mut wrong_version = encoded.clone();
881 wrong_version[0] = 2;
882 assert_eq!(
883 decode_staged_fragment(&wrong_version),
884 Err(FormatError::UnsupportedVersion(2))
885 );
886 let mut wrong_kind = encoded.clone();
887 wrong_kind[1] = FINAL_KIND;
888 assert_eq!(
889 decode_staged_fragment(&wrong_kind),
890 Err(FormatError::InvalidFragmentKind(FINAL_KIND))
891 );
892 let mut reserved = encoded.clone();
893 reserved[2] = 1;
894 assert_eq!(
895 decode_staged_fragment(&reserved),
896 Err(FormatError::NonZeroReserved)
897 );
898 let mut analysis_padding = encoded.clone();
899 analysis_padding[28] = 1;
900 assert_eq!(
901 decode_staged_fragment(&analysis_padding),
902 Err(FormatError::NonZeroPadding)
903 );
904 let mut staged_confirmation = encoded;
905 staged_confirmation[32] = 1;
906 assert_eq!(
907 decode_staged_fragment(&staged_confirmation),
908 Err(FormatError::NonZeroStagedConfirmation)
909 );
910 let final_value = final_fragment();
911 let mut final_bytes = encode_final_fragment(&final_value).expect("encode final");
912 final_bytes[44] = 1;
913 assert_eq!(
914 decode_final_fragment(&final_bytes),
915 Err(FormatError::NonZeroPadding)
916 );
917 }
918 #[test]
919 fn fragment_body_corruption_is_rejected() {
920 let value = StagedFragmentV1 {
921 analysis_txid: txid(1),
922 transcript: "hello".to_owned(),
923 speakers: vec![StagedSpeakerV1 {
924 speaker: LocalSpeakerLabel::new(1).expect("valid label"),
925 language: "English".to_owned(),
926 features: FeatureVector24::default(),
927 usable_for_training: true,
928 }],
929 };
930 let encoded = encode_staged_fragment(&value).expect("encode staged");
931 let mut length_overflow = encoded.clone();
932 length_overflow[BODY_OFFSET..BODY_OFFSET + 8].copy_from_slice(&u64::MAX.to_le_bytes());
933 assert_eq!(
934 decode_staged_fragment(&length_overflow),
935 Err(FormatError::LengthOverflow)
936 );
937 let mut invalid_utf8 = encoded.clone();
938 invalid_utf8[BODY_OFFSET + 8] = 0xff;
939 assert_eq!(
940 decode_staged_fragment(&invalid_utf8),
941 Err(FormatError::InvalidUtf8)
942 );
943 let (feature_offset, boolean_offset) = first_staged_feature_and_boolean_offsets(&encoded);
944 let mut invalid_feature = encoded.clone();
945 invalid_feature[feature_offset] = 2;
946 assert_eq!(
947 decode_staged_fragment(&invalid_feature),
948 Err(FormatError::InvalidFeatureBody)
949 );
950 let mut invalid_boolean = encoded.clone();
951 invalid_boolean[boolean_offset] = 2;
952 assert_eq!(
953 decode_staged_fragment(&invalid_boolean),
954 Err(FormatError::InvalidBoolean(2))
955 );
956 let mut trailing = encoded;
957 trailing.push(0);
958 assert_eq!(
959 decode_staged_fragment(&trailing),
960 Err(FormatError::TrailingBytes)
961 );
962 }
963 #[test]
964 fn malformed_label_and_internal_feature_trailing_bytes_are_rejected() {
965 let value = StagedFragmentV1 {
966 analysis_txid: txid(1),
967 transcript: "hello".to_owned(),
968 speakers: vec![StagedSpeakerV1 {
969 speaker: LocalSpeakerLabel::new(1).expect("valid label"),
970 language: "English".to_owned(),
971 features: FeatureVector24::default(),
972 usable_for_training: true,
973 }],
974 };
975 let encoded = encode_staged_fragment(&value).expect("encode staged");
976 let mut position = BODY_OFFSET;
977 let transcript_length = read_test_length(&encoded, &mut position);
978 position += transcript_length;
979 let _ = read_test_length(&encoded, &mut position);
980 let label_length = read_test_length(&encoded, &mut position);
981 assert_eq!(label_length, "Speaker 1".len());
982 let mut invalid_label = encoded.clone();
983 invalid_label[position] = b'X';
984 assert_eq!(
985 decode_staged_fragment(&invalid_label),
986 Err(FormatError::InvalidSpeakerLabel)
987 );
988 let (feature_offset, boolean_offset) = first_staged_feature_and_boolean_offsets(&encoded);
989 let mut feature_trailing = encoded;
990 feature_trailing.insert(boolean_offset, 0);
991 let feature_length_offset = feature_offset - 8;
992 let feature_length = boolean_offset - feature_offset + 1;
993 feature_trailing[feature_length_offset..feature_length_offset + 8]
994 .copy_from_slice(&(feature_length as u64).to_le_bytes());
995 assert_eq!(
996 decode_staged_fragment(&feature_trailing),
997 Err(FormatError::InvalidFeatureBody)
998 );
999 }
1000}