kcode_audio_session_ingress/

lib.rs

#![forbid(unsafe_code)]

use chrono::{DateTime, Utc};
use kcode_audio_history_handoff::{Error as HandoffError, Handoff};
pub use kcode_audio_ingress::SpeakerReviewAudio;
use kcode_audio_ingress::{
    AudioIngress, AudioInput, ChunkConfirmation, CorrectionPacket, ErrorKind as AudioErrorKind,
    RecordingState, RecordingStatus, confirmation_matches,
};
pub use kcode_audio_session_view::{IngressPiece, Recording, SpeakerReview};
use kcode_session_history::{SessionHistory, SessionRecord};
use serde_json::Value;
use uuid::Uuid;

mod legacy_review;

/// Application policy needed to coordinate audio with Session History.
#[derive(Clone, Debug)]
pub struct Config {
    /// Stable application user identifier attributed to accepted recordings.
    pub user_id: String,
    /// Effective context window available to an audio-ingress session.
    pub effective_context_tokens: u64,
}

/// Admitted recording bytes and source metadata.
#[derive(Clone, Debug)]
pub struct RecordingInput {
    /// Complete WAV bytes already accepted by the transport.
    pub bytes: Vec<u8>,
    /// Instant at which the original recording began.
    pub recorded_at: DateTime<Utc>,
    /// Original leaf filename, when known.
    pub original_filename: Option<String>,
}

/// Result of durably submitting one recording.
#[derive(Clone, Debug)]
pub struct RecordingSubmission {
    /// Current combined state after submission.
    pub recording: Recording,
    /// Whether AudioIngress already knew the same bytes.
    pub deduplicated: bool,
}

/// Detailed state for one recording.
#[derive(Clone, Debug)]
pub struct RecordingHistory {
    pub recording: Recording,
    pub final_transcript: Option<String>,
    pub correction_packet: Option<CorrectionPacket>,
    pub pieces: Vec<IngressPiece>,
}

/// Request to retry one transcript piece's memory ingress.
#[derive(Clone, Debug)]
pub struct RetryIngress {
    pub piece_id: String,
    pub expected_version: i64,
    /// Optional state retained for 0.2 wire compatibility. When supplied, it
    /// must exactly equal the current retained Session History state.
    pub state: Option<Value>,
}

/// Stable coordinator error category for transport mapping.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ErrorKind {
    InvalidInput,
    NotFound,
    Conflict,
    Internal,
}

/// Error returned by the audio/session-ingress coordinator.
#[derive(Debug)]
pub struct Error {
    kind: ErrorKind,
    message: String,
}

impl Error {
    fn new(kind: ErrorKind, message: impl Into<String>) -> Self {
        Self {
            kind,
            message: message.into(),
        }
    }

    fn invalid(message: impl Into<String>) -> Self {
        Self::new(ErrorKind::InvalidInput, message)
    }

    fn conflict(message: impl Into<String>) -> Self {
        Self::new(ErrorKind::Conflict, message)
    }

    fn not_found() -> Self {
        Self::new(
            ErrorKind::NotFound,
            "Audio recording or transcript piece not found.",
        )
    }

    fn internal(error: impl std::fmt::Display) -> Self {
        tracing::warn!(%error, "Audio session ingress operation failed");
        Self::new(
            ErrorKind::Internal,
            "An unexpected audio session ingress error occurred.",
        )
    }

    pub fn kind(&self) -> ErrorKind {
        self.kind
    }

    pub fn message(&self) -> &str {
        &self.message
    }
}

impl std::fmt::Display for Error {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str(&self.message)
    }
}

impl std::error::Error for Error {}

/// Cloneable typed coordinator over AudioIngress and Session History.
#[derive(Clone)]
pub struct Coordinator {
    audio: AudioIngress,
    handoff: Handoff,
    user_id: String,
}

impl Coordinator {
    pub fn new(
        audio: AudioIngress,
        history: SessionHistory,
        config: Config,
    ) -> Result<Self, Error> {
        let handoff = Handoff::new(
            history,
            config.user_id.clone(),
            config.effective_context_tokens,
        )
        .map_err(handoff_error)?;
        Ok(Self {
            audio,
            handoff,
            user_id: config.user_id,
        })
    }

    pub fn health(&self) -> Result<(), Error> {
        self.audio.status().map_err(audio_error)?;
        Ok(())
    }

    pub async fn submit(&self, input: RecordingInput) -> Result<RecordingSubmission, Error> {
        let submission = self
            .audio
            .submit(AudioInput {
                user_id: self.user_id.clone(),
                bytes: input.bytes,
                recorded_at: input.recorded_at,
                original_filename: input.original_filename,
            })
            .await
            .map_err(audio_error)?;
        let (recordings, projections) = self.prepared_recordings().await?;
        let (recording_status, projection) = recordings
            .into_iter()
            .zip(projections)
            .find(|(recording, _)| recording.id == submission.recording_id)
            .ok_or_else(Error::not_found)?;
        validate_submission_owner(&recording_status, &self.user_id)?;
        let view = kcode_audio_session_view::render(recording_status, projection);
        Ok(RecordingSubmission {
            recording: view.recording,
            deduplicated: submission.deduplicated,
        })
    }

    pub async fn recordings(&self) -> Result<Vec<Recording>, Error> {
        let (recordings, projections) = self.prepared_recordings().await?;
        Ok(recordings
            .into_iter()
            .zip(projections)
            .map(|(recording, projection)| {
                kcode_audio_session_view::render(recording, projection).recording
            })
            .collect())
    }

    pub async fn recording_by_sha256(&self, sha256: &str) -> Result<Recording, Error> {
        if sha256.len() != 64 || !sha256.bytes().all(|byte| byte.is_ascii_hexdigit()) {
            return Err(Error::invalid(
                "audio SHA-256 must contain exactly 64 hexadecimal characters",
            ));
        }
        let normalized = sha256.to_ascii_lowercase();
        self.recordings()
            .await?
            .into_iter()
            .find(|recording| recording.sha256 == normalized)
            .ok_or_else(Error::not_found)
    }

    pub async fn recording_history(&self, recording_id: Uuid) -> Result<RecordingHistory, Error> {
        self.recording_history_inner(recording_id, false).await
    }

    /// Returns recording history after re-scoring every unsigned speaker chunk.
    ///
    /// This explicit read is intended for a human opening one audio review. It
    /// runs synchronous classifier work on Tokio's blocking pool; routine
    /// background observation should use [`Coordinator::recording_history`].
    pub async fn recording_history_with_current_speaker_candidates(
        &self,
        recording_id: Uuid,
    ) -> Result<RecordingHistory, Error> {
        self.recording_history_inner(recording_id, true).await
    }

    async fn recording_history_inner(
        &self,
        recording_id: Uuid,
        refresh_speaker_candidates: bool,
    ) -> Result<RecordingHistory, Error> {
        let (recordings, projections) = self.prepared_recordings().await?;
        let (recording_status, projection) = recordings
            .into_iter()
            .zip(projections)
            .find(|(recording, _)| {
                recording.id == recording_id && recording_belongs_to(recording, &self.user_id)
            })
            .ok_or_else(Error::not_found)?;
        let final_transcript = match &recording_status.state {
            RecordingState::Complete { transcript } => Some(transcript.clone()),
            _ => None,
        };
        let correction_packet = if refresh_speaker_candidates
            && recording_status
                .correction_packet
                .as_ref()
                .is_some_and(|packet| packet.chunks.iter().any(|chunk| !chunk.signed_off))
        {
            let audio = self.audio.clone();
            Some(run_blocking_audio(move || audio.speaker_review_packet(recording_id)).await?)
        } else {
            recording_status.correction_packet.clone()
        };
        let view = kcode_audio_session_view::render(recording_status, projection);
        Ok(RecordingHistory {
            recording: view.recording,
            final_transcript,
            correction_packet,
            pieces: view.pieces,
        })
    }

    /// Returns one authorized correction-packet chunk as an exact WAV interval.
    pub async fn speaker_review_audio(
        &self,
        recording_id: Uuid,
        chunk_index: usize,
    ) -> Result<SpeakerReviewAudio, Error> {
        let owned = self
            .audio
            .status()
            .map_err(audio_error)?
            .recordings
            .into_iter()
            .any(|recording| {
                recording.id == recording_id && recording_belongs_to(&recording, &self.user_id)
            });
        if !owned {
            return Err(Error::not_found());
        }
        let audio = self.audio.clone();
        tokio::task::spawn_blocking(move || audio.speaker_review_audio(recording_id, chunk_index))
            .await
            .map_err(Error::internal)?
            .map_err(audio_error)
    }

    /// Returns the known-speaker dropdown choices for this coordinator.
    pub fn known_speakers(&self) -> Result<Vec<String>, Error> {
        self.audio.known_speakers().map_err(audio_error)
    }

    pub fn retry_recording(&self, recording_id: Uuid) -> Result<(), Error> {
        let owned = self
            .audio
            .status()
            .map_err(audio_error)?
            .recordings
            .into_iter()
            .any(|recording| {
                recording.id == recording_id && recording_belongs_to(&recording, &self.user_id)
            });
        if !owned {
            return Err(Error::not_found());
        }
        self.audio.retry(recording_id).map_err(audio_error)
    }

    pub async fn confirm_speakers(
        &self,
        confirmation: ChunkConfirmation,
    ) -> Result<CorrectionPacket, Error> {
        let recording_id = confirmation.recording_id;
        let (recordings, _) = self.prepared_recordings().await?;
        let recording = recordings
            .into_iter()
            .find(|recording| {
                recording.id == recording_id && recording_belongs_to(recording, &self.user_id)
            })
            .ok_or_else(Error::not_found)?;

        if let Some(packet) = recording
            .correction_packet
            .as_ref()
            .filter(|packet| confirmation_matches(packet, &confirmation))
        {
            self.handoff
                .synchronize(std::slice::from_ref(&recording))
                .await
                .map_err(handoff_error)?;
            return Ok(packet.clone());
        }
        if self
            .handoff
            .has_ingress(recording_id)
            .await
            .map_err(handoff_error)?
        {
            return Err(Error::conflict(
                "speaker labels are already bound to accepted transcript ingress",
            ));
        }

        let audio = self.audio.clone();
        let packet = run_blocking_audio(move || audio.confirm_speakers(confirmation)).await?;
        let recording = self
            .audio
            .status()
            .map_err(audio_error)?
            .recordings
            .into_iter()
            .find(|recording| {
                recording.id == recording_id && recording_belongs_to(recording, &self.user_id)
            })
            .ok_or_else(Error::not_found)?;
        self.handoff
            .synchronize(std::slice::from_ref(&recording))
            .await
            .map_err(handoff_error)?;
        Ok(packet)
    }

    pub async fn retry_ingress(&self, input: RetryIngress) -> Result<SessionRecord, Error> {
        let recordings = self.audio.status().map_err(audio_error)?.recordings;
        self.handoff
            .retry(
                &recordings,
                &input.piece_id,
                input.expected_version,
                input.state,
            )
            .await
            .map_err(handoff_error)
    }

    pub async fn synchronize_completed_transcripts(&self) -> Result<(), Error> {
        let mut recordings = self.owned_recordings()?;
        if legacy_review::has_candidates(&recordings) {
            let projections = self.project_recordings(&recordings).await?;
            if self
                .resolve_legacy_reviews(&recordings, &projections)
                .await?
            {
                recordings = self.owned_recordings()?;
            }
        }
        self.handoff
            .synchronize(&recordings)
            .await
            .map_err(handoff_error)
    }
}

fn recording_belongs_to(recording: &RecordingStatus, user_id: &str) -> bool {
    recording.user_id == user_id
}

fn validate_submission_owner(recording: &RecordingStatus, user_id: &str) -> Result<(), Error> {
    if recording_belongs_to(recording, user_id) {
        Ok(())
    } else {
        Err(Error::conflict(
            "identical audio is already attributed to another user",
        ))
    }
}

fn audio_error(error: kcode_audio_ingress::Error) -> Error {
    match error.kind() {
        AudioErrorKind::InvalidInput => Error::new(ErrorKind::InvalidInput, error.to_string()),
        AudioErrorKind::NotFound => Error::new(ErrorKind::NotFound, error.to_string()),
        AudioErrorKind::Conflict => Error::new(ErrorKind::Conflict, error.to_string()),
        AudioErrorKind::Internal => Error::internal(error),
    }
}

async fn run_blocking_audio<T, Operation>(operation: Operation) -> Result<T, Error>
where
    T: Send + 'static,
    Operation: FnOnce() -> Result<T, kcode_audio_ingress::Error> + Send + 'static,
{
    tokio::task::spawn_blocking(operation)
        .await
        .map_err(Error::internal)?
        .map_err(audio_error)
}

fn handoff_error(error: HandoffError) -> Error {
    match error {
        HandoffError::InvalidInput(message) => Error::new(ErrorKind::InvalidInput, message),
        HandoffError::NotFound(message) => Error::new(ErrorKind::NotFound, message),
        HandoffError::Conflict(message) => Error::new(ErrorKind::Conflict, message),
        HandoffError::Internal(message) => Error::new(ErrorKind::Internal, message),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn completed_recording_for_user(
        user_id: impl Into<String>,
        transcript: impl Into<String>,
    ) -> RecordingStatus {
        let now = Utc::now();
        RecordingStatus {
            id: Uuid::new_v4(),
            user_id: user_id.into(),
            sha256: "0".repeat(64),
            original_filename: "meeting.final.WAV".into(),
            size_bytes: 42,
            recorded_at: now,
            received_at: now,
            transcription_model: "transcription-model".into(),
            reconciliation_model: "reconciliation-model".into(),
            reconciliation_reasoning: "xhigh".into(),
            state: RecordingState::Complete {
                transcript: transcript.into(),
            },
            correction_packet: None,
        }
    }

    #[test]
    fn recording_filter_is_exactly_scoped_to_the_configured_user() {
        let own = completed_recording_for_user("own-user", "Own transcript");
        let foreign = completed_recording_for_user("foreign-user", "Foreign transcript");
        let visible = [own.clone(), foreign]
            .into_iter()
            .filter(|recording| recording_belongs_to(recording, "own-user"))
            .collect::<Vec<_>>();
        assert_eq!(visible.len(), 1);
        assert_eq!(visible[0].id, own.id);
    }

    #[test]
    fn cross_user_sha_deduplication_fails_closed() {
        let mut own = completed_recording_for_user("own-user", "Own transcript");
        own.sha256 = "a".repeat(64);
        let mut foreign = completed_recording_for_user("foreign-user", "Foreign transcript");
        foreign.sha256 = own.sha256.clone();

        assert!(validate_submission_owner(&own, "own-user").is_ok());
        let error = validate_submission_owner(&foreign, "own-user").unwrap_err();
        assert_eq!(error.kind(), ErrorKind::Conflict);
        assert_eq!(
            error.message(),
            "identical audio is already attributed to another user"
        );
    }

    #[test]
    fn handoff_errors_preserve_category_and_message() {
        let cases = [
            (
                HandoffError::InvalidInput("invalid".into()),
                ErrorKind::InvalidInput,
                "invalid",
            ),
            (
                HandoffError::NotFound("missing".into()),
                ErrorKind::NotFound,
                "missing",
            ),
            (
                HandoffError::Conflict("conflict".into()),
                ErrorKind::Conflict,
                "conflict",
            ),
            (
                HandoffError::Internal("raw storage message".into()),
                ErrorKind::Internal,
                "raw storage message",
            ),
        ];

        for (source, kind, message) in cases {
            let error = handoff_error(source);
            assert_eq!(error.kind(), kind);
            assert_eq!(error.message(), message);
        }
    }

    #[test]
    fn blocking_audio_work_does_not_occupy_the_async_runtime_thread() {
        use std::{
            sync::{
                Arc,
                atomic::{AtomicBool, Ordering},
            },
            time::{Duration, Instant},
        };

        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_time()
            .build()
            .unwrap();
        runtime.block_on(async {
            let started = Arc::new(AtomicBool::new(false));
            let worker_started = Arc::clone(&started);
            let work = tokio::spawn(run_blocking_audio(move || {
                worker_started.store(true, Ordering::Release);
                std::thread::sleep(Duration::from_millis(100));
                Ok::<_, kcode_audio_ingress::Error>(())
            }));
            while !started.load(Ordering::Acquire) {
                tokio::task::yield_now().await;
            }
            let before = Instant::now();
            tokio::time::sleep(Duration::from_millis(10)).await;
            assert!(before.elapsed() < Duration::from_millis(75));
            work.await.unwrap().unwrap();
        });
    }
}