use std::collections::BTreeSet;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use kcode_k1_audio_classification::{
AudioClassification, FragmentId, FragmentStatus, OverallState,
};
use kcode_k1_objects::{K1Objects, Object};
const MANIFEST_FILE_TYPE: &str = "k1-full-audio-manifest-v1";
const MANIFEST_VERSION: u8 = 1;
const MANIFEST_HEADER_LEN: usize = 9;
const MANIFEST_ENTRY_LEN: usize = 28;
pub type FullAudioId = kcode_k1_objects::TxId;
pub struct K1FullAudio {
ffmpeg_path: PathBuf,
objects: Arc<K1Objects>,
classification: Arc<AudioClassification>,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum FullAudioState {
Processing,
AwaitingLabels,
NeedsAttention,
Complete,
}
#[derive(Clone, Debug, PartialEq)]
pub struct FullAudioFragmentStatus {
pub fragment_id: FragmentId,
pub start_sample_48k: u64,
pub end_sample_48k: u64,
pub status: FragmentStatus,
}
#[derive(Clone, Debug, PartialEq)]
pub struct FullAudioStatus {
pub state: FullAudioState,
pub fragments: Vec<FullAudioFragmentStatus>,
pub final_transcript: Option<String>,
}
#[derive(Clone, Debug, Eq, PartialEq)]
struct ManifestEntry {
fragment_id: FragmentId,
start_sample_48k: u64,
end_sample_48k: u64,
}
struct PendingFragment {
start_sample_48k: u64,
end_sample_48k: u64,
ogg_bytes: Vec<u8>,
}
impl K1FullAudio {
pub fn open(
ffmpeg_path: impl AsRef<Path>,
objects: Arc<K1Objects>,
classification: Arc<AudioClassification>,
) -> Result<Self, String> {
Ok(Self {
ffmpeg_path: validate_ffmpeg_path(ffmpeg_path.as_ref())?,
objects,
classification,
})
}
pub fn submit(&self, audio: &[u8]) -> Result<FullAudioId, String> {
submit_orchestration(
audio,
|input| kcode_audio_to_ogg_opus::convert_to_ogg_opus(&self.ffmpeg_path, input),
|ogg_bytes| {
kcode_ogg_opus_fragments::split_ogg_opus(ogg_bytes).map(|fragments| {
fragments
.into_iter()
.map(|fragment| PendingFragment {
start_sample_48k: fragment.start_sample_48k,
end_sample_48k: fragment.end_sample_48k,
ogg_bytes: fragment.ogg_bytes,
})
.collect()
})
},
|ogg_bytes| self.classification.submit(ogg_bytes),
|manifest| self.objects.save("", MANIFEST_FILE_TYPE, "", manifest),
)
}
pub fn status(&self, id: FullAudioId) -> Result<FullAudioStatus, String> {
status_orchestration(
id,
|manifest_id| self.objects.load(manifest_id),
|fragment_id| self.classification.status(fragment_id),
)
}
}
fn validate_ffmpeg_path(path: &Path) -> Result<PathBuf, String> {
if !path.is_absolute() {
return Err("ffmpeg path must be absolute".to_owned());
}
Ok(path.to_path_buf())
}
fn submit_orchestration<Convert, Split, Submit, Save>(
audio: &[u8],
convert: Convert,
split: Split,
mut submit: Submit,
save: Save,
) -> Result<FullAudioId, String>
where
Convert: FnOnce(&[u8]) -> Result<Vec<u8>, String>,
Split: FnOnce(&[u8]) -> Result<Vec<PendingFragment>, String>,
Submit: FnMut(&[u8]) -> Result<FragmentId, String>,
Save: FnOnce(&[u8]) -> Result<FullAudioId, String>,
{
let ogg_bytes = convert(audio)?;
let fragments = split(&ogg_bytes)?;
validate_pending_fragments(&fragments)?;
let mut entries = Vec::new();
entries
.try_reserve_exact(fragments.len())
.map_err(|error| format!("allocate manifest entries: {error}"))?;
for fragment in fragments {
let fragment_id = submit(&fragment.ogg_bytes)?;
entries.push(ManifestEntry {
fragment_id,
start_sample_48k: fragment.start_sample_48k,
end_sample_48k: fragment.end_sample_48k,
});
}
let manifest = encode_manifest(&entries)?;
save(&manifest)
}
fn status_orchestration<Load, Status>(
id: FullAudioId,
mut load: Load,
mut status: Status,
) -> Result<FullAudioStatus, String>
where
Load: FnMut(FullAudioId) -> Result<Option<Object>, String>,
Status: FnMut(FragmentId) -> Result<Option<FragmentStatus>, String>,
{
let object = load(id)?.ok_or_else(|| "unknown full-audio ID".to_owned())?;
if object.file_type != MANIFEST_FILE_TYPE
|| !object.filename.is_empty()
|| !object.description.is_empty()
{
return Err("object is not a full-audio manifest".to_owned());
}
let entries = decode_manifest(&object.data)?;
let mut fragments = Vec::new();
fragments
.try_reserve_exact(entries.len())
.map_err(|error| format!("allocate fragment statuses: {error}"))?;
for entry in entries {
let fragment_status = status(entry.fragment_id)?
.ok_or_else(|| format!("unknown audio fragment {}", entry.fragment_id))?;
fragments.push(FullAudioFragmentStatus {
fragment_id: entry.fragment_id,
start_sample_48k: entry.start_sample_48k,
end_sample_48k: entry.end_sample_48k,
status: fragment_status,
});
}
let state = derive_state(&fragments)?;
let final_transcript = if state == FullAudioState::Complete {
let transcripts = fragments
.iter()
.filter_map(|fragment| fragment.status.final_transcript.as_deref())
.collect::<Vec<_>>();
Some(stitch_transcripts(&transcripts))
} else {
None
};
Ok(FullAudioStatus {
state,
fragments,
final_transcript,
})
}
fn validate_pending_fragments(fragments: &[PendingFragment]) -> Result<(), String> {
if fragments.is_empty() {
return Err("audio produced no fragments".to_owned());
}
let mut previous = None;
for fragment in fragments {
validate_geometry(previous, fragment.start_sample_48k, fragment.end_sample_48k)?;
previous = Some((fragment.start_sample_48k, fragment.end_sample_48k));
}
Ok(())
}
fn validate_manifest_entries(entries: &[ManifestEntry]) -> Result<(), String> {
if entries.is_empty() {
return Err("manifest has no fragments".to_owned());
}
let mut ids = BTreeSet::new();
let mut previous = None;
for entry in entries {
if !ids.insert(entry.fragment_id) {
return Err("manifest has duplicate fragment IDs".to_owned());
}
validate_geometry(previous, entry.start_sample_48k, entry.end_sample_48k)?;
previous = Some((entry.start_sample_48k, entry.end_sample_48k));
}
Ok(())
}
fn validate_geometry(
previous: Option<(u64, u64)>,
start_sample_48k: u64,
end_sample_48k: u64,
) -> Result<(), String> {
if start_sample_48k >= end_sample_48k {
return Err("invalid fragment interval".to_owned());
}
match previous {
None if start_sample_48k != 0 => {
return Err("first fragment must start at zero".to_owned());
}
Some((previous_start, previous_end))
if start_sample_48k <= previous_start
|| end_sample_48k <= previous_end
|| start_sample_48k >= previous_end =>
{
return Err("fragment geometry is not chronological and overlapping".to_owned());
}
_ => {}
}
Ok(())
}
fn manifest_len(count: u64) -> Result<usize, String> {
let count = usize::try_from(count).map_err(|_| "manifest length overflow".to_owned())?;
let body = count
.checked_mul(MANIFEST_ENTRY_LEN)
.ok_or_else(|| "manifest length overflow".to_owned())?;
MANIFEST_HEADER_LEN
.checked_add(body)
.ok_or_else(|| "manifest length overflow".to_owned())
}
fn encode_manifest(entries: &[ManifestEntry]) -> Result<Vec<u8>, String> {
validate_manifest_entries(entries)?;
let count = u64::try_from(entries.len()).map_err(|_| "too many fragments".to_owned())?;
let length = manifest_len(count)?;
let mut bytes = Vec::new();
bytes
.try_reserve_exact(length)
.map_err(|error| format!("allocate manifest: {error}"))?;
bytes.push(MANIFEST_VERSION);
bytes.extend_from_slice(&count.to_le_bytes());
for entry in entries {
bytes.extend_from_slice(entry.fragment_id.as_bytes());
bytes.extend_from_slice(&entry.start_sample_48k.to_le_bytes());
bytes.extend_from_slice(&entry.end_sample_48k.to_le_bytes());
}
Ok(bytes)
}
fn decode_manifest(bytes: &[u8]) -> Result<Vec<ManifestEntry>, String> {
if bytes.is_empty() {
return Err("truncated manifest".to_owned());
}
if bytes[0] != MANIFEST_VERSION {
return Err("unsupported manifest version".to_owned());
}
if bytes.len() < MANIFEST_HEADER_LEN {
return Err("truncated manifest".to_owned());
}
let mut count_bytes = [0_u8; 8];
count_bytes.copy_from_slice(&bytes[1..MANIFEST_HEADER_LEN]);
let count = u64::from_le_bytes(count_bytes);
if count == 0 {
return Err("manifest has no fragments".to_owned());
}
let expected = manifest_len(count)?;
if bytes.len() < expected {
return Err("truncated manifest".to_owned());
}
if bytes.len() > expected {
return Err("manifest has trailing bytes".to_owned());
}
let capacity = usize::try_from(count).map_err(|_| "manifest length overflow".to_owned())?;
let mut entries = Vec::new();
entries
.try_reserve_exact(capacity)
.map_err(|error| format!("allocate decoded manifest: {error}"))?;
let mut offset = MANIFEST_HEADER_LEN;
for _ in 0..count {
let fragment_id = FragmentId::from_bytes(take_array(bytes, &mut offset)?);
let start_sample_48k = u64::from_le_bytes(take_array(bytes, &mut offset)?);
let end_sample_48k = u64::from_le_bytes(take_array(bytes, &mut offset)?);
entries.push(ManifestEntry {
fragment_id,
start_sample_48k,
end_sample_48k,
});
}
validate_manifest_entries(&entries)?;
Ok(entries)
}
fn take_array<const LENGTH: usize>(
bytes: &[u8],
offset: &mut usize,
) -> Result<[u8; LENGTH], String> {
let end = offset
.checked_add(LENGTH)
.ok_or_else(|| "manifest offset overflow".to_owned())?;
let source = bytes
.get(*offset..end)
.ok_or_else(|| "truncated manifest".to_owned())?;
let mut value = [0_u8; LENGTH];
value.copy_from_slice(source);
*offset = end;
Ok(value)
}
fn derive_state(fragments: &[FullAudioFragmentStatus]) -> Result<FullAudioState, String> {
if fragments.is_empty() {
return Err("manifest has no fragments".to_owned());
}
for fragment in fragments {
match (&fragment.status.state, &fragment.status.final_transcript) {
(OverallState::Confirmed, None) => {
return Err("confirmed fragment has no final transcript".to_owned());
}
(OverallState::Confirmed, Some(_)) => {}
(_, Some(_)) => {
return Err("non-confirmed fragment has a final transcript".to_owned());
}
(_, None) => {}
}
}
if fragments.iter().any(|fragment| {
matches!(
fragment.status.state,
OverallState::Failed | OverallState::Discarded
)
}) {
return Ok(FullAudioState::NeedsAttention);
}
if fragments.iter().any(|fragment| {
matches!(
fragment.status.state,
OverallState::Queued | OverallState::Running
)
}) {
return Ok(FullAudioState::Processing);
}
if fragments
.iter()
.all(|fragment| fragment.status.state == OverallState::Confirmed)
{
return Ok(FullAudioState::Complete);
}
if fragments.iter().all(|fragment| {
matches!(
fragment.status.state,
OverallState::Completed | OverallState::Confirmed
)
}) {
return Ok(FullAudioState::AwaitingLabels);
}
Err("classification returned an inconsistent state".to_owned())
}
struct SpeechToken {
normalized: String,
start: usize,
line_start: usize,
prefix_end: Option<usize>,
}
fn stitch_transcripts(transcripts: &[&str]) -> String {
let mut result = String::new();
let mut previous = None;
for transcript in transcripts {
if transcript.is_empty() {
continue;
}
if let Some(previous_transcript) = previous {
if let Some(remainder) = duplicate_remainder(previous_transcript, transcript) {
append_transcript(&mut result, &remainder);
} else {
append_transcript(&mut result, transcript);
}
} else {
result.push_str(transcript);
}
previous = Some(*transcript);
}
result
}
fn append_transcript(result: &mut String, transcript: &str) {
if transcript.is_empty() {
return;
}
if !result.is_empty() && !result.ends_with('\n') && !transcript.starts_with('\n') {
result.push('\n');
}
result.push_str(transcript);
}
fn duplicate_remainder(previous: &str, later: &str) -> Option<String> {
if previous == later {
return Some(String::new());
}
if let Some(remainder) = exact_line_duplicate_remainder(previous, later) {
return Some(remainder);
}
let previous_tokens = speech_tokens(previous);
let later_tokens = speech_tokens(later);
let overlap = longest_token_overlap(&previous_tokens, &later_tokens);
if overlap < 3 {
return None;
}
Some(cut_duplicate_prefix(later, &later_tokens, overlap))
}
fn exact_line_duplicate_remainder(previous: &str, later: &str) -> Option<String> {
let previous_end = previous.strip_suffix('\n').map_or(previous.len(), str::len);
let previous_start = previous[..previous_end]
.rfind('\n')
.map_or(0, |index| index + 1);
let mut previous_line_end = previous_end;
if previous_line_end > previous_start && previous.as_bytes()[previous_line_end - 1] == b'\r' {
previous_line_end -= 1;
}
let later_newline = later.find('\n');
let later_end = later_newline.unwrap_or(later.len());
let mut later_line_end = later_end;
if later_line_end > 0 && later.as_bytes()[later_line_end - 1] == b'\r' {
later_line_end -= 1;
}
let previous_line = &previous[previous_start..previous_line_end];
let later_line = &later[..later_line_end];
if previous_line.is_empty() || previous_line != later_line {
return None;
}
let remainder_start = later_newline.map_or(later.len(), |index| index + 1);
Some(later[remainder_start..].to_owned())
}
fn speech_tokens(text: &str) -> Vec<SpeechToken> {
let mut tokens = Vec::new();
let mut line_start = 0;
loop {
let line_end = text[line_start..]
.find('\n')
.map_or(text.len(), |index| line_start + index);
let mut content_end = line_end;
if content_end > line_start && text.as_bytes()[content_end - 1] == b'\r' {
content_end -= 1;
}
let line = &text[line_start..content_end];
let relative_prefix_end = transcript_prefix_end(line);
let speech_start = relative_prefix_end.unwrap_or(0);
let segment = &line[speech_start..];
let mut token_start = None;
for (index, character) in segment.char_indices() {
if character.is_alphanumeric() {
if token_start.is_none() {
token_start = Some(index);
}
} else if let Some(start) = token_start.take() {
push_speech_token(
&mut tokens,
segment,
start,
index,
line_start,
speech_start,
relative_prefix_end,
);
}
}
if let Some(start) = token_start {
push_speech_token(
&mut tokens,
segment,
start,
segment.len(),
line_start,
speech_start,
relative_prefix_end,
);
}
if line_end == text.len() {
break;
}
line_start = line_end + 1;
}
tokens
}
fn push_speech_token(
tokens: &mut Vec<SpeechToken>,
segment: &str,
start: usize,
end: usize,
line_start: usize,
speech_start: usize,
relative_prefix_end: Option<usize>,
) {
tokens.push(SpeechToken {
normalized: segment[start..end].to_lowercase(),
start: line_start + speech_start + start,
line_start,
prefix_end: relative_prefix_end.map(|value| line_start + value),
});
}
fn transcript_prefix_end(line: &str) -> Option<usize> {
let marker = ["[high] ", "[medium] ", "[low] "]
.into_iter()
.find(|marker| line.starts_with(marker))?;
let colon = marker.len() + line[marker.len()..].find(':')?;
let label = &line[marker.len()..colon];
let speaker = label.strip_suffix(" [overlap]").unwrap_or(label);
if speaker.trim().is_empty() || speaker.contains("[overlap]") {
return None;
}
let mut speech_start = colon + 1;
for character in line[speech_start..].chars() {
if !character.is_whitespace() {
break;
}
speech_start += character.len_utf8();
}
Some(speech_start)
}
fn longest_token_overlap(previous: &[SpeechToken], later: &[SpeechToken]) -> usize {
if previous.is_empty() || later.is_empty() {
return 0;
}
let mut prefix_lengths = vec![0; later.len()];
for (index, token) in later.iter().enumerate().skip(1) {
let mut length = prefix_lengths[index - 1];
while length > 0 && later[length].normalized != token.normalized {
length = prefix_lengths[length - 1];
}
if later[length].normalized == token.normalized {
length += 1;
}
prefix_lengths[index] = length;
}
let mut matched = 0;
for token in previous {
if matched == later.len() {
matched = prefix_lengths[matched - 1];
}
while matched > 0 && later[matched].normalized != token.normalized {
matched = prefix_lengths[matched - 1];
}
if later[matched].normalized == token.normalized {
matched += 1;
}
}
matched
}
fn cut_duplicate_prefix(later: &str, tokens: &[SpeechToken], count: usize) -> String {
if count >= tokens.len() {
return String::new();
}
let matched = &tokens[count - 1];
let next = &tokens[count];
if matched.line_start != next.line_start {
return later[next.line_start..].to_owned();
}
if let Some(prefix_end) = next.prefix_end {
let mut remainder = String::new();
remainder.push_str(&later[next.line_start..prefix_end]);
remainder.push_str(&later[next.start..]);
return remainder;
}
later[next.start..].to_owned()
}
#[cfg(test)]
mod tests {
use super::*;
use kcode_k1_audio_classification::{FragmentStageV1, StageState, StageStatus};
use std::cell::{Cell, RefCell};
use std::rc::Rc;
use std::sync::mpsc::{self, Receiver, SyncSender};
use std::thread;
use std::time::Duration;
fn id(value: u8) -> FragmentId {
FragmentId::from_bytes([value; 12])
}
fn entry(value: u8, start: u64, end: u64) -> ManifestEntry {
ManifestEntry {
fragment_id: id(value),
start_sample_48k: start,
end_sample_48k: end,
}
}
fn raw_manifest(entries: &[(FragmentId, u64, u64)]) -> Vec<u8> {
let mut bytes = vec![MANIFEST_VERSION];
let count = u64::try_from(entries.len()).expect("test entry count");
bytes.extend_from_slice(&count.to_le_bytes());
for (fragment_id, start, end) in entries {
bytes.extend_from_slice(fragment_id.as_bytes());
bytes.extend_from_slice(&start.to_le_bytes());
bytes.extend_from_slice(&end.to_le_bytes());
}
bytes
}
#[test]
fn manifest_round_trip_is_stable() {
let entries = vec![entry(1, 0, 100), entry(2, 80, 180), entry(3, 160, 250)];
let bytes = encode_manifest(&entries).expect("encode");
assert_eq!(decode_manifest(&bytes).expect("decode"), entries);
assert_eq!(
encode_manifest(&decode_manifest(&bytes).expect("decode")).expect("encode"),
bytes
);
}
#[test]
fn manifest_rejects_version_truncation_overflow_trailing_and_empty() {
assert!(decode_manifest(&[2]).is_err());
assert!(decode_manifest(&[MANIFEST_VERSION]).is_err());
let mut truncated = raw_manifest(&[(id(1), 0, 10)]);
truncated.pop();
assert!(decode_manifest(&truncated).is_err());
let mut overflow = vec![MANIFEST_VERSION];
overflow.extend_from_slice(&u64::MAX.to_le_bytes());
assert!(decode_manifest(&overflow).is_err());
let mut trailing = raw_manifest(&[(id(1), 0, 10)]);
trailing.push(0);
assert!(decode_manifest(&trailing).is_err());
let mut empty = vec![MANIFEST_VERSION];
empty.extend_from_slice(&0_u64.to_le_bytes());
assert!(decode_manifest(&empty).is_err());
}
#[test]
fn manifest_rejects_invalid_duplicate_and_nonchronological_entries() {
assert!(decode_manifest(&raw_manifest(&[(id(1), 0, 0)])).is_err());
assert!(decode_manifest(&raw_manifest(&[(id(1), 0, 10), (id(1), 5, 15)])).is_err());
assert!(decode_manifest(&raw_manifest(&[(id(1), 1, 10)])).is_err());
assert!(decode_manifest(&raw_manifest(&[(id(1), 0, 10), (id(2), 0, 15)])).is_err());
assert!(decode_manifest(&raw_manifest(&[(id(1), 0, 10), (id(2), 5, 9)])).is_err());
assert!(decode_manifest(&raw_manifest(&[(id(1), 0, 10), (id(2), 10, 20)])).is_err());
}
fn pending(value: u8, start: u64, end: u64) -> PendingFragment {
PendingFragment {
start_sample_48k: start,
end_sample_48k: end,
ogg_bytes: vec![value],
}
}
#[test]
fn submit_orders_work_and_returns_manifest_id_without_status_wait() {
let events = Rc::new(RefCell::new(Vec::new()));
let convert_events = events.clone();
let split_events = events.clone();
let submit_events = events.clone();
let save_events = events.clone();
let returned = submit_orchestration(
b"input",
move |audio| {
assert_eq!(audio, b"input");
convert_events.borrow_mut().push("convert".to_owned());
Ok(b"ogg".to_vec())
},
move |ogg| {
assert_eq!(ogg, b"ogg");
split_events.borrow_mut().push("split".to_owned());
Ok(vec![pending(10, 0, 100), pending(20, 80, 180)])
},
move |fragment| {
submit_events
.borrow_mut()
.push(format!("submit-{}", fragment[0]));
Ok(if fragment[0] == 10 { id(1) } else { id(2) })
},
move |manifest| {
save_events.borrow_mut().push("save".to_owned());
assert_eq!(
decode_manifest(manifest).expect("saved manifest"),
vec![entry(1, 0, 100), entry(2, 80, 180)]
);
Ok(id(9))
},
)
.expect("submit");
assert_eq!(returned, id(9));
assert_eq!(
events.borrow().as_slice(),
["convert", "split", "submit-10", "submit-20", "save"]
);
}
#[test]
fn partial_submit_leaves_no_manifest() {
let submissions = Rc::new(Cell::new(0));
let saves = Rc::new(Cell::new(0));
let submit_count = submissions.clone();
let save_count = saves.clone();
let result = submit_orchestration(
b"input",
|_| Ok(vec![1]),
|_| Ok(vec![pending(1, 0, 100), pending(2, 80, 180)]),
move |_| {
let count = submit_count.get() + 1;
submit_count.set(count);
if count == 2 {
Err("second submission failed".to_owned())
} else {
Ok(id(1))
}
},
move |_| {
save_count.set(save_count.get() + 1);
Ok(id(9))
},
);
assert!(result.is_err());
assert_eq!(submissions.get(), 2);
assert_eq!(saves.get(), 0);
}
fn stage(stage: FragmentStageV1) -> StageStatus {
StageStatus {
stage,
state: StageState::Pending,
}
}
fn classification_status(
state: OverallState,
final_transcript: Option<&str>,
) -> FragmentStatus {
FragmentStatus {
state,
queue: stage(FragmentStageV1::Queue),
transcript: stage(FragmentStageV1::Transcript),
speaker_labels: stage(FragmentStageV1::SpeakerLabels),
speaker_features: stage(FragmentStageV1::SpeakerFeatures),
structuring: stage(FragmentStageV1::Structuring),
label_confirmation: stage(FragmentStageV1::LabelConfirmation),
attempt_count: 0,
jobs: Vec::new(),
interim_txid: None,
analysis: None,
confirmed_labels: Vec::new(),
final_transcript: final_transcript.map(str::to_owned),
errors: Vec::new(),
errors_truncated: false,
}
}
fn object_with_data(data: Vec<u8>) -> Object {
Object {
filename: String::new(),
file_type: MANIFEST_FILE_TYPE.to_owned(),
description: String::new(),
data,
}
}
fn status_result(statuses: Vec<FragmentStatus>) -> Result<FullAudioStatus, String> {
let entries = statuses
.iter()
.enumerate()
.map(|(index, _)| {
let value = u8::try_from(index + 1).expect("test fragment ID");
let start = u64::try_from(index).expect("test index") * 80;
entry(value, start, start + 100)
})
.collect::<Vec<_>>();
let object = object_with_data(encode_manifest(&entries).expect("manifest"));
status_orchestration(
id(90),
|_| Ok(Some(object.clone())),
|fragment_id| {
Ok(entries
.iter()
.position(|entry| entry.fragment_id == fragment_id)
.map(|index| statuses[index].clone()))
},
)
}
#[test]
fn status_derives_all_states_with_precedence() {
let attention = status_result(vec![
classification_status(OverallState::Failed, None),
classification_status(OverallState::Queued, None),
])
.expect("attention");
assert_eq!(attention.state, FullAudioState::NeedsAttention);
assert_eq!(attention.final_transcript, None);
let processing = status_result(vec![
classification_status(OverallState::Completed, None),
classification_status(OverallState::Running, None),
])
.expect("processing");
assert_eq!(processing.state, FullAudioState::Processing);
let awaiting = status_result(vec![
classification_status(OverallState::Completed, None),
classification_status(OverallState::Confirmed, Some("ready")),
])
.expect("awaiting");
assert_eq!(awaiting.state, FullAudioState::AwaitingLabels);
assert_eq!(awaiting.final_transcript, None);
let complete = status_result(vec![
classification_status(
OverallState::Confirmed,
Some("[high] A: one two three four"),
),
classification_status(
OverallState::Confirmed,
Some("[medium] B: two three four five"),
),
])
.expect("complete");
assert_eq!(complete.state, FullAudioState::Complete);
assert_eq!(
complete.final_transcript.as_deref(),
Some("[high] A: one two three four\n[medium] B: five")
);
assert_eq!(complete.fragments[0].fragment_id, id(1));
assert_eq!(complete.fragments[1].start_sample_48k, 80);
}
#[test]
fn status_rejects_transcript_inconsistency() {
assert!(status_result(vec![classification_status(OverallState::Confirmed, None)]).is_err());
assert!(
status_result(vec![classification_status(
OverallState::Completed,
Some("not confirmed")
)])
.is_err()
);
}
#[test]
fn status_rejects_unknown_wrong_and_malformed_objects() {
assert!(status_orchestration(id(90), |_| Ok(None), |_| Ok(None)).is_err());
let entries = vec![entry(1, 0, 100)];
let valid = object_with_data(encode_manifest(&entries).expect("manifest"));
assert!(status_orchestration(id(90), |_| Ok(Some(valid.clone())), |_| Ok(None)).is_err());
let mut wrong_type = valid.clone();
wrong_type.file_type = "wrong".to_owned();
assert!(
status_orchestration(id(90), |_| Ok(Some(wrong_type.clone())), |_| Ok(None)).is_err()
);
let mut wrong_filename = valid;
wrong_filename.filename = "named".to_owned();
assert!(
status_orchestration(id(90), |_| Ok(Some(wrong_filename.clone())), |_| Ok(None))
.is_err()
);
let malformed = object_with_data(vec![MANIFEST_VERSION]);
assert!(
status_orchestration(id(90), |_| Ok(Some(malformed.clone())), |_| Ok(None)).is_err()
);
}
#[test]
fn repeated_complete_status_is_stable() {
let statuses = vec![
classification_status(OverallState::Confirmed, Some("Alpha, beta gamma delta")),
classification_status(OverallState::Confirmed, Some("BETA GAMMA DELTA epsilon")),
];
let first = status_result(statuses.clone()).expect("first status");
let second = status_result(statuses).expect("second status");
assert_eq!(first, second);
assert_eq!(
first.final_transcript.as_deref(),
Some("Alpha, beta gamma delta\nepsilon")
);
}
#[test]
fn transcript_stitches_exact_case_punctuation_and_prefix_overlaps() {
assert_eq!(
stitch_transcripts(&[
"[high] Alice: alpha beta gamma delta",
"[high] Alice: beta gamma delta epsilon"
]),
"[high] Alice: alpha beta gamma delta\n[high] Alice: epsilon"
);
assert_eq!(
stitch_transcripts(&["Hello, BRAVE new world!", "hello brave NEW world; again"]),
"Hello, BRAVE new world!\nagain"
);
assert_eq!(
stitch_transcripts(&[
"[high] Speaker 1: we need three matching words now",
"[medium] David [overlap]: three matching words now please"
]),
"[high] Speaker 1: we need three matching words now\n[medium] David [overlap]: please"
);
}
#[test]
fn transcript_retains_uncertain_seams_and_handles_exact_short_lines() {
assert_eq!(
stitch_transcripts(&["one two three", "three four"]),
"one two three\nthree four"
);
assert_eq!(
stitch_transcripts(&["one two three four", "three four five"]),
"one two three four\nthree four five"
);
assert_eq!(
stitch_transcripts(&[
"[high] A: yes now",
"[high] A: yes now\n[medium] B: continue"
]),
"[high] A: yes now\n[medium] B: continue"
);
}
#[test]
fn transcript_assembly_is_repeatedly_stable_and_handles_empty_single_values() {
let transcripts = [
"[high] A: alpha beta gamma delta",
"[low] B: beta gamma delta epsilon",
];
let first = stitch_transcripts(&transcripts);
let second = stitch_transcripts(&transcripts);
assert_eq!(first, second);
assert_eq!(stitch_transcripts(&[]), "");
assert_eq!(stitch_transcripts(&[""]), "");
assert_eq!(stitch_transcripts(&["single source"]), "single source");
}
struct TestFacadeAdapter {
fragment_id: FragmentId,
manifest_id: FullAudioId,
entered: Option<SyncSender<()>>,
release: Option<Receiver<()>>,
}
impl TestFacadeAdapter {
fn submit(mut self) -> Result<FullAudioId, String> {
let mut entered = self.entered.take();
let mut release = self.release.take();
submit_orchestration(
b"input",
|_| Ok(vec![1]),
|_| Ok(vec![pending(1, 0, 100)]),
|_| {
if let Some(sender) = entered.take() {
sender
.send(())
.map_err(|error| format!("signal blocked submit: {error}"))?;
}
if let Some(receiver) = release.take() {
receiver
.recv()
.map_err(|error| format!("release blocked submit: {error}"))?;
}
Ok(self.fragment_id)
},
|_| Ok(self.manifest_id),
)
}
}
#[test]
fn blocked_operation_does_not_stall_an_independent_adapter() {
let (entered_sender, entered_receiver) = mpsc::sync_channel(1);
let (release_sender, release_receiver) = mpsc::channel();
let blocked = TestFacadeAdapter {
fragment_id: id(1),
manifest_id: id(10),
entered: Some(entered_sender),
release: Some(release_receiver),
};
let blocked_thread = thread::spawn(move || blocked.submit());
entered_receiver
.recv_timeout(Duration::from_secs(2))
.expect("blocked operation entered");
let independent = TestFacadeAdapter {
fragment_id: id(2),
manifest_id: id(20),
entered: None,
release: None,
};
assert_eq!(independent.submit().expect("independent submit"), id(20));
release_sender.send(()).expect("release blocked operation");
assert_eq!(
blocked_thread
.join()
.expect("join blocked operation")
.expect("blocked submit"),
id(10)
);
}
#[test]
fn absolute_path_validation_does_not_probe_the_filesystem() {
assert_eq!(
validate_ffmpeg_path(Path::new("/definitely/not/present")).expect("absolute path"),
PathBuf::from("/definitely/not/present")
);
assert!(validate_ffmpeg_path(Path::new("relative/ffmpeg")).is_err());
}
}