use kcode_audio_history_handoff::RecordingProjection;
use kcode_audio_ingress::{
ChunkConfirmation, ConfirmationState, CorrectionPacket, LegacyReviewDisposition,
RecordingState, RecordingStatus,
};
use uuid::Uuid;
use crate::{Coordinator, Error, audio_error, handoff_error, recording_belongs_to};
impl Coordinator {
pub(crate) fn owned_recordings(&self) -> Result<Vec<RecordingStatus>, Error> {
Ok(self
.audio
.status()
.map_err(audio_error)?
.recordings
.into_iter()
.filter(|recording| recording_belongs_to(recording, &self.user_id))
.collect())
}
pub(crate) async fn project_recordings(
&self,
recordings: &[RecordingStatus],
) -> Result<Vec<RecordingProjection>, Error> {
let projections = self
.handoff
.project(recordings)
.await
.map_err(handoff_error)?;
if recordings.len() != projections.len() {
return Err(Error::internal("handoff returned incomplete projections"));
}
Ok(projections)
}
pub(crate) async fn resolve_legacy_reviews(
&self,
recordings: &[RecordingStatus],
projections: &[RecordingProjection],
) -> Result<bool, Error> {
let decisions = decisions(recordings, projections);
for (recording_id, disposition) in &decisions {
self.audio
.resolve_legacy_review(*recording_id, *disposition)
.map_err(audio_error)?;
}
Ok(!decisions.is_empty())
}
pub(crate) async fn prepared_recordings(
&self,
) -> Result<(Vec<RecordingStatus>, Vec<RecordingProjection>), Error> {
let mut recordings = self.owned_recordings()?;
let mut projections = self.project_recordings(&recordings).await?;
if self
.resolve_legacy_reviews(&recordings, &projections)
.await?
{
recordings = self.owned_recordings()?;
projections = self.project_recordings(&recordings).await?;
}
Ok((recordings, projections))
}
}
pub(crate) fn decisions(
recordings: &[RecordingStatus],
projections: &[RecordingProjection],
) -> Vec<(Uuid, LegacyReviewDisposition)> {
recordings
.iter()
.zip(projections)
.filter_map(|(recording, projection)| {
debug_assert_eq!(recording.id, projection.recording_id);
disposition(recording, !projection.pieces.is_empty())
.map(|disposition| (recording.id, disposition))
})
.collect()
}
pub(crate) fn has_candidates(recordings: &[RecordingStatus]) -> bool {
recordings
.iter()
.any(|recording| disposition(recording, false).is_some())
}
fn disposition(recording: &RecordingStatus, has_ingress: bool) -> Option<LegacyReviewDisposition> {
if !matches!(&recording.state, RecordingState::Complete { .. })
|| recording.correction_packet.as_ref()?.confirmation_state == ConfirmationState::Confirmed
{
return None;
}
Some(if has_ingress {
LegacyReviewDisposition::Complete
} else {
LegacyReviewDisposition::Reprocess
})
}
pub(crate) fn confirmation_matches(
packet: &CorrectionPacket,
confirmation: &ChunkConfirmation,
) -> bool {
let Some(chunk) = packet
.chunks
.get(confirmation.chunk_index)
.filter(|chunk| chunk.chunk_index == confirmation.chunk_index && chunk.signed_off)
else {
return false;
};
chunk.observations.len() == confirmation.observations.len()
&& confirmation.observations.iter().all(|observation| {
chunk
.observations
.iter()
.find(|existing| existing.observation_key == observation.observation_key)
.and_then(|existing| existing.resolution.as_ref())
== Some(&observation.resolution)
})
}
#[cfg(test)]
mod tests {
use chrono::Utc;
use kcode_audio_ingress::{
ConfirmationState, CorrectionPacket, LegacyReviewDisposition, RecordingState,
RecordingStatus,
};
use uuid::Uuid;
use super::{disposition, has_candidates};
fn finalized(state: ConfirmationState) -> RecordingStatus {
let id = Uuid::new_v4();
let now = Utc::now();
RecordingStatus {
id,
user_id: "user".into(),
sha256: "0".repeat(64),
original_filename: "legacy.wav".into(),
size_bytes: 42,
recorded_at: now,
received_at: now,
transcription_model: "gemini".into(),
reconciliation_model: "gpt".into(),
reconciliation_reasoning: "xhigh".into(),
state: RecordingState::Complete {
transcript: "Legacy transcript".into(),
},
correction_packet: Some(CorrectionPacket {
recording_id: id,
user_id: "user".into(),
sha256: "0".repeat(64),
original_filename: "legacy.wav".into(),
size_bytes: 42,
recorded_at: now,
chunk_count: 0,
chunks: Vec::new(),
confirmation_state: state,
}),
}
}
#[test]
fn unresolved_finalized_packets_follow_history_ingress() {
let recording = finalized(ConfirmationState::Unconfirmed);
assert_eq!(
disposition(&recording, false),
Some(LegacyReviewDisposition::Reprocess)
);
assert_eq!(
disposition(&recording, true),
Some(LegacyReviewDisposition::Complete)
);
assert!(has_candidates(std::slice::from_ref(&recording)));
let confirmed = finalized(ConfirmationState::Confirmed);
assert_eq!(disposition(&confirmed, false), None);
assert!(!has_candidates(&[confirmed]));
let automatically_trained = finalized(ConfirmationState::AutomaticallyTrained);
assert_eq!(
disposition(&automatically_trained, false),
Some(LegacyReviewDisposition::Reprocess)
);
}
}