kcode_audio_session_ingress/

lib.rs

#![forbid(unsafe_code)]

use std::collections::HashSet;

use chrono::{DateTime, Utc};
use kcode_audio_ingress::{
    AudioIngress, AudioInput, ErrorKind as AudioErrorKind, RecordingState, RecordingStatus,
};
use kcode_session_history::{
    NewIngressSession, RetryIngress as HistoryRetryIngress, SessionHistory, SessionRecord,
    chatend::SessionKind,
};
use serde_json::{Value, json};
use uuid::Uuid;

const ESTIMATED_CHARACTERS_PER_TOKEN: u64 = 4;
const INGRESS_CONTEXT_DIVISOR: u64 = 4;
const SEGMENTATION_VERSION: u64 = 1;

/// 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,
}

/// Combined recording and memory-ingress status.
#[derive(Clone, Debug)]
pub struct Recording {
    pub id: Uuid,
    pub sha256: String,
    pub original_filename: String,
    pub content_type: &'static str,
    pub size_bytes: u64,
    pub source_created_at: String,
    pub received_at: String,
    pub status: String,
    pub transcription_model: String,
    pub reconciliation_model: String,
    pub reconciliation_reasoning: String,
    /// Current or exhausted attempt count when AudioIngress exposes it.
    pub attempt_count: Option<u8>,
    /// Total number of pieces in the current validated transcript segmentation.
    pub transcript_piece_count: usize,
    /// Number of current pieces whose Session History ingress is complete.
    pub completed_piece_count: usize,
}

/// One deterministic transcript piece and its Session History lifecycle.
#[derive(Clone, Debug)]
pub struct IngressPiece {
    pub id: String,
    pub recording_id: Uuid,
    pub sha256: String,
    pub original_filename: String,
    pub source_created_at: String,
    pub piece_index: u32,
    pub piece_count: u32,
    pub transcript_text: String,
    pub estimated_tokens: u64,
    pub phase: String,
    pub provenance_id: Option<String>,
    pub state: Value,
    pub version: i64,
    pub ingress_failure_count: i64,
    pub ingress_failures: Value,
    pub created_at: String,
    pub updated_at: String,
}

/// Detailed state for one recording.
#[derive(Clone, Debug)]
pub struct RecordingHistory {
    pub recording: Recording,
    pub final_transcript: Option<String>,
    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,
}

/// 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 not_found() -> Self {
        Self::new(
            ErrorKind::NotFound,
            "Audio recording or transcript piece not found.",
        )
    }

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

    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.",
        )
    }

    /// Returns the stable error category.
    pub fn kind(&self) -> ErrorKind {
        self.kind
    }

    /// Returns the sanitized actionable message.
    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,
    history: SessionHistory,
    user_id: String,
    effective_context_tokens: u64,
    maximum_piece_characters: usize,
}

impl Coordinator {
    /// Constructs a coordinator over already-opened capability handles.
    pub fn new(
        audio: AudioIngress,
        history: SessionHistory,
        config: Config,
    ) -> Result<Self, Error> {
        if config.user_id.trim().is_empty() {
            return Err(Error::invalid("audio user ID must not be empty"));
        }
        let maximum_piece_characters = maximum_piece_characters(config.effective_context_tokens)?;
        Ok(Self {
            audio,
            history,
            user_id: config.user_id,
            effective_context_tokens: config.effective_context_tokens,
            maximum_piece_characters,
        })
    }

    /// Checks that both underlying capabilities can read their current state.
    pub fn health(&self) -> Result<(), Error> {
        self.audio.status().map_err(audio_error)?;
        self.history.health().map_err(history_error)?;
        Ok(())
    }

    /// Durably submits one admitted recording and returns its combined state.
    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 recording_status = self
            .audio
            .status()
            .map_err(audio_error)?
            .recordings
            .into_iter()
            .find(|recording| recording.id == submission.recording_id)
            .ok_or_else(Error::not_found)?;
        if recording_status.user_id != self.user_id {
            return Err(Error::conflict(
                "The submitted audio already belongs to a different configured user.",
            ));
        }
        let histories = self.history.list().await.map_err(history_error)?;
        let projection = ingress_projection(
            &recording_status,
            &histories,
            self.effective_context_tokens,
            self.maximum_piece_characters,
        )?;
        let recording = Recording::from_status(recording_status, &projection);
        Ok(RecordingSubmission {
            recording,
            deduplicated: submission.deduplicated,
        })
    }

    /// Returns configured-user recordings with their correlated ingress state.
    pub async fn recordings(&self) -> Result<Vec<Recording>, Error> {
        let histories = self.history.list().await.map_err(history_error)?;
        self.audio
            .status()
            .map_err(audio_error)?
            .recordings
            .into_iter()
            .filter(|recording| recording.user_id == self.user_id)
            .map(|recording| {
                let projection = ingress_projection(
                    &recording,
                    &histories,
                    self.effective_context_tokens,
                    self.maximum_piece_characters,
                )?;
                Ok(Recording::from_status(recording, &projection))
            })
            .collect()
    }

    /// Finds a configured-user recording by its SHA-256 digest.
    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)
    }

    /// Returns one configured-user recording and its correlated pieces.
    pub async fn recording_history(&self, recording_id: Uuid) -> Result<RecordingHistory, Error> {
        let recording_status = self
            .audio
            .status()
            .map_err(audio_error)?
            .recordings
            .into_iter()
            .find(|recording| recording.id == recording_id && recording.user_id == self.user_id)
            .ok_or_else(Error::not_found)?;
        let histories = self.history.list().await.map_err(history_error)?;
        let projection = ingress_projection(
            &recording_status,
            &histories,
            self.effective_context_tokens,
            self.maximum_piece_characters,
        )?;
        let final_transcript = match &recording_status.state {
            RecordingState::Complete { transcript } => Some(transcript.clone()),
            _ => None,
        };
        let recording = Recording::from_status(recording_status, &projection);
        Ok(RecordingHistory {
            recording,
            final_transcript,
            pieces: projection.pieces,
        })
    }

    /// Gives one failed configured-user recording a fresh processing budget.
    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.user_id == self.user_id);
        if !owned {
            return Err(Error::not_found());
        }
        self.audio.retry(recording_id).map_err(audio_error)
    }

    /// Retries memory ingress only for a currently correlated audio piece.
    pub async fn retry_ingress(&self, input: RetryIngress) -> Result<SessionRecord, Error> {
        let current = self
            .history
            .get(&input.piece_id)
            .await
            .map_err(history_error)?;
        let recordings = self
            .audio
            .status()
            .map_err(audio_error)?
            .recordings
            .into_iter()
            .filter(|recording| recording.user_id == self.user_id)
            .collect::<Vec<_>>();
        validate_retry_target(
            &current,
            &recordings,
            self.effective_context_tokens,
            self.maximum_piece_characters,
        )?;
        self.history
            .retry_ingress(
                &input.piece_id,
                HistoryRetryIngress {
                    expected_version: input.expected_version,
                    state: current.state,
                },
            )
            .await
            .map_err(history_error)
    }

    /// Submits every missing configured-user transcript piece to Session History.
    pub async fn synchronize_completed_transcripts(&self) -> Result<(), Error> {
        let recordings = self
            .audio
            .status()
            .map_err(audio_error)?
            .recordings
            .into_iter()
            .filter(|recording| recording.user_id == self.user_id)
            .collect::<Vec<_>>();
        synchronize_recordings(
            &self.history,
            &recordings,
            self.effective_context_tokens,
            self.maximum_piece_characters,
        )
        .await
    }
}

#[derive(Debug)]
struct PieceSpec {
    id: String,
    index: u32,
    count: u32,
    text: String,
    fingerprint: String,
}

#[derive(Debug, Default)]
struct IngressProjection {
    expected_piece_count: usize,
    pieces: Vec<IngressPiece>,
}

impl Recording {
    fn from_status(recording: RecordingStatus, projection: &IngressProjection) -> Self {
        let (mut status, attempt_count) = match recording.state {
            RecordingState::Queued => ("uploaded".into(), Some(0)),
            RecordingState::Processing { attempt, progress } => {
                (processing_stage(&progress).into(), Some(attempt))
            }
            RecordingState::Complete { .. } => ("ready_for_ingress".into(), None),
            RecordingState::Failed { attempts, .. } => ("failed".into(), Some(attempts)),
        };
        if projection.expected_piece_count != 0 {
            status = if projection
                .pieces
                .iter()
                .any(|piece| piece.phase == "ingress_failed")
            {
                "ingress_failed".into()
            } else if projection
                .pieces
                .iter()
                .any(|piece| piece.phase == "ingress_in_progress")
            {
                "ingressing".into()
            } else if projection.pieces.len() == projection.expected_piece_count
                && projection
                    .pieces
                    .iter()
                    .all(|piece| piece.phase == "complete")
            {
                "complete".into()
            } else {
                "ready_for_ingress".into()
            };
        }
        let completed_piece_count = projection
            .pieces
            .iter()
            .filter(|piece| piece.phase == "complete")
            .count();
        Self {
            id: recording.id,
            sha256: recording.sha256,
            original_filename: recording.original_filename,
            content_type: "audio/wav",
            size_bytes: recording.size_bytes,
            source_created_at: recording.recorded_at.to_rfc3339(),
            received_at: recording.received_at.to_rfc3339(),
            status,
            transcription_model: recording.transcription_model,
            reconciliation_model: recording.reconciliation_model,
            reconciliation_reasoning: recording.reconciliation_reasoning,
            attempt_count,
            transcript_piece_count: projection.expected_piece_count,
            completed_piece_count,
        }
    }
}

impl IngressPiece {
    fn from_record(recording: &RecordingStatus, spec: &PieceSpec, record: &SessionRecord) -> Self {
        Self {
            id: record.id.clone(),
            recording_id: recording.id,
            sha256: recording.sha256.clone(),
            original_filename: recording.original_filename.clone(),
            source_created_at: recording.recorded_at.to_rfc3339(),
            piece_index: spec.index,
            piece_count: spec.count,
            estimated_tokens: estimate_tokens(&spec.text),
            transcript_text: spec.text.clone(),
            phase: record.phase.clone(),
            provenance_id: record.provenance_id.clone(),
            state: record.state.clone(),
            version: record.version,
            ingress_failure_count: record.ingress_failure_count,
            ingress_failures: record.ingress_failures.clone(),
            created_at: record.started_at.clone(),
            updated_at: record.updated_at.clone(),
        }
    }
}

async fn synchronize_recordings(
    history: &SessionHistory,
    recordings: &[RecordingStatus],
    effective_context_tokens: u64,
    maximum_piece_characters: usize,
) -> Result<(), Error> {
    let histories = history.list().await.map_err(history_error)?;
    for recording in recordings {
        if !matches!(&recording.state, RecordingState::Complete { .. }) {
            continue;
        }
        let specs = transcript_piece_specs(recording, maximum_piece_characters)?;
        validate_existing_piece_records(
            recording,
            &histories,
            &specs,
            effective_context_tokens,
            maximum_piece_characters,
        )?;
        let mut existing = histories
            .iter()
            .filter_map(ingress_source_id)
            .map(str::to_owned)
            .collect::<HashSet<_>>();
        for spec in specs {
            if existing.contains(&spec.id) {
                continue;
            }
            let created = history
                .enqueue_ingress(NewIngressSession {
                    idempotency_id: spec.id.clone(),
                    started_at: recording.recorded_at.to_rfc3339(),
                    source_session_type: "audio".into(),
                    kind: SessionKind::AudioIngress,
                    effective_context_tokens,
                    text: format_ingress_piece(recording, &spec),
                    metadata: audio_piece_metadata(
                        recording,
                        &spec,
                        effective_context_tokens,
                        maximum_piece_characters,
                    ),
                })
                .await
                .map_err(history_error)?;
            validate_piece_record(
                &created.value,
                recording,
                &spec,
                effective_context_tokens,
                maximum_piece_characters,
            )?;
            existing.insert(spec.id);
        }
    }
    Ok(())
}

fn ingress_projection(
    recording: &RecordingStatus,
    histories: &[SessionRecord],
    effective_context_tokens: u64,
    maximum_piece_characters: usize,
) -> Result<IngressProjection, Error> {
    if !matches!(&recording.state, RecordingState::Complete { .. }) {
        return Ok(IngressProjection::default());
    }
    let specs = transcript_piece_specs(recording, maximum_piece_characters)?;
    validate_existing_piece_records(
        recording,
        histories,
        &specs,
        effective_context_tokens,
        maximum_piece_characters,
    )?;
    let mut pieces = Vec::with_capacity(specs.len());
    for spec in &specs {
        let Some(record) = histories
            .iter()
            .find(|record| ingress_source_id(record) == Some(spec.id.as_str()))
        else {
            continue;
        };
        pieces.push(IngressPiece::from_record(recording, spec, record));
    }
    Ok(IngressProjection {
        expected_piece_count: specs.len(),
        pieces,
    })
}

fn transcript_piece_specs(
    recording: &RecordingStatus,
    maximum_piece_characters: usize,
) -> Result<Vec<PieceSpec>, Error> {
    let RecordingState::Complete { transcript } = &recording.state else {
        return Ok(Vec::new());
    };
    let pieces = split_transcript(transcript, maximum_piece_characters)?;
    let piece_count = u32::try_from(pieces.len())
        .map_err(|_| Error::internal("audio transcript contains too many pieces"))?;
    pieces
        .into_iter()
        .enumerate()
        .map(|(index, text)| {
            let piece_index = u32::try_from(index)
                .map_err(|_| Error::internal("audio transcript piece index exceeds u32"))?;
            Ok(PieceSpec {
                id: audio_piece_id(recording.id, piece_index),
                index: piece_index,
                count: piece_count,
                fingerprint: piece_fingerprint(&text),
                text,
            })
        })
        .collect()
}

fn validate_existing_piece_records(
    recording: &RecordingStatus,
    histories: &[SessionRecord],
    specs: &[PieceSpec],
    effective_context_tokens: u64,
    maximum_piece_characters: usize,
) -> Result<(), Error> {
    for record in histories {
        if let Some(spec) = ingress_source_id(record)
            .and_then(|source_id| specs.iter().find(|spec| spec.id == source_id))
        {
            validate_piece_record(
                record,
                recording,
                spec,
                effective_context_tokens,
                maximum_piece_characters,
            )?;
        } else if metadata_recording_id(record) == Some(recording.id) {
            return Err(segmentation_conflict());
        }
    }
    Ok(())
}

fn validate_retry_target(
    record: &SessionRecord,
    recordings: &[RecordingStatus],
    effective_context_tokens: u64,
    maximum_piece_characters: usize,
) -> Result<(), Error> {
    let Some(source_id) = ingress_source_id(record) else {
        return Err(Error::not_found());
    };
    for recording in recordings {
        if !matches!(&recording.state, RecordingState::Complete { .. }) {
            continue;
        }
        let specs = transcript_piece_specs(recording, maximum_piece_characters)?;
        if let Some(spec) = specs.iter().find(|spec| spec.id == source_id) {
            return validate_piece_record(
                record,
                recording,
                spec,
                effective_context_tokens,
                maximum_piece_characters,
            );
        }
        if metadata_recording_id(record) == Some(recording.id) {
            return Err(segmentation_conflict());
        }
    }
    Err(Error::not_found())
}

fn validate_piece_record(
    record: &SessionRecord,
    recording: &RecordingStatus,
    spec: &PieceSpec,
    effective_context_tokens: u64,
    maximum_piece_characters: usize,
) -> Result<(), Error> {
    let Some(metadata) = ingress_metadata(record) else {
        return Err(segmentation_conflict());
    };
    let maximum_piece_characters = u64::try_from(maximum_piece_characters)
        .map_err(|_| Error::internal("piece limit exceeds u64"))?;
    let piece_characters = u64::try_from(spec.text.chars().count())
        .map_err(|_| Error::internal("audio transcript piece exceeds u64 characters"))?;
    let recording_id = recording.id.to_string();
    let valid = ingress_source_id(record) == Some(spec.id.as_str())
        && metadata.get("kind").and_then(Value::as_str) == Some("audio-transcript")
        && metadata.get("recordingId").and_then(Value::as_str) == Some(recording_id.as_str())
        && metadata.get("sha256").and_then(Value::as_str) == Some(recording.sha256.as_str())
        && metadata.get("segmentationVersion").and_then(Value::as_u64)
            == Some(SEGMENTATION_VERSION)
        && metadata
            .get("effectiveContextTokens")
            .and_then(Value::as_u64)
            == Some(effective_context_tokens)
        && metadata
            .get("maximumPieceCharacters")
            .and_then(Value::as_u64)
            == Some(maximum_piece_characters)
        && metadata.get("pieceIndex").and_then(Value::as_u64) == Some(u64::from(spec.index))
        && metadata.get("pieceCount").and_then(Value::as_u64) == Some(u64::from(spec.count))
        && metadata.get("pieceCharacters").and_then(Value::as_u64) == Some(piece_characters)
        && metadata.get("pieceFingerprint").and_then(Value::as_str)
            == Some(spec.fingerprint.as_str());
    if valid {
        Ok(())
    } else {
        Err(segmentation_conflict())
    }
}

fn segmentation_conflict() -> Error {
    Error::conflict(
        "Existing Session History audio pieces do not match the configured transcript segmentation.",
    )
}

fn ingress_source_id(record: &SessionRecord) -> Option<&str> {
    record
        .state
        .pointer("/ingressSource/idempotencyId")
        .and_then(Value::as_str)
}

fn ingress_metadata(record: &SessionRecord) -> Option<&Value> {
    record.state.pointer("/ingressSource/metadata")
}

fn metadata_recording_id(record: &SessionRecord) -> Option<Uuid> {
    let metadata = ingress_metadata(record)?;
    if metadata.get("kind").and_then(Value::as_str) != Some("audio-transcript") {
        return None;
    }
    metadata
        .get("recordingId")
        .and_then(Value::as_str)
        .and_then(|value| Uuid::parse_str(value).ok())
}

fn audio_piece_id(recording_id: Uuid, piece_index: u32) -> String {
    format!("audio:{recording_id}:{piece_index}")
}

fn audio_piece_metadata(
    recording: &RecordingStatus,
    spec: &PieceSpec,
    effective_context_tokens: u64,
    maximum_piece_characters: usize,
) -> Value {
    json!({
        "kind":"audio-transcript",
        "recordingId":recording.id.to_string(),
        "sha256":recording.sha256,
        "originalFilename":recording.original_filename,
        "extension":file_name_extension(&recording.original_filename),
        "mimeType":"audio/wav",
        "sizeBytes":recording.size_bytes,
        "sourceCreatedAt":recording.recorded_at.to_rfc3339(),
        "segmentationVersion":SEGMENTATION_VERSION,
        "effectiveContextTokens":effective_context_tokens,
        "maximumPieceCharacters":maximum_piece_characters,
        "pieceIndex":spec.index,
        "pieceCount":spec.count,
        "pieceCharacters":spec.text.chars().count(),
        "pieceFingerprint":spec.fingerprint,
    })
}

fn format_ingress_piece(recording: &RecordingStatus, spec: &PieceSpec) -> String {
    format!(
        "Vnote final transcript piece\n\nRecording began: {}\nRecording SHA-256: {}\nOriginal filename: {}\nExtension: {}\nMIME type: audio/wav\nSize: {} bytes\nTranscript piece: {} of {}\n\n{}",
        recording.recorded_at.to_rfc3339(),
        recording.sha256,
        recording.original_filename,
        file_name_extension(&recording.original_filename),
        recording.size_bytes,
        spec.index + 1,
        spec.count,
        spec.text,
    )
}

fn piece_fingerprint(value: &str) -> String {
    let mut hash = 0xcbf29ce484222325_u64;
    for byte in value.as_bytes() {
        hash ^= u64::from(*byte);
        hash = hash.wrapping_mul(0x100000001b3);
    }
    format!("fnv1a64:{hash:016x}")
}

fn file_name_extension(file_name: &str) -> String {
    file_name
        .rsplit_once('.')
        .and_then(|(stem, extension)| {
            (!stem.is_empty() && !extension.is_empty()).then_some(extension)
        })
        .map(|extension| format!(".{extension}"))
        .unwrap_or_else(|| "(none)".into())
}

fn processing_stage(status: &kcode_audio_ingress::TranscriptionStatus) -> &'static str {
    let plan_complete = status.steps.iter().any(|entry| {
        entry.step == kcode_audio_ingress::Step::PlanChunks
            && entry.state == kcode_audio_ingress::StepState::Completed
    });
    if !plan_complete {
        return "chunking";
    }
    let mut saw_chunk = false;
    for entry in &status.steps {
        if matches!(
            entry.step,
            kcode_audio_ingress::Step::TranscribeChunk { .. }
        ) {
            saw_chunk = true;
            if entry.state != kcode_audio_ingress::StepState::Completed {
                return "transcribing";
            }
        }
    }
    if saw_chunk {
        "reconciling"
    } else {
        "transcribing"
    }
}

fn maximum_piece_characters(effective_context_tokens: u64) -> Result<usize, Error> {
    let piece_tokens = effective_context_tokens / INGRESS_CONTEXT_DIVISOR;
    if piece_tokens == 0 {
        return Err(Error::invalid(
            "effective ingress context must contain at least four tokens",
        ));
    }
    let characters = piece_tokens
        .checked_mul(ESTIMATED_CHARACTERS_PER_TOKEN)
        .ok_or_else(|| Error::invalid("effective ingress context is too large"))?;
    usize::try_from(characters)
        .map_err(|_| Error::invalid("effective ingress context exceeds platform limits"))
}

fn split_transcript(
    transcript: &str,
    maximum_piece_characters: usize,
) -> Result<Vec<String>, Error> {
    let mut remaining = transcript.trim();
    if remaining.is_empty() {
        return Err(Error::internal("completed audio transcript is empty"));
    }
    let mut pieces = Vec::new();
    while remaining.chars().count() > maximum_piece_characters {
        let cutoff = remaining
            .char_indices()
            .nth(maximum_piece_characters)
            .map(|(index, _)| index)
            .unwrap_or(remaining.len());
        let prefix = &remaining[..cutoff];
        let minimum = prefix
            .char_indices()
            .nth(maximum_piece_characters / 2)
            .map(|(index, _)| index)
            .unwrap_or(0);
        let boundary = prefix
            .rfind("\n\n")
            .filter(|index| *index >= minimum)
            .or_else(|| prefix.rfind('\n').filter(|index| *index >= minimum))
            .unwrap_or(cutoff);
        let piece = remaining[..boundary].trim();
        if piece.is_empty() {
            return Err(Error::internal("could not split audio transcript"));
        }
        pieces.push(piece.to_owned());
        remaining = remaining[boundary..].trim();
    }
    if !remaining.is_empty() {
        pieces.push(remaining.to_owned());
    }
    Ok(pieces)
}

fn estimate_tokens(value: &str) -> u64 {
    (value.chars().count() as u64).div_ceil(ESTIMATED_CHARACTERS_PER_TOKEN)
}

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),
    }
}

fn history_error(error: kcode_session_history::Error) -> Error {
    let kind = match error.kind {
        kcode_session_history::ErrorKind::InvalidInput => ErrorKind::InvalidInput,
        kcode_session_history::ErrorKind::NotFound => ErrorKind::NotFound,
        kcode_session_history::ErrorKind::Conflict => ErrorKind::Conflict,
        kcode_session_history::ErrorKind::Storage => ErrorKind::Internal,
    };
    Error::new(kind, error.message)
}

#[cfg(test)]
mod tests {
    use std::path::{Path, PathBuf};

    use super::*;

    struct TestRoot(PathBuf);

    impl TestRoot {
        fn new() -> Self {
            let path = std::env::temp_dir().join(format!(
                "kcode-audio-session-ingress-test-{}",
                Uuid::new_v4()
            ));
            std::fs::create_dir(&path).unwrap();
            Self(path)
        }

        fn path(&self) -> &Path {
            &self.0
        }
    }

    impl Drop for TestRoot {
        fn drop(&mut self) {
            let _ = std::fs::remove_dir_all(&self.0);
        }
    }

    fn history(root: &Path) -> SessionHistory {
        SessionHistory::open(kcode_session_history::Config {
            directory: root.join("active"),
            completed_list: root.join("completed.txt"),
        })
        .unwrap()
    }

    fn completed_recording(transcript: impl Into<String>) -> RecordingStatus {
        let now = Utc::now();
        RecordingStatus {
            id: Uuid::new_v4(),
            user_id: "user".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(),
            },
        }
    }

    #[test]
    fn transcript_piece_limit_is_one_quarter_of_effective_context() {
        let effective_context_tokens = 400;
        let maximum_characters = maximum_piece_characters(effective_context_tokens).unwrap();
        let pieces = split_transcript(&"a".repeat(801), maximum_characters).unwrap();
        assert_eq!(maximum_characters, 400);
        assert_eq!(pieces.len(), 3);
        assert!(pieces.iter().all(|piece| {
            estimate_tokens(piece) <= effective_context_tokens / INGRESS_CONTEXT_DIVISOR
        }));
    }

    #[test]
    fn transcript_splitting_prefers_a_late_paragraph_boundary() {
        let transcript = format!("{}\n\n{}", "a".repeat(250), "b".repeat(200));
        let pieces = split_transcript(&transcript, 400).unwrap();
        assert_eq!(pieces, vec!["a".repeat(250), "b".repeat(200)]);
    }

    #[test]
    fn transcript_splitting_counts_unicode_scalars_not_bytes() {
        let transcript = "😀".repeat(5);
        let pieces = split_transcript(&transcript, 2).unwrap();
        assert_eq!(pieces.concat(), transcript);
        assert!(pieces.iter().all(|piece| piece.chars().count() <= 2));
    }

    #[test]
    fn audio_piece_ids_are_stable_across_configuration() {
        let recording_id = Uuid::new_v4();
        assert_eq!(
            audio_piece_id(recording_id, 2),
            format!("audio:{recording_id}:2")
        );
    }

    #[test]
    fn ingress_exposes_the_complete_file_metadata_contract() {
        let recording = completed_recording("Transcript");
        let spec = transcript_piece_specs(&recording, 400).unwrap().remove(0);
        let metadata = audio_piece_metadata(&recording, &spec, 400, 400);
        assert_eq!(metadata["originalFilename"], "meeting.final.WAV");
        assert_eq!(metadata["extension"], ".WAV");
        assert_eq!(metadata["mimeType"], "audio/wav");
        assert_eq!(metadata["sizeBytes"], 42);
        assert_eq!(metadata["segmentationVersion"], SEGMENTATION_VERSION);
        assert_eq!(metadata["effectiveContextTokens"], 400);
        assert_eq!(metadata["maximumPieceCharacters"], 400);
        assert_eq!(
            metadata["pieceFingerprint"],
            piece_fingerprint("Transcript")
        );
        let text = format_ingress_piece(&recording, &spec);
        assert!(text.contains("Original filename: meeting.final.WAV"));
        assert!(text.contains("Extension: .WAV"));
        assert!(text.contains("MIME type: audio/wav"));
        assert!(text.contains("Size: 42 bytes"));
    }

    #[test]
    fn reads_report_expected_pieces_before_worker_synchronization() {
        let recording = completed_recording("a".repeat(801));
        let projection = ingress_projection(&recording, &[], 400, 400).unwrap();
        assert_eq!(projection.expected_piece_count, 3);
        assert!(projection.pieces.is_empty());
        let combined = Recording::from_status(recording, &projection);
        assert_eq!(combined.status, "ready_for_ingress");
        assert_eq!(combined.transcript_piece_count, 3);
        assert_eq!(combined.completed_piece_count, 0);
    }

    #[tokio::test]
    async fn synchronization_is_idempotent_and_uses_quarter_window_pieces() {
        let root = TestRoot::new();
        let history = history(root.path());
        let recording = completed_recording("a".repeat(801));

        synchronize_recordings(&history, std::slice::from_ref(&recording), 400, 400)
            .await
            .unwrap();
        synchronize_recordings(&history, std::slice::from_ref(&recording), 400, 400)
            .await
            .unwrap();

        let records = history.list().await.unwrap();
        assert_eq!(records.len(), 3);
        let ids = records
            .iter()
            .filter_map(ingress_source_id)
            .collect::<HashSet<_>>();
        assert_eq!(ids.len(), 3);
        assert!(ids.contains(format!("audio:{}:0", recording.id).as_str()));
        assert!(ids.contains(format!("audio:{}:1", recording.id).as_str()));
        assert!(ids.contains(format!("audio:{}:2", recording.id).as_str()));
        assert!(records.iter().all(|record| {
            record
                .state
                .pointer("/ingressSource/metadata/pieceCount")
                .and_then(Value::as_u64)
                == Some(3)
        }));
    }

    #[tokio::test]
    async fn changed_segmentation_conflicts_without_creating_new_work() {
        let root = TestRoot::new();
        let history = history(root.path());
        let recording = completed_recording("a".repeat(801));

        synchronize_recordings(&history, std::slice::from_ref(&recording), 400, 400)
            .await
            .unwrap();
        let error = synchronize_recordings(&history, std::slice::from_ref(&recording), 800, 800)
            .await
            .unwrap_err();

        assert_eq!(error.kind(), ErrorKind::Conflict);
        assert_eq!(history.list().await.unwrap().len(), 3);
    }

    #[tokio::test]
    async fn retry_validation_rejects_unrelated_history_records() {
        let root = TestRoot::new();
        let history = history(root.path());
        let unrelated = history
            .enqueue_ingress(NewIngressSession {
                idempotency_id: "unrelated".into(),
                started_at: Utc::now().to_rfc3339(),
                source_session_type: "other".into(),
                kind: SessionKind::AudioIngress,
                effective_context_tokens: 400,
                text: "Unrelated".into(),
                metadata: json!({"kind":"other"}),
            })
            .await
            .unwrap()
            .value;
        let recording = completed_recording("Transcript");

        let error = validate_retry_target(&unrelated, &[recording], 400, 400).unwrap_err();
        assert_eq!(error.kind(), ErrorKind::NotFound);
    }

    #[tokio::test]
    async fn retry_validation_accepts_only_current_correlated_pieces() {
        let root = TestRoot::new();
        let history = history(root.path());
        let recording = completed_recording("Transcript");
        synchronize_recordings(&history, std::slice::from_ref(&recording), 400, 400)
            .await
            .unwrap();
        let record = history.list().await.unwrap().remove(0);

        validate_retry_target(&record, &[recording], 400, 400).unwrap();
    }
}