use serde::{Deserialize, Serialize};
use std::fmt::{Display, Formatter};
use std::path::{Component, Path, PathBuf};
pub use kcode_k1_transaction_id::TxId;
pub use kcode_speaker_v3_analysis::{ExecutedAnalysis, FeatureVector24, LocalSpeakerLabel};
const EVENT_VERSION: u8 = 2;
const QUEUE_TAG: u8 = 1;
const PROCESSED_TAG: u8 = 2;
const FAILED_TAG: u8 = 3;
const DISCARDED_TAG: u8 = 4;
const CONFIRMED_TAG: u8 = 5;
const FRAGMENT_VERSION: u8 = 1;
const STAGED_KIND: u8 = 1;
const FINAL_KIND: u8 = 2;
const HEADER_LEN: usize = 16;
const ANALYSIS_SLOT_OFFSET: usize = 16;
const CONFIRMATION_SLOT_OFFSET: usize = 32;
const BODY_OFFSET: usize = 48;
const BASE64_ALPHABET: &[u8; 64] =
b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_";
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum AnalysisStageV1 {
GeminiTranscript,
TerraLabels,
GeminiFeatures,
TerraStructuring,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct QueueV1 {
#[serde(with = "txid_serde")]
pub audio_object_id: TxId,
pub duration_ms: u64,
pub filename: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ProcessedV1 {
#[serde(with = "txid_serde")]
pub queue_id: TxId,
#[serde(with = "txid_serde")]
pub audio_object_id: TxId,
pub analysis: ExecutedAnalysis,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FailedV1 {
#[serde(with = "txid_serde")]
pub queue_id: TxId,
#[serde(with = "txid_serde")]
pub audio_object_id: TxId,
pub final_stage: AnalysisStageV1,
pub final_error: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DiscardedV1 {
#[serde(with = "txid_serde")]
pub failed_queue_id: TxId,
#[serde(with = "txid_serde")]
pub audio_object_id: TxId,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ConfirmedSpeakerV1 {
pub speaker: LocalSpeakerLabel,
pub person_id: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ConfirmedV1 {
#[serde(with = "txid_serde")]
pub queue_id: TxId,
#[serde(with = "txid_serde")]
pub audio_object_id: TxId,
#[serde(with = "txid_serde")]
pub analysis_txid: TxId,
pub speakers: Vec<ConfirmedSpeakerV1>,
}
#[allow(clippy::large_enum_variant)]
#[derive(Debug, Clone, PartialEq)]
pub enum AudioClassificationEventV1 {
Queue(QueueV1),
Processed(ProcessedV1),
Failed(FailedV1),
Discarded(DiscardedV1),
Confirmed(ConfirmedV1),
}
#[derive(Debug, Clone, PartialEq)]
pub struct StagedSpeakerV1 {
pub speaker: LocalSpeakerLabel,
pub language: String,
pub features: FeatureVector24,
pub usable_for_training: bool,
}
#[derive(Debug, Clone, PartialEq)]
pub struct StagedFragmentV1 {
pub analysis_txid: TxId,
pub transcript: String,
pub speakers: Vec<StagedSpeakerV1>,
}
#[derive(Debug, Clone, PartialEq)]
pub struct FinalSpeakerV1 {
pub speaker: LocalSpeakerLabel,
pub person_id: String,
pub language: String,
pub features: FeatureVector24,
pub usable_for_training: bool,
}
#[derive(Debug, Clone, PartialEq)]
pub struct FinalFragmentV1 {
pub analysis_txid: TxId,
pub confirmation_txid: TxId,
pub transcript: String,
pub speakers: Vec<FinalSpeakerV1>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum FormatError {
Truncated,
LengthOverflow,
UnsupportedVersion(u8),
UnknownEventTag(u8),
InvalidEventBody,
InvalidFragmentKind(u8),
NonZeroReserved,
NonZeroPadding,
NonZeroStagedConfirmation,
InvalidTxIdSlotLength(usize),
InvalidUtf8,
InvalidSpeakerLabel,
InvalidFeatureBody,
InvalidBoolean(u8),
TrailingBytes,
InvalidPath,
}
impl Display for FormatError {
fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
match self {
Self::Truncated => formatter.write_str("truncated input"),
Self::LengthOverflow => formatter.write_str("encoded length overflow"),
Self::UnsupportedVersion(version) => {
write!(formatter, "unsupported format version {version}")
}
Self::UnknownEventTag(tag) => write!(formatter, "unknown event tag {tag}"),
Self::InvalidEventBody => formatter.write_str("invalid event body"),
Self::InvalidFragmentKind(kind) => write!(formatter, "invalid fragment kind {kind}"),
Self::NonZeroReserved => formatter.write_str("nonzero reserved bytes"),
Self::NonZeroPadding => formatter.write_str("nonzero transaction ID slot padding"),
Self::NonZeroStagedConfirmation => {
formatter.write_str("nonzero staged confirmation slot")
}
Self::InvalidTxIdSlotLength(length) => {
write!(formatter, "invalid transaction ID slot length {length}")
}
Self::InvalidUtf8 => formatter.write_str("invalid UTF-8"),
Self::InvalidSpeakerLabel => formatter.write_str("invalid speaker label"),
Self::InvalidFeatureBody => formatter.write_str("invalid feature body"),
Self::InvalidBoolean(value) => write!(formatter, "invalid boolean byte {value}"),
Self::TrailingBytes => formatter.write_str("trailing bytes"),
Self::InvalidPath => formatter.write_str("invalid transaction ID path"),
}
}
}
impl std::error::Error for FormatError {}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct TxIdSlot {
txid: TxId,
}
impl TxIdSlot {
pub const LEN: usize = 16;
pub const PADDING_LEN: usize = 4;
pub const fn new(txid: TxId) -> Self {
Self { txid }
}
pub const fn txid(self) -> TxId {
self.txid
}
pub fn encode(self) -> [u8; Self::LEN] {
let mut encoded = [0_u8; Self::LEN];
encoded[..12].copy_from_slice(self.txid.as_bytes());
encoded
}
pub fn decode(bytes: &[u8]) -> Result<Self, FormatError> {
if bytes.len() != Self::LEN {
return Err(FormatError::InvalidTxIdSlotLength(bytes.len()));
}
if bytes[12..].iter().any(|byte| *byte != 0) {
return Err(FormatError::NonZeroPadding);
}
let mut txid_bytes = [0_u8; 12];
txid_bytes.copy_from_slice(&bytes[..12]);
Ok(Self::new(TxId::from_bytes(txid_bytes)))
}
}
pub fn encode_event(event: &AudioClassificationEventV1) -> Result<Vec<u8>, FormatError> {
let (tag, body) = match event {
AudioClassificationEventV1::Queue(value) => (QUEUE_TAG, postcard::to_allocvec(value)),
AudioClassificationEventV1::Processed(value) => {
(PROCESSED_TAG, postcard::to_allocvec(value))
}
AudioClassificationEventV1::Failed(value) => (FAILED_TAG, postcard::to_allocvec(value)),
AudioClassificationEventV1::Discarded(value) => {
(DISCARDED_TAG, postcard::to_allocvec(value))
}
AudioClassificationEventV1::Confirmed(value) => {
(CONFIRMED_TAG, postcard::to_allocvec(value))
}
};
let body = body.map_err(|_| FormatError::InvalidEventBody)?;
let mut encoded = Vec::with_capacity(2 + body.len());
encoded.push(EVENT_VERSION);
encoded.push(tag);
encoded.extend_from_slice(&body);
Ok(encoded)
}
pub fn decode_event(bytes: &[u8]) -> Result<AudioClassificationEventV1, FormatError> {
if bytes.len() < 2 {
return Err(FormatError::Truncated);
}
if bytes[0] != EVENT_VERSION {
return Err(FormatError::UnsupportedVersion(bytes[0]));
}
let body = &bytes[2..];
match bytes[1] {
QUEUE_TAG => decode_event_body(body).map(AudioClassificationEventV1::Queue),
PROCESSED_TAG => decode_event_body(body).map(AudioClassificationEventV1::Processed),
FAILED_TAG => decode_event_body(body).map(AudioClassificationEventV1::Failed),
DISCARDED_TAG => decode_event_body(body).map(AudioClassificationEventV1::Discarded),
CONFIRMED_TAG => decode_event_body(body).map(AudioClassificationEventV1::Confirmed),
tag => Err(FormatError::UnknownEventTag(tag)),
}
}
pub fn encode_staged_fragment(value: &StagedFragmentV1) -> Result<Vec<u8>, FormatError> {
let mut encoded = encode_fragment_prefix(STAGED_KIND, value.analysis_txid, None);
append_string(&mut encoded, &value.transcript)?;
append_length(&mut encoded, value.speakers.len())?;
for speaker in &value.speakers {
append_string(&mut encoded, &speaker.speaker.to_string())?;
append_string(&mut encoded, &speaker.language)?;
append_features(&mut encoded, &speaker.features)?;
encoded.push(u8::from(speaker.usable_for_training));
}
Ok(encoded)
}
pub fn decode_staged_fragment(bytes: &[u8]) -> Result<StagedFragmentV1, FormatError> {
let (analysis_txid, _) = decode_fragment_header(bytes, STAGED_KIND, false)?;
let mut decoder = BodyDecoder::new(&bytes[BODY_OFFSET..]);
let transcript = decoder.read_string()?;
let speaker_count = decoder.read_count()?;
let mut speakers = Vec::new();
for _ in 0..speaker_count {
speakers.push(StagedSpeakerV1 {
speaker: decoder.read_speaker_label()?,
language: decoder.read_string()?,
features: decoder.read_features()?,
usable_for_training: decoder.read_boolean()?,
});
}
decoder.finish()?;
Ok(StagedFragmentV1 {
analysis_txid,
transcript,
speakers,
})
}
pub fn encode_final_fragment(value: &FinalFragmentV1) -> Result<Vec<u8>, FormatError> {
let mut encoded = encode_fragment_prefix(
FINAL_KIND,
value.analysis_txid,
Some(value.confirmation_txid),
);
append_string(&mut encoded, &value.transcript)?;
append_length(&mut encoded, value.speakers.len())?;
for speaker in &value.speakers {
append_string(&mut encoded, &speaker.speaker.to_string())?;
append_string(&mut encoded, &speaker.person_id)?;
append_string(&mut encoded, &speaker.language)?;
append_features(&mut encoded, &speaker.features)?;
encoded.push(u8::from(speaker.usable_for_training));
}
Ok(encoded)
}
pub fn decode_final_fragment(bytes: &[u8]) -> Result<FinalFragmentV1, FormatError> {
let (analysis_txid, confirmation_txid) = decode_fragment_header(bytes, FINAL_KIND, true)?;
let mut decoder = BodyDecoder::new(&bytes[BODY_OFFSET..]);
let transcript = decoder.read_string()?;
let speaker_count = decoder.read_count()?;
let mut speakers = Vec::new();
for _ in 0..speaker_count {
speakers.push(FinalSpeakerV1 {
speaker: decoder.read_speaker_label()?,
person_id: decoder.read_string()?,
language: decoder.read_string()?,
features: decoder.read_features()?,
usable_for_training: decoder.read_boolean()?,
});
}
decoder.finish()?;
Ok(FinalFragmentV1 {
analysis_txid,
confirmation_txid: confirmation_txid.expect("final header has a confirmation slot"),
transcript,
speakers,
})
}
pub fn txid_path(txid: TxId) -> PathBuf {
let encoded = encode_txid_base64(txid);
let first = String::from_utf8(encoded[..1].to_vec()).expect("base64 is ASCII");
let remaining = String::from_utf8(encoded[1..].to_vec()).expect("base64 is ASCII");
PathBuf::from(first).join(format!("{remaining}.dat"))
}
pub fn txid_from_path(path: impl AsRef<Path>) -> Result<TxId, FormatError> {
let mut components = path.as_ref().components();
let first = normal_utf8_component(components.next())?;
let filename = normal_utf8_component(components.next())?;
if components.next().is_some() || first.len() != 1 {
return Err(FormatError::InvalidPath);
}
let remaining = filename
.strip_suffix(".dat")
.ok_or(FormatError::InvalidPath)?;
if remaining.len() != 15 {
return Err(FormatError::InvalidPath);
}
let mut encoded = [0_u8; 16];
encoded[0] = first.as_bytes()[0];
encoded[1..].copy_from_slice(remaining.as_bytes());
decode_txid_base64(encoded)
}
fn decode_event_body<T>(bytes: &[u8]) -> Result<T, FormatError>
where
T: for<'de> Deserialize<'de>,
{
let (value, remaining) =
postcard::take_from_bytes(bytes).map_err(|_| FormatError::InvalidEventBody)?;
if !remaining.is_empty() {
return Err(FormatError::TrailingBytes);
}
Ok(value)
}
fn encode_fragment_prefix(kind: u8, analysis: TxId, confirmation: Option<TxId>) -> Vec<u8> {
let mut encoded = vec![0_u8; BODY_OFFSET];
encoded[0] = FRAGMENT_VERSION;
encoded[1] = kind;
encoded[ANALYSIS_SLOT_OFFSET..ANALYSIS_SLOT_OFFSET + TxIdSlot::LEN]
.copy_from_slice(&TxIdSlot::new(analysis).encode());
if let Some(confirmation) = confirmation {
encoded[CONFIRMATION_SLOT_OFFSET..CONFIRMATION_SLOT_OFFSET + TxIdSlot::LEN]
.copy_from_slice(&TxIdSlot::new(confirmation).encode());
}
encoded
}
fn decode_fragment_header(
bytes: &[u8],
expected_kind: u8,
needs_confirmation: bool,
) -> Result<(TxId, Option<TxId>), FormatError> {
if bytes.len() < BODY_OFFSET {
return Err(FormatError::Truncated);
}
if bytes[0] != FRAGMENT_VERSION {
return Err(FormatError::UnsupportedVersion(bytes[0]));
}
if bytes[1] != expected_kind {
return Err(FormatError::InvalidFragmentKind(bytes[1]));
}
if bytes[2..HEADER_LEN].iter().any(|byte| *byte != 0) {
return Err(FormatError::NonZeroReserved);
}
let analysis =
TxIdSlot::decode(&bytes[ANALYSIS_SLOT_OFFSET..ANALYSIS_SLOT_OFFSET + TxIdSlot::LEN])?
.txid();
let confirmation_bytes =
&bytes[CONFIRMATION_SLOT_OFFSET..CONFIRMATION_SLOT_OFFSET + TxIdSlot::LEN];
let confirmation = if needs_confirmation {
Some(TxIdSlot::decode(confirmation_bytes)?.txid())
} else {
if confirmation_bytes.iter().any(|byte| *byte != 0) {
return Err(FormatError::NonZeroStagedConfirmation);
}
None
};
Ok((analysis, confirmation))
}
fn append_length(encoded: &mut Vec<u8>, length: usize) -> Result<(), FormatError> {
encoded.extend_from_slice(
&u64::try_from(length)
.map_err(|_| FormatError::LengthOverflow)?
.to_le_bytes(),
);
Ok(())
}
fn append_bytes(encoded: &mut Vec<u8>, bytes: &[u8]) -> Result<(), FormatError> {
append_length(encoded, bytes.len())?;
encoded.extend_from_slice(bytes);
Ok(())
}
fn append_string(encoded: &mut Vec<u8>, value: &str) -> Result<(), FormatError> {
append_bytes(encoded, value.as_bytes())
}
fn append_features(encoded: &mut Vec<u8>, features: &FeatureVector24) -> Result<(), FormatError> {
append_bytes(
encoded,
&postcard::to_allocvec(features).map_err(|_| FormatError::InvalidFeatureBody)?,
)
}
struct BodyDecoder<'a> {
bytes: &'a [u8],
position: usize,
}
impl<'a> BodyDecoder<'a> {
fn new(bytes: &'a [u8]) -> Self {
Self { bytes, position: 0 }
}
fn take(&mut self, length: usize) -> Result<&'a [u8], FormatError> {
let end = self
.position
.checked_add(length)
.ok_or(FormatError::LengthOverflow)?;
let value = self
.bytes
.get(self.position..end)
.ok_or(FormatError::Truncated)?;
self.position = end;
Ok(value)
}
fn read_u64(&mut self) -> Result<u64, FormatError> {
let bytes = self.take(8)?;
let mut value = [0_u8; 8];
value.copy_from_slice(bytes);
Ok(u64::from_le_bytes(value))
}
fn read_length(&mut self) -> Result<usize, FormatError> {
usize::try_from(self.read_u64()?).map_err(|_| FormatError::LengthOverflow)
}
fn read_count(&mut self) -> Result<usize, FormatError> {
self.read_length()
}
fn read_bytes(&mut self) -> Result<&'a [u8], FormatError> {
let length = self.read_length()?;
self.take(length)
}
fn read_string(&mut self) -> Result<String, FormatError> {
Ok(std::str::from_utf8(self.read_bytes()?)
.map_err(|_| FormatError::InvalidUtf8)?
.to_owned())
}
fn read_speaker_label(&mut self) -> Result<LocalSpeakerLabel, FormatError> {
self.read_string()?
.parse()
.map_err(|_| FormatError::InvalidSpeakerLabel)
}
fn read_features(&mut self) -> Result<FeatureVector24, FormatError> {
let bytes = self.read_bytes()?;
let (features, remaining) =
postcard::take_from_bytes(bytes).map_err(|_| FormatError::InvalidFeatureBody)?;
if !remaining.is_empty() {
return Err(FormatError::InvalidFeatureBody);
}
Ok(features)
}
fn read_boolean(&mut self) -> Result<bool, FormatError> {
match self.take(1)?[0] {
0 => Ok(false),
1 => Ok(true),
value => Err(FormatError::InvalidBoolean(value)),
}
}
fn finish(self) -> Result<(), FormatError> {
if self.position == self.bytes.len() {
Ok(())
} else {
Err(FormatError::TrailingBytes)
}
}
}
fn normal_utf8_component(component: Option<Component<'_>>) -> Result<&str, FormatError> {
match component {
Some(Component::Normal(value)) => value.to_str().ok_or(FormatError::InvalidPath),
_ => Err(FormatError::InvalidPath),
}
}
fn encode_txid_base64(txid: TxId) -> [u8; 16] {
let bytes = txid.into_bytes();
let mut encoded = [0_u8; 16];
for group in 0..4 {
let input = group * 3;
let output = group * 4;
encoded[output] = BASE64_ALPHABET[(bytes[input] >> 2) as usize];
encoded[output + 1] =
BASE64_ALPHABET[(((bytes[input] & 3) << 4) | (bytes[input + 1] >> 4)) as usize];
encoded[output + 2] =
BASE64_ALPHABET[(((bytes[input + 1] & 15) << 2) | (bytes[input + 2] >> 6)) as usize];
encoded[output + 3] = BASE64_ALPHABET[(bytes[input + 2] & 63) as usize];
}
encoded
}
fn decode_txid_base64(encoded: [u8; 16]) -> Result<TxId, FormatError> {
let mut bytes = [0_u8; 12];
for group in 0..4 {
let input = group * 4;
let output = group * 3;
let first = decode_base64_character(encoded[input])?;
let second = decode_base64_character(encoded[input + 1])?;
let third = decode_base64_character(encoded[input + 2])?;
let fourth = decode_base64_character(encoded[input + 3])?;
bytes[output] = (first << 2) | (second >> 4);
bytes[output + 1] = (second << 4) | (third >> 2);
bytes[output + 2] = (third << 6) | fourth;
}
Ok(TxId::from_bytes(bytes))
}
fn decode_base64_character(value: u8) -> Result<u8, FormatError> {
match value {
b'A'..=b'Z' => Ok(value - b'A'),
b'a'..=b'z' => Ok(value - b'a' + 26),
b'0'..=b'9' => Ok(value - b'0' + 52),
b'-' => Ok(62),
b'_' => Ok(63),
_ => Err(FormatError::InvalidPath),
}
}
mod txid_serde {
use super::TxId;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
pub fn serialize<S>(value: &TxId, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
value.into_bytes().serialize(serializer)
}
pub fn deserialize<'de, D>(deserializer: D) -> Result<TxId, D::Error>
where
D: Deserializer<'de>,
{
Ok(TxId::from_bytes(<[u8; 12]>::deserialize(deserializer)?))
}
}
#[cfg(test)]
mod tests {
use super::*;
use kcode_speaker_v3_analysis::{
AnalysisEnvelope, GeminiCohort, OggAudioMetadata, StructuredAnalysis, StructuredSpeaker,
StructurerProvenance,
};
fn txid(seed: u8) -> TxId {
let mut bytes = [0_u8; 12];
for (index, byte) in bytes.iter_mut().enumerate() {
*byte = seed.wrapping_add(index as u8);
}
TxId::from_bytes(bytes)
}
fn features(seed: f64, name: Option<&str>) -> FeatureVector24 {
FeatureVector24 {
median_f0_hz: Some(seed),
high_front_vowel_f1_hz: None,
dominant_rhotic_realization: name.map(str::to_owned),
hypernasality_0_to_4: Some(seed / 100.0),
..FeatureVector24::default()
}
}
fn executed_analysis() -> ExecutedAnalysis {
let mut ogg = vec![0_u8; 29];
ogg[..4].copy_from_slice(b"OggS");
ogg[4] = 0;
ogg[26] = 1;
ogg[27] = 1;
ogg[28] = 0;
let audio = OggAudioMetadata::from_bytes(&ogg, 1_250, Some("sample.ogg".to_owned()))
.expect("valid Ogg metadata");
let analysis = StructuredAnalysis {
transcript: "Speaker 1: Héllo from 東京".to_owned(),
speakers: vec![StructuredSpeaker {
speaker: LocalSpeakerLabel::new(1).expect("valid label"),
language: "English / 日本語".to_owned(),
features: features(182.5, Some("approximant")),
features_usable_for_training: true,
}],
};
ExecutedAnalysis {
envelope: AnalysisEnvelope {
audio,
analysis,
gemini: GeminiCohort::new("gemini-3.1-pro"),
structurer: StructurerProvenance::new("terra-structurer"),
},
label_extractor: StructurerProvenance::new("terra-labeler"),
}
}
fn staged_fragment() -> StagedFragmentV1 {
StagedFragmentV1 {
analysis_txid: txid(20),
transcript: "Élodie: bonjour 🌍\n話者 2: こんにちは".to_owned(),
speakers: vec![
StagedSpeakerV1 {
speaker: LocalSpeakerLabel::new(1).expect("valid label"),
language: "français".to_owned(),
features: features(201.25, Some("uvulaire")),
usable_for_training: true,
},
StagedSpeakerV1 {
speaker: LocalSpeakerLabel::new(2).expect("valid label"),
language: "日本語".to_owned(),
features: FeatureVector24 {
low_vowel_f2_hz: Some(1_234.5),
dominant_lateral_realization: Some("明瞭".to_owned()),
..FeatureVector24::default()
},
usable_for_training: false,
},
],
}
}
fn final_fragment() -> FinalFragmentV1 {
FinalFragmentV1 {
analysis_txid: txid(30),
confirmation_txid: txid(40),
transcript: "Élodie: bonjour 🌍\n話者 2: こんにちは".to_owned(),
speakers: vec![
FinalSpeakerV1 {
speaker: LocalSpeakerLabel::new(1).expect("valid label"),
person_id: "person-éloïse".to_owned(),
language: "français".to_owned(),
features: features(201.25, Some("uvulaire")),
usable_for_training: false,
},
FinalSpeakerV1 {
speaker: LocalSpeakerLabel::new(2).expect("valid label"),
person_id: "人物-東京".to_owned(),
language: "日本語".to_owned(),
features: FeatureVector24::default(),
usable_for_training: true,
},
],
}
}
fn read_test_length(bytes: &[u8], position: &mut usize) -> usize {
let end = *position + 8;
let mut value = [0_u8; 8];
value.copy_from_slice(&bytes[*position..end]);
*position = end;
usize::try_from(u64::from_le_bytes(value)).expect("test length fits")
}
fn first_staged_feature_and_boolean_offsets(bytes: &[u8]) -> (usize, usize) {
let mut position = BODY_OFFSET;
let transcript_length = read_test_length(bytes, &mut position);
position += transcript_length;
let speaker_count = read_test_length(bytes, &mut position);
assert!(speaker_count > 0);
let label_length = read_test_length(bytes, &mut position);
position += label_length;
let language_length = read_test_length(bytes, &mut position);
position += language_length;
let feature_length = read_test_length(bytes, &mut position);
let feature_offset = position;
position += feature_length;
(feature_offset, position)
}
#[test]
fn txid_slot_has_exact_bytes_and_padding() {
let id = txid(3);
let encoded = TxIdSlot::new(id).encode();
assert_eq!(encoded.len(), TxIdSlot::LEN);
assert_eq!(TxIdSlot::PADDING_LEN, 4);
assert_eq!(&encoded[..12], id.as_bytes());
assert_eq!(&encoded[12..], &[0_u8; 4]);
assert_eq!(TxIdSlot::decode(&encoded).expect("slot").txid(), id);
assert_eq!(
TxIdSlot::decode(&encoded[..15]),
Err(FormatError::InvalidTxIdSlotLength(15))
);
let mut corrupt = encoded;
corrupt[12] = 1;
assert_eq!(TxIdSlot::decode(&corrupt), Err(FormatError::NonZeroPadding));
}
#[test]
fn synthetic_slots_start_on_sixteen_byte_boundaries() {
let mut bytes = [0_u8; 64];
for (index, offset) in (0..64).step_by(TxIdSlot::LEN).enumerate() {
assert_eq!(offset % TxIdSlot::LEN, 0);
let slot = TxIdSlot::new(txid(index as u8)).encode();
bytes[offset..offset + TxIdSlot::LEN].copy_from_slice(&slot);
}
for (index, offset) in (0..64).step_by(TxIdSlot::LEN).enumerate() {
let slot =
TxIdSlot::decode(&bytes[offset..offset + TxIdSlot::LEN]).expect("aligned slot");
assert_eq!(slot.txid(), txid(index as u8));
}
}
#[test]
fn fragment_slots_are_at_fixed_offsets() {
let staged = staged_fragment();
let staged_bytes = encode_staged_fragment(&staged).expect("encode staged");
assert_eq!(
&staged_bytes[16..32],
&TxIdSlot::new(staged.analysis_txid).encode()
);
assert_eq!(&staged_bytes[32..48], &[0_u8; 16]);
let final_value = final_fragment();
let final_bytes = encode_final_fragment(&final_value).expect("encode final");
assert_eq!(
&final_bytes[16..32],
&TxIdSlot::new(final_value.analysis_txid).encode()
);
assert_eq!(
&final_bytes[32..48],
&TxIdSlot::new(final_value.confirmation_txid).encode()
);
}
#[test]
fn every_shard_character_roundtrips() {
for shard in 0..64 {
let mut bytes = [0_u8; 12];
bytes[0] = (shard as u8) << 2;
bytes[11] = shard as u8;
let id = TxId::from_bytes(bytes);
let path = txid_path(id);
let first = path
.components()
.next()
.expect("first component")
.as_os_str()
.to_str()
.expect("UTF-8");
assert_eq!(first.as_bytes(), &BASE64_ALPHABET[shard..shard + 1]);
assert_eq!(txid_from_path(&path), Ok(id));
}
}
#[test]
fn malformed_paths_are_rejected() {
for path in [
"",
"A",
"A/B.dat",
"AA/AAAAAAAAAAAAAAA.dat",
"A/AAAAAAAAAAAAAAA.bin",
"A/AAAAAAAAAAAAAA!.dat",
"/A/AAAAAAAAAAAAAAA.dat",
"A/../B.dat",
"A/AAAAAAAAAAAAAAA.dat/extra",
] {
assert_eq!(txid_from_path(path), Err(FormatError::InvalidPath));
}
}
#[test]
fn every_event_variant_roundtrips_with_stable_tags() {
let events = vec![
AudioClassificationEventV1::Queue(QueueV1 {
audio_object_id: txid(1),
duration_ms: 98_765,
filename: Some("réunion.ogg".to_owned()),
}),
AudioClassificationEventV1::Processed(ProcessedV1 {
queue_id: txid(3),
audio_object_id: txid(4),
analysis: executed_analysis(),
}),
AudioClassificationEventV1::Failed(FailedV1 {
queue_id: txid(5),
audio_object_id: txid(6),
final_stage: AnalysisStageV1::GeminiFeatures,
final_error: "packet 2 unavailable".to_owned(),
}),
AudioClassificationEventV1::Discarded(DiscardedV1 {
failed_queue_id: txid(7),
audio_object_id: txid(8),
}),
AudioClassificationEventV1::Confirmed(ConfirmedV1 {
queue_id: txid(9),
audio_object_id: txid(10),
analysis_txid: txid(11),
speakers: vec![ConfirmedSpeakerV1 {
speaker: LocalSpeakerLabel::new(1).expect("valid label"),
person_id: "person-α".to_owned(),
}],
}),
];
for (index, event) in events.into_iter().enumerate() {
let encoded = encode_event(&event).expect("encode event");
assert_eq!(encoded[0], 2);
assert_eq!(encoded[1], index as u8 + 1);
assert_eq!(decode_event(&encoded), Ok(event));
}
}
#[test]
fn every_analysis_stage_roundtrips() {
for stage in [
AnalysisStageV1::GeminiTranscript,
AnalysisStageV1::TerraLabels,
AnalysisStageV1::GeminiFeatures,
AnalysisStageV1::TerraStructuring,
] {
let event = AudioClassificationEventV1::Failed(FailedV1 {
queue_id: txid(1),
audio_object_id: txid(2),
final_stage: stage,
final_error: "failure".to_owned(),
});
let encoded = encode_event(&event).expect("encode event");
assert_eq!(decode_event(&encoded), Ok(event));
}
}
#[test]
fn event_corruption_and_version_one_are_rejected() {
let event = AudioClassificationEventV1::Queue(QueueV1 {
audio_object_id: txid(1),
duration_ms: 500,
filename: None,
});
let encoded = encode_event(&event).expect("encode event");
assert_eq!(decode_event(&[]), Err(FormatError::Truncated));
let mut version_one = encoded.clone();
version_one[0] = 1;
assert_eq!(
decode_event(&version_one),
Err(FormatError::UnsupportedVersion(1))
);
let mut wrong_tag = encoded.clone();
wrong_tag[1] = 99;
assert_eq!(
decode_event(&wrong_tag),
Err(FormatError::UnknownEventTag(99))
);
assert_eq!(
decode_event(&encoded[..2]),
Err(FormatError::InvalidEventBody)
);
let mut trailing = encoded;
trailing.push(0);
assert_eq!(decode_event(&trailing), Err(FormatError::TrailingBytes));
}
#[test]
fn staged_and_final_fragments_roundtrip_utf8_speakers() {
let staged = staged_fragment();
let staged_bytes = encode_staged_fragment(&staged).expect("encode staged");
assert_eq!(decode_staged_fragment(&staged_bytes), Ok(staged));
let final_value = final_fragment();
let final_bytes = encode_final_fragment(&final_value).expect("encode final");
assert_eq!(decode_final_fragment(&final_bytes), Ok(final_value));
}
#[test]
fn fragment_header_and_slot_corruption_is_rejected() {
let staged = staged_fragment();
let encoded = encode_staged_fragment(&staged).expect("encode staged");
for cut in [0, 15, 47, encoded.len() - 1] {
assert!(decode_staged_fragment(&encoded[..cut]).is_err());
}
let mut wrong_version = encoded.clone();
wrong_version[0] = 2;
assert_eq!(
decode_staged_fragment(&wrong_version),
Err(FormatError::UnsupportedVersion(2))
);
let mut wrong_kind = encoded.clone();
wrong_kind[1] = FINAL_KIND;
assert_eq!(
decode_staged_fragment(&wrong_kind),
Err(FormatError::InvalidFragmentKind(FINAL_KIND))
);
let mut reserved = encoded.clone();
reserved[2] = 1;
assert_eq!(
decode_staged_fragment(&reserved),
Err(FormatError::NonZeroReserved)
);
let mut analysis_padding = encoded.clone();
analysis_padding[28] = 1;
assert_eq!(
decode_staged_fragment(&analysis_padding),
Err(FormatError::NonZeroPadding)
);
let mut staged_confirmation = encoded;
staged_confirmation[32] = 1;
assert_eq!(
decode_staged_fragment(&staged_confirmation),
Err(FormatError::NonZeroStagedConfirmation)
);
let final_value = final_fragment();
let mut final_bytes = encode_final_fragment(&final_value).expect("encode final");
final_bytes[44] = 1;
assert_eq!(
decode_final_fragment(&final_bytes),
Err(FormatError::NonZeroPadding)
);
}
#[test]
fn fragment_body_corruption_is_rejected() {
let value = StagedFragmentV1 {
analysis_txid: txid(1),
transcript: "hello".to_owned(),
speakers: vec![StagedSpeakerV1 {
speaker: LocalSpeakerLabel::new(1).expect("valid label"),
language: "English".to_owned(),
features: FeatureVector24::default(),
usable_for_training: true,
}],
};
let encoded = encode_staged_fragment(&value).expect("encode staged");
let mut length_overflow = encoded.clone();
length_overflow[BODY_OFFSET..BODY_OFFSET + 8].copy_from_slice(&u64::MAX.to_le_bytes());
assert_eq!(
decode_staged_fragment(&length_overflow),
Err(FormatError::LengthOverflow)
);
let mut invalid_utf8 = encoded.clone();
invalid_utf8[BODY_OFFSET + 8] = 0xff;
assert_eq!(
decode_staged_fragment(&invalid_utf8),
Err(FormatError::InvalidUtf8)
);
let (feature_offset, boolean_offset) = first_staged_feature_and_boolean_offsets(&encoded);
let mut invalid_feature = encoded.clone();
invalid_feature[feature_offset] = 2;
assert_eq!(
decode_staged_fragment(&invalid_feature),
Err(FormatError::InvalidFeatureBody)
);
let mut invalid_boolean = encoded.clone();
invalid_boolean[boolean_offset] = 2;
assert_eq!(
decode_staged_fragment(&invalid_boolean),
Err(FormatError::InvalidBoolean(2))
);
let mut trailing = encoded;
trailing.push(0);
assert_eq!(
decode_staged_fragment(&trailing),
Err(FormatError::TrailingBytes)
);
}
#[test]
fn malformed_label_and_internal_feature_trailing_bytes_are_rejected() {
let value = StagedFragmentV1 {
analysis_txid: txid(1),
transcript: "hello".to_owned(),
speakers: vec![StagedSpeakerV1 {
speaker: LocalSpeakerLabel::new(1).expect("valid label"),
language: "English".to_owned(),
features: FeatureVector24::default(),
usable_for_training: true,
}],
};
let encoded = encode_staged_fragment(&value).expect("encode staged");
let mut position = BODY_OFFSET;
let transcript_length = read_test_length(&encoded, &mut position);
position += transcript_length;
let _ = read_test_length(&encoded, &mut position);
let label_length = read_test_length(&encoded, &mut position);
assert_eq!(label_length, "Speaker 1".len());
let mut invalid_label = encoded.clone();
invalid_label[position] = b'X';
assert_eq!(
decode_staged_fragment(&invalid_label),
Err(FormatError::InvalidSpeakerLabel)
);
let (feature_offset, boolean_offset) = first_staged_feature_and_boolean_offsets(&encoded);
let mut feature_trailing = encoded;
feature_trailing.insert(boolean_offset, 0);
let feature_length_offset = feature_offset - 8;
let feature_length = boolean_offset - feature_offset + 1;
feature_trailing[feature_length_offset..feature_length_offset + 8]
.copy_from_slice(&(feature_length as u64).to_le_bytes());
assert_eq!(
decode_staged_fragment(&feature_trailing),
Err(FormatError::InvalidFeatureBody)
);
}
}