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 = 4;
const QUEUE_TAG: u8 = 1;
const TRANSCRIPTION_COMPLETE_TAG: u8 = 2;
const FAILED_TAG: u8 = 3;
const DISCARDED_TAG: u8 = 4;
const LABEL_CONFIRMATION_TAG: u8 = 5;
const PROGRESS_TAG: u8 = 6;
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 FragmentStageV1 {
Queue,
Transcript,
SpeakerLabels,
SpeakerFeatures,
Structuring,
LabelConfirmation,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct QueueV2 {
#[serde(with = "txid_serde")]
pub audio_object_id: TxId,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ProgressV1 {
#[serde(with = "txid_serde")]
pub fragment_id: TxId,
pub update: ProgressUpdateV1,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum ProgressUpdateV1 {
LlmJobStarted {
sequence: u64,
stage: FragmentStageV1,
name: String,
},
LlmJobSucceeded {
sequence: u64,
},
LlmJobFailed {
sequence: u64,
error: String,
},
StageCompleted {
stage: FragmentStageV1,
},
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct TranscriptionCompleteV1 {
#[serde(with = "txid_serde")]
pub fragment_id: TxId,
pub analysis: ExecutedAnalysis,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FailedV2 {
#[serde(with = "txid_serde")]
pub fragment_id: TxId,
pub stage: FragmentStageV1,
pub llm_job_sequence: Option<u64>,
pub error: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DiscardedV2 {
#[serde(with = "txid_serde")]
pub fragment_id: TxId,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SpeakerLabelV1 {
pub speaker: LocalSpeakerLabel,
pub person_id: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct LabelConfirmationV1 {
#[serde(with = "txid_serde")]
pub fragment_id: TxId,
#[serde(with = "txid_serde")]
pub interim_txid: TxId,
pub speakers: Vec<SpeakerLabelV1>,
}
#[allow(clippy::large_enum_variant)]
#[derive(Debug, Clone, PartialEq)]
pub enum AudioClassificationEventV3 {
Queue(QueueV2),
Progress(ProgressV1),
TranscriptionComplete(TranscriptionCompleteV1),
Failed(FailedV2),
Discarded(DiscardedV2),
LabelConfirmation(LabelConfirmationV1),
}
#[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 output = [0; Self::LEN];
output[..12].copy_from_slice(self.txid.as_bytes());
output
}
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 = [0; 12];
txid.copy_from_slice(&bytes[..12]);
Ok(Self::new(TxId::from_bytes(txid)))
}
}
pub fn encode_event(event: &AudioClassificationEventV3) -> Result<Vec<u8>, FormatError> {
let (tag, body) = match event {
AudioClassificationEventV3::Queue(value) => (QUEUE_TAG, postcard::to_allocvec(value)),
AudioClassificationEventV3::TranscriptionComplete(value) => {
(TRANSCRIPTION_COMPLETE_TAG, postcard::to_allocvec(value))
}
AudioClassificationEventV3::Failed(value) => (FAILED_TAG, postcard::to_allocvec(value)),
AudioClassificationEventV3::Discarded(value) => {
(DISCARDED_TAG, postcard::to_allocvec(value))
}
AudioClassificationEventV3::LabelConfirmation(value) => {
(LABEL_CONFIRMATION_TAG, postcard::to_allocvec(value))
}
AudioClassificationEventV3::Progress(value) => (PROGRESS_TAG, postcard::to_allocvec(value)),
};
let body = body.map_err(|_| FormatError::InvalidEventBody)?;
let mut output = Vec::with_capacity(2 + body.len());
output.extend_from_slice(&[EVENT_VERSION, tag]);
output.extend_from_slice(&body);
Ok(output)
}
pub fn decode_event(bytes: &[u8]) -> Result<AudioClassificationEventV3, FormatError> {
if bytes.len() < 2 {
return Err(FormatError::Truncated);
}
if bytes[0] != EVENT_VERSION {
return Err(FormatError::UnsupportedVersion(bytes[0]));
}
match bytes[1] {
QUEUE_TAG => decode_event_body(&bytes[2..]).map(AudioClassificationEventV3::Queue),
TRANSCRIPTION_COMPLETE_TAG => {
decode_event_body(&bytes[2..]).map(AudioClassificationEventV3::TranscriptionComplete)
}
FAILED_TAG => decode_event_body(&bytes[2..]).map(AudioClassificationEventV3::Failed),
DISCARDED_TAG => decode_event_body(&bytes[2..]).map(AudioClassificationEventV3::Discarded),
LABEL_CONFIRMATION_TAG => {
decode_event_body(&bytes[2..]).map(AudioClassificationEventV3::LabelConfirmation)
}
PROGRESS_TAG => decode_event_body(&bytes[2..]).map(AudioClassificationEventV3::Progress),
tag => Err(FormatError::UnknownEventTag(tag)),
}
}
pub fn encode_staged_fragment(value: &StagedFragmentV1) -> Result<Vec<u8>, FormatError> {
let mut output = encode_fragment_prefix(STAGED_KIND, value.analysis_txid, None);
append_string(&mut output, &value.transcript)?;
append_length(&mut output, value.speakers.len())?;
for speaker in &value.speakers {
append_string(&mut output, &speaker.speaker.to_string())?;
append_string(&mut output, &speaker.language)?;
append_features(&mut output, &speaker.features)?;
output.push(u8::from(speaker.usable_for_training));
}
Ok(output)
}
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 output = encode_fragment_prefix(
FINAL_KIND,
value.analysis_txid,
Some(value.confirmation_txid),
);
append_string(&mut output, &value.transcript)?;
append_length(&mut output, value.speakers.len())?;
for speaker in &value.speakers {
append_string(&mut output, &speaker.speaker.to_string())?;
append_string(&mut output, &speaker.person_id)?;
append_string(&mut output, &speaker.language)?;
append_features(&mut output, &speaker.features)?;
output.push(u8::from(speaker.usable_for_training));
}
Ok(output)
}
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 fragment header always contains a confirmation slot"),
transcript,
speakers,
})
}
pub fn txid_path(txid: TxId) -> PathBuf {
let encoded = encode_txid_base64(txid);
let shard = String::from_utf8(encoded[..1].to_vec()).expect("base64 alphabet is ASCII");
let name = String::from_utf8(encoded[1..].to_vec()).expect("base64 alphabet is ASCII");
PathBuf::from(shard).join(format!("{name}.dat"))
}
pub fn txid_from_path(path: impl AsRef<Path>) -> Result<TxId, FormatError> {
let mut components = path.as_ref().components();
let shard = normal_utf8_component(components.next())?;
let filename = normal_utf8_component(components.next())?;
if components.next().is_some() || shard.len() != 1 {
return Err(FormatError::InvalidPath);
}
let name = filename
.strip_suffix(".dat")
.ok_or(FormatError::InvalidPath)?;
if name.len() != 15 {
return Err(FormatError::InvalidPath);
}
let mut encoded = [0; 16];
encoded[0] = shard.as_bytes()[0];
encoded[1..].copy_from_slice(name.as_bytes());
decode_txid_base64(encoded)
}
fn decode_event_body<T: for<'de> Deserialize<'de>>(bytes: &[u8]) -> Result<T, FormatError> {
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: TxId,
confirmation_txid: Option<TxId>,
) -> Vec<u8> {
let mut output = vec![0; BODY_OFFSET];
output[0] = FRAGMENT_VERSION;
output[1] = kind;
output[ANALYSIS_SLOT_OFFSET..ANALYSIS_SLOT_OFFSET + TxIdSlot::LEN]
.copy_from_slice(&TxIdSlot::new(analysis_txid).encode());
if let Some(txid) = confirmation_txid {
output[CONFIRMATION_SLOT_OFFSET..CONFIRMATION_SLOT_OFFSET + TxIdSlot::LEN]
.copy_from_slice(&TxIdSlot::new(txid).encode());
}
output
}
fn decode_fragment_header(
bytes: &[u8],
expected_kind: u8,
has_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_txid =
TxIdSlot::decode(&bytes[ANALYSIS_SLOT_OFFSET..CONFIRMATION_SLOT_OFFSET])?.txid();
let confirmation_slot = &bytes[CONFIRMATION_SLOT_OFFSET..BODY_OFFSET];
let confirmation_txid = if has_confirmation {
Some(TxIdSlot::decode(confirmation_slot)?.txid())
} else {
if confirmation_slot.iter().any(|byte| *byte != 0) {
return Err(FormatError::NonZeroStagedConfirmation);
}
None
};
Ok((analysis_txid, confirmation_txid))
}
fn append_length(output: &mut Vec<u8>, length: usize) -> Result<(), FormatError> {
output.extend_from_slice(
&u64::try_from(length)
.map_err(|_| FormatError::LengthOverflow)?
.to_le_bytes(),
);
Ok(())
}
fn append_bytes(output: &mut Vec<u8>, bytes: &[u8]) -> Result<(), FormatError> {
append_length(output, bytes.len())?;
output.extend_from_slice(bytes);
Ok(())
}
fn append_string(output: &mut Vec<u8>, value: &str) -> Result<(), FormatError> {
append_bytes(output, value.as_bytes())
}
fn append_features(output: &mut Vec<u8>, features: &FeatureVector24) -> Result<(), FormatError> {
let bytes = postcard::to_allocvec(features).map_err(|_| FormatError::InvalidFeatureBody)?;
append_bytes(output, &bytes)
}
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 mut bytes = [0; 8];
bytes.copy_from_slice(self.take(8)?);
Ok(u64::from_le_bytes(bytes))
}
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 (value, remaining) =
postcard::take_from_bytes(bytes).map_err(|_| FormatError::InvalidFeatureBody)?;
if !remaining.is_empty() {
return Err(FormatError::InvalidFeatureBody);
}
Ok(value)
}
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; 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; 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: Serializer>(value: &TxId, serializer: S) -> Result<S::Ok, S::Error> {
value.into_bytes().serialize(serializer)
}
pub fn deserialize<'de, D: Deserializer<'de>>(deserializer: D) -> Result<TxId, D::Error> {
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; 12];
for (index, value) in bytes.iter_mut().enumerate() {
*value = seed.wrapping_add(index as u8);
}
TxId::from_bytes(bytes)
}
fn features(seed: f64) -> FeatureVector24 {
FeatureVector24 {
median_f0_hz: Some(seed),
..FeatureVector24::default()
}
}
fn analysis() -> ExecutedAnalysis {
let mut ogg = vec![0; 29];
ogg[..4].copy_from_slice(b"OggS");
ogg[26] = 1;
ogg[27] = 1;
ExecutedAnalysis {
envelope: AnalysisEnvelope {
audio: OggAudioMetadata::from_bytes(&ogg, 1_250, Some("sample.ogg".into()))
.expect("Ogg metadata"),
analysis: StructuredAnalysis {
transcript: "Speaker 1".into(),
speakers: vec![StructuredSpeaker {
speaker: LocalSpeakerLabel::new(1).expect("speaker label"),
language: "English".into(),
features: features(182.5),
features_usable_for_training: true,
}],
},
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: "hello".into(),
speakers: vec![StagedSpeakerV1 {
speaker: LocalSpeakerLabel::new(1).expect("speaker label"),
language: "English".into(),
features: features(201.25),
usable_for_training: true,
}],
}
}
fn final_fragment() -> FinalFragmentV1 {
FinalFragmentV1 {
analysis_txid: txid(30),
confirmation_txid: txid(40),
transcript: "hello".into(),
speakers: vec![FinalSpeakerV1 {
speaker: LocalSpeakerLabel::new(1).expect("speaker label"),
person_id: "person-1".into(),
language: "English".into(),
features: features(201.25),
usable_for_training: false,
}],
}
}
fn assert_event_roundtrip(event: AudioClassificationEventV3, expected_tag: u8) {
let bytes = encode_event(&event).expect("encode event");
assert_eq!(bytes[0], EVENT_VERSION);
assert_eq!(bytes[1], expected_tag);
assert_eq!(decode_event(&bytes), Ok(event));
}
#[test]
fn slot_roundtrips_and_rejects_padding() {
let id = txid(3);
let encoded = TxIdSlot::new(id).encode();
assert_eq!(TxIdSlot::decode(&encoded).expect("decode slot").txid(), id);
let mut bad_padding = encoded;
bad_padding[12] = 1;
assert_eq!(
TxIdSlot::decode(&bad_padding),
Err(FormatError::NonZeroPadding)
);
}
#[test]
fn all_v4_events_roundtrip_with_stable_tags() {
let speaker = LocalSpeakerLabel::new(1).expect("speaker label");
assert_event_roundtrip(
AudioClassificationEventV3::Queue(QueueV2 {
audio_object_id: txid(1),
}),
QUEUE_TAG,
);
assert_event_roundtrip(
AudioClassificationEventV3::TranscriptionComplete(TranscriptionCompleteV1 {
fragment_id: txid(2),
analysis: analysis(),
}),
TRANSCRIPTION_COMPLETE_TAG,
);
assert_event_roundtrip(
AudioClassificationEventV3::Failed(FailedV2 {
fragment_id: txid(3),
stage: FragmentStageV1::SpeakerFeatures,
llm_job_sequence: Some(7),
error: "failure".into(),
}),
FAILED_TAG,
);
assert_event_roundtrip(
AudioClassificationEventV3::Discarded(DiscardedV2 {
fragment_id: txid(4),
}),
DISCARDED_TAG,
);
assert_event_roundtrip(
AudioClassificationEventV3::LabelConfirmation(LabelConfirmationV1 {
fragment_id: txid(5),
interim_txid: txid(6),
speakers: vec![SpeakerLabelV1 {
speaker,
person_id: "person-1".into(),
}],
}),
LABEL_CONFIRMATION_TAG,
);
assert_event_roundtrip(
AudioClassificationEventV3::Progress(ProgressV1 {
fragment_id: txid(7),
update: ProgressUpdateV1::LlmJobStarted {
sequence: 9,
stage: FragmentStageV1::Transcript,
name: "transcribe".into(),
},
}),
PROGRESS_TAG,
);
}
#[test]
fn every_progress_update_roundtrips() {
let updates = [
ProgressUpdateV1::LlmJobStarted {
sequence: 1,
stage: FragmentStageV1::Transcript,
name: "transcript".into(),
},
ProgressUpdateV1::LlmJobSucceeded { sequence: 1 },
ProgressUpdateV1::LlmJobFailed {
sequence: 2,
error: "provider failed".into(),
},
ProgressUpdateV1::StageCompleted {
stage: FragmentStageV1::Transcript,
},
];
for update in updates {
assert_event_roundtrip(
AudioClassificationEventV3::Progress(ProgressV1 {
fragment_id: txid(8),
update,
}),
PROGRESS_TAG,
);
}
}
#[test]
fn events_reject_v3_unknown_tags_malformed_bodies_and_trailing_bytes() {
let event = AudioClassificationEventV3::Queue(QueueV2 {
audio_object_id: txid(1),
});
let bytes = encode_event(&event).expect("encode event");
let mut version_three = bytes.clone();
version_three[0] = 3;
assert_eq!(
decode_event(&version_three),
Err(FormatError::UnsupportedVersion(3))
);
let mut unknown_tag = bytes.clone();
unknown_tag[1] = 7;
assert_eq!(
decode_event(&unknown_tag),
Err(FormatError::UnknownEventTag(7))
);
assert_eq!(
decode_event(&bytes[..2]),
Err(FormatError::InvalidEventBody)
);
let mut trailing = bytes;
trailing.push(0);
assert_eq!(decode_event(&trailing), Err(FormatError::TrailingBytes));
}
#[test]
fn fragments_roundtrip_and_corruption_rejects() {
let staged = staged_fragment();
let staged_bytes = encode_staged_fragment(&staged).expect("encode staged fragment");
assert_eq!(
&staged_bytes[ANALYSIS_SLOT_OFFSET..CONFIRMATION_SLOT_OFFSET],
&TxIdSlot::new(staged.analysis_txid).encode()
);
assert_eq!(
&staged_bytes[CONFIRMATION_SLOT_OFFSET..BODY_OFFSET],
&[0; TxIdSlot::LEN]
);
assert_eq!(decode_staged_fragment(&staged_bytes), Ok(staged));
let final_value = final_fragment();
assert_eq!(
decode_final_fragment(
&encode_final_fragment(&final_value).expect("encode final fragment")
),
Ok(final_value)
);
let mut bad_reserved = staged_bytes;
bad_reserved[2] = 1;
assert_eq!(
decode_staged_fragment(&bad_reserved),
Err(FormatError::NonZeroReserved)
);
}
#[test]
fn paths_roundtrip() {
let id = txid(99);
assert_eq!(txid_from_path(txid_path(id)), Ok(id));
assert_eq!(
txid_from_path("A/invalid.dat"),
Err(FormatError::InvalidPath)
);
}
}