kcode-audio-session-ingress 0.2.2

#![forbid(unsafe_code)]

use std::collections::{HashMap, HashSet};

use chrono::{DateTime, Utc};
use kcode_audio_ingress::{
    AudioIngress, AudioInput, ConfirmationState, CorrectionPacket, ErrorKind as AudioErrorKind,
    RecordingConfirmation, RecordingState, RecordingStatus,
};
use kcode_audio_transcript_plan::{
    Error as PlanError, TranscriptPiece, TranscriptPlan, parse_audio_piece_id,
};
use kcode_session_history::{
    NewIngressSession, RetryIngress as HistoryRetryIngress, SessionHistory, SessionRecord,
    chatend::SessionKind,
};
use serde_json::Value;
use uuid::Uuid;

/// 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 updated_at: String,
    pub status: String,
    pub transcription_model: String,
    pub reconciliation_model: String,
    pub reconciliation_reasoning: String,
    pub transcription_status: Option<Value>,
    pub attempt_count: i64,
    pub next_attempt_at: Option<String>,
    pub last_error: Option<String>,
    pub speaker_review: Option<SpeakerReview>,
    pub transcript_piece_count: usize,
    pub completed_piece_count: usize,
}

/// Bounded review summary for one classifier-aware recording.
#[derive(Clone, Debug)]
pub struct SpeakerReview {
    pub clean: bool,
    pub confirmation_state: ConfirmationState,
    pub observation_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 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,
    history: SessionHistory,
    user_id: String,
    effective_context_tokens: u64,
}

impl Coordinator {
    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"));
        }
        validate_effective_context(config.effective_context_tokens)?;
        Ok(Self {
            audio,
            history,
            user_id: config.user_id,
            effective_context_tokens: config.effective_context_tokens,
        })
    }

    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 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)?;
        validate_submission_owner(&recording_status, &self.user_id)?;

        let histories = self.history.list().await.map_err(history_error)?;
        let projection =
            ingress_projection(&recording_status, &histories, self.effective_context_tokens)?;
        Ok(RecordingSubmission {
            recording: Recording::from_status(recording_status, &projection),
            deduplicated: submission.deduplicated,
        })
    }

    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_belongs_to(recording, &self.user_id))
            .map(|recording| {
                let projection =
                    ingress_projection(&recording, &histories, self.effective_context_tokens)?;
                Ok(Recording::from_status(recording, &projection))
            })
            .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> {
        let recording_status = 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)?;
        let histories = self.history.list().await.map_err(history_error)?;
        let projection =
            ingress_projection(&recording_status, &histories, self.effective_context_tokens)?;
        let final_transcript = match &recording_status.state {
            RecordingState::Complete { transcript } => Some(transcript.clone()),
            _ => None,
        };
        let correction_packet = recording_status.correction_packet.clone();
        let recording = Recording::from_status(recording_status, &projection);
        Ok(RecordingHistory {
            recording,
            final_transcript,
            correction_packet,
            pieces: projection.pieces,
        })
    }

    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: RecordingConfirmation,
    ) -> Result<CorrectionPacket, Error> {
        let recording_id = confirmation.recording_id;
        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)?;
        let histories = self.history.list().await.map_err(history_error)?;

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

        let packet = self
            .audio
            .confirm_speakers(confirmation)
            .map_err(audio_error)?;
        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)?;
        synchronize_recordings(
            &self.history,
            std::slice::from_ref(&recording),
            &self.user_id,
            self.effective_context_tokens,
        )
        .await?;
        Ok(packet)
    }

    pub async fn retry_ingress(&self, input: RetryIngress) -> Result<SessionRecord, Error> {
        let recordings = self.audio.status().map_err(audio_error)?.recordings;
        retry_ingress_exact(
            &self.history,
            &recordings,
            &self.user_id,
            self.effective_context_tokens,
            input,
        )
        .await
    }

    pub async fn synchronize_completed_transcripts(&self) -> Result<(), Error> {
        let recordings = self.audio.status().map_err(audio_error)?.recordings;
        synchronize_recordings(
            &self.history,
            &recordings,
            &self.user_id,
            self.effective_context_tokens,
        )
        .await
    }
}

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 confirmation_matches(packet: &CorrectionPacket, confirmation: &RecordingConfirmation) -> bool {
    if packet.confirmation_state != ConfirmationState::Confirmed {
        return false;
    }
    let existing = packet
        .chunks
        .iter()
        .flat_map(|chunk| &chunk.observations)
        .filter_map(|observation| {
            observation
                .confirmed_full_name
                .as_deref()
                .map(|name| (&observation.observation_key, name))
        })
        .collect::<HashMap<_, _>>();
    existing.len() == confirmation.observations.len()
        && confirmation.observations.iter().all(|observation| {
            existing.get(&observation.observation_key).copied()
                == Some(observation.confirmed_full_name.trim())
        })
}

fn recording_has_ingress(recording_id: Uuid, histories: &[SessionRecord]) -> bool {
    let prefix = format!("audio:{recording_id}:");
    histories.iter().any(|record| {
        ingress_source_id(record).is_some_and(|source_id| source_id.starts_with(&prefix))
    })
}

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

impl Recording {
    fn from_status(recording: RecordingStatus, projection: &IngressProjection) -> Self {
        let speaker_review = recording
            .correction_packet
            .as_ref()
            .map(|packet| SpeakerReview {
                clean: packet.clean,
                confirmation_state: packet.confirmation_state,
                observation_count: packet
                    .chunks
                    .iter()
                    .map(|chunk| chunk.observations.len())
                    .sum(),
            });
        let awaiting_speaker_review = speaker_review
            .as_ref()
            .is_some_and(|review| review.confirmation_state != ConfirmationState::Confirmed);
        let (mut status, transcription_status, attempt_count, last_error) = match recording.state {
            RecordingState::Queued => ("uploaded".into(), None, 0, None),
            RecordingState::Processing { attempt, progress } => (
                processing_stage(&progress).into(),
                serde_json::to_value(progress).ok(),
                i64::from(attempt),
                None,
            ),
            RecordingState::Complete { .. } if awaiting_speaker_review => {
                ("speaker_review".into(), None, 0, None)
            }
            RecordingState::Complete { .. } => ("ready_for_ingress".into(), None, 0, None),
            RecordingState::Failed {
                attempts, error, ..
            } => ("failed".into(), None, i64::from(attempts), Some(error)),
        };
        if !projection.pieces.is_empty() {
            status = if projection
                .pieces
                .iter()
                .all(|piece| piece.phase == "complete")
            {
                "complete".into()
            } else 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 {
                "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(),
            updated_at: recording.received_at.to_rfc3339(),
            status,
            transcription_model: recording.transcription_model,
            reconciliation_model: recording.reconciliation_model,
            reconciliation_reasoning: recording.reconciliation_reasoning,
            transcription_status,
            attempt_count,
            next_attempt_at: None,
            last_error,
            speaker_review,
            transcript_piece_count: projection.pieces.len(),
            completed_piece_count,
        }
    }
}

impl IngressPiece {
    fn from_record(
        recording: &RecordingStatus,
        piece_index: u32,
        piece_count: u32,
        transcript_text: String,
        estimated_tokens: u64,
        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,
            piece_count,
            transcript_text,
            estimated_tokens,
            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 retry_ingress_exact(
    history: &SessionHistory,
    recordings: &[RecordingStatus],
    user_id: &str,
    effective_context_tokens: u64,
    input: RetryIngress,
) -> Result<SessionRecord, Error> {
    let current = history.get(&input.piece_id).await.map_err(history_error)?;
    let source_id = ingress_source_id(&current).ok_or_else(Error::not_found)?;
    let (recording_id, piece_index) =
        parse_audio_piece_id(source_id).ok_or_else(Error::not_found)?;
    let recording = recordings
        .iter()
        .find(|recording| {
            recording.id == recording_id
                && recording_belongs_to(recording, user_id)
                && matches!(&recording.state, RecordingState::Complete { .. })
        })
        .ok_or_else(Error::not_found)?;

    let plan = transcript_plan(recording, effective_context_tokens)?;
    let piece = plan
        .pieces()
        .get(piece_index as usize)
        .filter(|piece| piece.index == piece_index && piece.id == source_id)
        .ok_or_else(|| {
            Error::conflict(
                "transcript piece no longer matches the authoritative audio segmentation",
            )
        })?;
    validate_persisted_record(recording, piece, &current)?;

    if input
        .state
        .as_ref()
        .is_some_and(|state| state != &current.state)
    {
        return Err(Error::conflict(
            "retry state does not match the current retained Session History state",
        ));
    }

    history
        .retry_ingress(
            &input.piece_id,
            HistoryRetryIngress {
                expected_version: input.expected_version,
                state: current.state,
            },
        )
        .await
        .map_err(history_error)
}

async fn synchronize_recordings(
    history: &SessionHistory,
    recordings: &[RecordingStatus],
    user_id: &str,
    effective_context_tokens: u64,
) -> Result<(), Error> {
    let histories = history.list().await.map_err(history_error)?;
    let mut existing = histories
        .iter()
        .filter_map(ingress_source_id)
        .map(str::to_owned)
        .collect::<HashSet<_>>();

    for recording in recordings {
        if !recording_belongs_to(recording, user_id)
            || !matches!(&recording.state, RecordingState::Complete { .. })
            || !speaker_labels_authorized(recording)
        {
            continue;
        }

        let persisted = persisted_audio_records(recording.id, &histories)?;
        let persisted_count = persisted_piece_count(&persisted)?;
        if persisted_count.is_some_and(|count| persisted.len() == count as usize) {
            continue;
        }

        let plan = transcript_plan(recording, effective_context_tokens)?;
        if persisted_count.is_some_and(|count| count != plan.pieces().len() as u32) {
            continue;
        }

        for (index, summary) in &persisted {
            let Some(piece) = plan.pieces().get(*index as usize) else {
                return Err(Error::conflict(
                    "persisted audio piece is outside the authoritative segmentation",
                ));
            };
            let current = history.get(&summary.id).await.map_err(history_error)?;
            validate_persisted_record(recording, piece, &current)?;
        }

        for piece in plan.pieces() {
            if existing.contains(&piece.id) {
                continue;
            }
            let created = history
                .enqueue_ingress(NewIngressSession {
                    idempotency_id: piece.id.clone(),
                    started_at: recording.recorded_at.to_rfc3339(),
                    source_session_type: "audio".into(),
                    kind: SessionKind::AudioIngress,
                    effective_context_tokens,
                    text: piece.formatted_text(recording).map_err(plan_error)?,
                    metadata: piece.metadata(recording),
                })
                .await
                .map_err(history_error)?;
            if !created.created {
                validate_persisted_record(recording, piece, &created.value)?;
            }
            existing.insert(piece.id.clone());
        }
    }
    Ok(())
}

fn speaker_labels_authorized(recording: &RecordingStatus) -> bool {
    recording
        .correction_packet
        .as_ref()
        .is_none_or(|packet| packet.confirmation_state == ConfirmationState::Confirmed)
}

fn ingress_projection(
    recording: &RecordingStatus,
    histories: &[SessionRecord],
    effective_context_tokens: u64,
) -> Result<IngressProjection, Error> {
    if !matches!(&recording.state, RecordingState::Complete { .. }) {
        return Ok(IngressProjection::default());
    }
    let persisted = persisted_audio_records(recording.id, histories)?;
    if persisted.is_empty() {
        return Ok(IngressProjection::default());
    }
    let persisted_count = persisted_piece_count(&persisted)?
        .unwrap_or_else(|| persisted.last().map_or(0, |(index, _)| index + 1));
    let current = transcript_plan(recording, effective_context_tokens)?;
    let current_matches = current.pieces().len() == persisted_count as usize;
    let whole = if persisted_count == 1 && !current_matches {
        Some(transcript_plan(recording, u64::MAX)?)
    } else {
        None
    };

    let mut pieces = Vec::with_capacity(persisted.len());
    for (index, record) in persisted {
        let planned = if persisted_count == 1 {
            if current_matches {
                current.pieces().first()
            } else {
                whole.as_ref().and_then(|plan| plan.pieces().first())
            }
        } else if current_matches {
            current.pieces().get(index as usize)
        } else {
            None
        };
        let (text, estimated_tokens) = planned.map_or_else(
            || (String::new(), 0),
            |piece| (piece.text.clone(), piece.estimated_tokens),
        );
        pieces.push(IngressPiece::from_record(
            recording,
            index,
            persisted_count,
            text,
            estimated_tokens,
            record,
        ));
    }
    Ok(IngressProjection { pieces })
}

fn persisted_audio_records(
    recording_id: Uuid,
    histories: &[SessionRecord],
) -> Result<Vec<(u32, &SessionRecord)>, Error> {
    let prefix = format!("audio:{recording_id}:");
    let mut records = Vec::new();
    let mut indexes = HashSet::new();
    for record in histories {
        let Some(source_id) = ingress_source_id(record) else {
            continue;
        };
        let Some(suffix) = source_id.strip_prefix(&prefix) else {
            continue;
        };
        let Some((_, index)) = parse_audio_piece_id(source_id) else {
            let message = if suffix.parse::<u32>().is_err() {
                "persisted audio piece has an invalid deterministic identity"
            } else {
                "persisted audio piece has a noncanonical deterministic identity"
            };
            return Err(Error::conflict(message));
        };
        if !indexes.insert(index) {
            return Err(Error::conflict(
                "multiple persisted audio pieces claim the same deterministic identity",
            ));
        }
        records.push((index, record));
    }
    records.sort_by_key(|(index, _)| *index);
    Ok(records)
}

fn persisted_piece_count(persisted: &[(u32, &SessionRecord)]) -> Result<Option<u32>, Error> {
    let mut declared_count = None;
    for (id_index, record) in persisted {
        let Some(metadata) = record
            .state
            .pointer("/ingressSource/metadata")
            .and_then(Value::as_object)
        else {
            continue;
        };
        if let Some(index) = metadata.get("pieceIndex").and_then(Value::as_u64) {
            let index = u32::try_from(index)
                .map_err(|_| Error::conflict("persisted audio piece index exceeds u32"))?;
            if index != *id_index {
                return Err(Error::conflict(
                    "persisted audio piece index conflicts with its deterministic identity",
                ));
            }
        }
        let Some(count) = metadata.get("pieceCount").and_then(Value::as_u64) else {
            continue;
        };
        let count = u32::try_from(count)
            .map_err(|_| Error::conflict("persisted audio piece count exceeds u32"))?;
        if count == 0 || *id_index >= count {
            return Err(Error::conflict(
                "persisted audio piece count conflicts with its deterministic identity",
            ));
        }
        if declared_count
            .replace(count)
            .is_some_and(|prior| prior != count)
        {
            return Err(Error::conflict(
                "persisted audio pieces disagree about their piece count",
            ));
        }
    }
    Ok(declared_count)
}

fn validate_persisted_record(
    recording: &RecordingStatus,
    piece: &TranscriptPiece,
    record: &SessionRecord,
) -> Result<(), Error> {
    let source_id = ingress_source_id(record).ok_or_else(|| {
        Error::conflict("replayed audio ingress omitted its deterministic identity")
    })?;
    if source_id != piece.id {
        return Err(Error::conflict(
            "replayed audio ingress returned a different deterministic identity",
        ));
    }
    if record.state.get("sessionType").and_then(Value::as_str) != Some("audio")
        || record
            .state
            .pointer("/chatendMetadata/kind")
            .and_then(Value::as_str)
            != Some("audio_ingress")
    {
        return Err(Error::conflict(
            "persisted transcript piece is not an authoritative audio ingress session",
        ));
    }
    if record.state.pointer("/ingressSource/metadata") != Some(&piece.metadata(recording)) {
        return Err(Error::conflict(
            "persisted transcript piece metadata no longer matches the authoritative audio piece",
        ));
    }
    let expected_text = piece.formatted_text(recording).map_err(plan_error)?;
    if record
        .state
        .pointer("/transcript/0/content")
        .and_then(Value::as_str)
        != Some(expected_text.as_str())
    {
        return Err(Error::conflict(
            "persisted transcript piece text no longer matches the authoritative audio piece",
        ));
    }
    Ok(())
}

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

fn transcript_plan(
    recording: &RecordingStatus,
    effective_context_tokens: u64,
) -> Result<TranscriptPlan, Error> {
    TranscriptPlan::new(recording, effective_context_tokens).map_err(plan_error)
}

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

fn plan_error(error: PlanError) -> Error {
    match error {
        PlanError::InvalidInput(message) => Error::invalid(message),
        PlanError::Conflict(message) => Error::conflict(message),
        PlanError::Internal(message) => Error::internal(message),
    }
}

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 chunks_complete = status
        .steps
        .iter()
        .filter(|entry| {
            matches!(
                entry.step,
                kcode_audio_ingress::Step::TranscribeChunk { .. }
            )
        })
        .all(|entry| entry.state == kcode_audio_ingress::StepState::Completed);
    if chunks_complete {
        let analyses_complete = status
            .steps
            .iter()
            .filter(|entry| matches!(entry.step, kcode_audio_ingress::Step::ParseChunk { .. }))
            .all(|entry| entry.state == kcode_audio_ingress::StepState::Completed);
        if !analyses_complete {
            return "analyzing_speakers";
        }
        let training_active = status.steps.iter().any(|entry| {
            entry.step == kcode_audio_ingress::Step::TrainIdentities
                && matches!(
                    entry.state,
                    kcode_audio_ingress::StepState::Running
                        | kcode_audio_ingress::StepState::Retrying
                )
        });
        if training_active {
            "training_speakers"
        } else {
            "reconciling"
        }
    } else {
        "transcribing"
    }
}

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 serde_json::json;

    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"),
            provider_cost_compatibility: None,
        })
        .unwrap()
    }

    fn completed_recording(transcript: impl Into<String>) -> RecordingStatus {
        completed_recording_for_user("user", transcript)
    }

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

    fn with_speaker_packet(
        mut recording: RecordingStatus,
        state: ConfirmationState,
        confirmed_name: Option<&str>,
    ) -> RecordingStatus {
        recording.correction_packet = Some(CorrectionPacket {
            recording_id: recording.id,
            user_id: recording.user_id.clone(),
            sha256: recording.sha256.clone(),
            original_filename: recording.original_filename.clone(),
            size_bytes: recording.size_bytes,
            recorded_at: recording.recorded_at,
            clean: true,
            chunk_count: 1,
            chunks: vec![kcode_audio_ingress::CorrectionChunk {
                chunk_index: 0,
                chunk_count: 1,
                audio_start_ms: 0,
                audio_end_ms: 1_000,
                raw_gemini_response: "raw".into(),
                parsed: kcode_audio_ingress::ParsedChunk {
                    utterances: Vec::new(),
                    notes: Vec::new(),
                    clip_valid: true,
                    clip_validity_reason: None,
                    speakers: Vec::new(),
                },
                observations: vec![kcode_audio_ingress::CorrectionObservation {
                    local_label: "Speaker A".into(),
                    speaker_ordinal: 0,
                    observation_key: kcode_audio_ingress::ObservationKey {
                        object_id: format!(
                            "kcode-audio-ingress/recording/{}/chunk/0",
                            recording.id
                        ),
                        piece_index: 0,
                    },
                    candidate: Some(kcode_audio_ingress::CandidateMapping {
                        full_name: "Classifier Candidate".into(),
                        cost: 1.0,
                        confidence: 3.0,
                        runner_up_full_name: None,
                        runner_up_cost: None,
                        background_population_cost: 4.0,
                    }),
                    identified_full_name: Some("Classifier Candidate".into()),
                    confirmed_full_name: confirmed_name.map(str::to_owned),
                }],
                clean: true,
            }],
            confirmation_state: state,
        });
        recording
    }

    fn planned_piece_id(
        recording: &RecordingStatus,
        effective_context_tokens: u64,
        index: usize,
    ) -> String {
        TranscriptPlan::new(recording, effective_context_tokens)
            .unwrap()
            .pieces()[index]
            .id
            .clone()
    }

    async fn fail_piece(history: &SessionHistory, record: SessionRecord) -> SessionRecord {
        let started = history
            .start_ingress(
                &record.id,
                kcode_session_history::StartIngress {
                    expected_version: record.version,
                    provenance_id: format!("test:{}", record.id),
                },
            )
            .await
            .unwrap();
        history
            .fail_ingress(
                &record.id,
                kcode_session_history::IngressFailure {
                    expected_version: started.version,
                    stage: "test".into(),
                    code: Some("input_too_large".into()),
                    message: "terminal test failure".into(),
                    rounds_used: None,
                    context_tokens: None,
                    context_window_tokens: None,
                },
            )
            .await
            .unwrap()
    }

    async fn synchronized_failed_piece(
        history: &SessionHistory,
        recording: &RecordingStatus,
        user_id: &str,
        effective_context_tokens: u64,
    ) -> SessionRecord {
        synchronize_recordings(
            history,
            std::slice::from_ref(recording),
            user_id,
            effective_context_tokens,
        )
        .await
        .unwrap();
        let source_id = planned_piece_id(recording, effective_context_tokens, 0);
        let record = history
            .list()
            .await
            .unwrap()
            .into_iter()
            .find(|record| ingress_source_id(record) == Some(source_id.as_str()))
            .unwrap();
        fail_piece(history, record).await
    }

    #[test]
    fn shared_status_is_filtered_to_the_configured_user() {
        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();
        let recordings = [own.clone(), foreign.clone()];

        let visible = recordings
            .iter()
            .filter(|recording| recording_belongs_to(recording, "own-user"))
            .collect::<Vec<_>>();
        assert_eq!(visible.len(), 1);
        assert_eq!(visible[0].id, own.id);
        assert_eq!(
            validate_submission_owner(&foreign, "own-user")
                .unwrap_err()
                .kind(),
            ErrorKind::Conflict
        );
    }

    #[test]
    fn confirmed_label_retries_must_match_the_complete_mapping() {
        let recording = with_speaker_packet(
            completed_recording("Transcript"),
            ConfirmationState::Confirmed,
            Some("Human Choice"),
        );
        let packet = recording.correction_packet.as_ref().unwrap();
        let key = packet.chunks[0].observations[0].observation_key.clone();
        let matching = RecordingConfirmation {
            recording_id: recording.id,
            observations: vec![kcode_audio_ingress::ObservationConfirmation {
                observation_key: key.clone(),
                confirmed_full_name: " Human Choice ".into(),
            }],
        };
        let conflicting = RecordingConfirmation {
            recording_id: recording.id,
            observations: vec![kcode_audio_ingress::ObservationConfirmation {
                observation_key: key,
                confirmed_full_name: "Different Choice".into(),
            }],
        };
        assert!(confirmation_matches(packet, &matching));
        assert!(!confirmation_matches(packet, &conflicting));
    }

    #[tokio::test]
    async fn classifier_aware_recordings_wait_for_human_review() {
        let root = TestRoot::new();
        let history = history(root.path());
        let recording = with_speaker_packet(
            completed_recording("Transcript"),
            ConfirmationState::AutomaticallyTrained,
            None,
        );
        synchronize_recordings(&history, &[recording], "user", 400)
            .await
            .unwrap();
        assert!(history.list().await.unwrap().is_empty());
    }

    #[tokio::test]
    async fn confirmed_labels_accompany_the_transcript() {
        let root = TestRoot::new();
        let history = history(root.path());
        let recording = with_speaker_packet(
            completed_recording("Transcript"),
            ConfirmationState::Confirmed,
            Some("Human Choice"),
        );
        synchronize_recordings(&history, std::slice::from_ref(&recording), "user", 400)
            .await
            .unwrap();
        let records = history.list().await.unwrap();
        assert_eq!(records.len(), 1);
        let state = serde_json::to_string(&records[0].state).unwrap();
        assert!(state.contains("Human-confirmed speaker-label data"));
        assert!(state.contains("Human Choice"));
        assert!(state.contains("Classifier Candidate"));
    }

    #[tokio::test]
    async fn bulk_synchronization_creates_no_foreign_pieces() {
        let root = TestRoot::new();
        let history = history(root.path());
        let own = completed_recording_for_user("own-user", "Own transcript");
        let foreign = completed_recording_for_user("foreign-user", "Foreign transcript");

        synchronize_recordings(&history, &[own.clone(), foreign.clone()], "own-user", 400)
            .await
            .unwrap();

        let records = history.list().await.unwrap();
        let own_source = planned_piece_id(&own, 400, 0);
        let foreign_source = planned_piece_id(&foreign, 400, 0);
        assert_eq!(records.len(), 1);
        assert_eq!(ingress_source_id(&records[0]), Some(own_source.as_str()));
        assert!(
            records
                .iter()
                .all(|record| ingress_source_id(record) != Some(foreign_source.as_str()))
        );
    }

    #[tokio::test]
    async fn synchronization_is_idempotent() {
        let root = TestRoot::new();
        let history = history(root.path());
        let recording = completed_recording("a".repeat(801));
        for _ in 0..2 {
            synchronize_recordings(&history, std::slice::from_ref(&recording), "user", 400)
                .await
                .unwrap();
        }
        assert_eq!(history.list().await.unwrap().len(), 3);
    }

    #[tokio::test]
    async fn context_changes_leave_complete_accepted_set_untouched() {
        let root = TestRoot::new();
        let history = history(root.path());
        let recording = completed_recording("a".repeat(801));
        synchronize_recordings(&history, std::slice::from_ref(&recording), "user", 400)
            .await
            .unwrap();
        let before = history
            .list()
            .await
            .unwrap()
            .iter()
            .filter_map(ingress_source_id)
            .map(str::to_owned)
            .collect::<HashSet<_>>();
        synchronize_recordings(&history, std::slice::from_ref(&recording), "user", 800)
            .await
            .unwrap();
        let after = history
            .list()
            .await
            .unwrap()
            .iter()
            .filter_map(ingress_source_id)
            .map(str::to_owned)
            .collect::<HashSet<_>>();
        assert_eq!(before, after);
    }

    #[tokio::test]
    async fn exact_current_piece_retry_accepts_equal_optional_state() {
        let root = TestRoot::new();
        let history = history(root.path());
        let recording = completed_recording("Transcript");
        let failed = synchronized_failed_piece(&history, &recording, "user", 400).await;

        let retried = retry_ingress_exact(
            &history,
            std::slice::from_ref(&recording),
            "user",
            400,
            RetryIngress {
                piece_id: failed.id,
                expected_version: failed.version,
                state: Some(failed.state),
            },
        )
        .await
        .unwrap();
        assert_eq!(retried.phase, "ingress_pending");
    }

    #[tokio::test]
    async fn retry_without_optional_state_remains_compatible() {
        let root = TestRoot::new();
        let history = history(root.path());
        let recording = completed_recording("Transcript");
        let failed = synchronized_failed_piece(&history, &recording, "user", 400).await;

        let retried = retry_ingress_exact(
            &history,
            std::slice::from_ref(&recording),
            "user",
            400,
            RetryIngress {
                piece_id: failed.id,
                expected_version: failed.version,
                state: None,
            },
        )
        .await
        .unwrap();
        assert_eq!(retried.phase, "ingress_pending");
    }

    #[tokio::test]
    async fn changed_retry_state_conflicts_without_mutation() {
        let root = TestRoot::new();
        let history = history(root.path());
        let recording = completed_recording("Transcript");
        let failed = synchronized_failed_piece(&history, &recording, "user", 400).await;
        let mut changed = failed.state.clone();
        changed["pendingTurn"] = json!(true);

        let error = retry_ingress_exact(
            &history,
            std::slice::from_ref(&recording),
            "user",
            400,
            RetryIngress {
                piece_id: failed.id.clone(),
                expected_version: failed.version,
                state: Some(changed),
            },
        )
        .await
        .unwrap_err();
        assert_eq!(error.kind(), ErrorKind::Conflict);
        assert_eq!(
            history.get(&failed.id).await.unwrap().version,
            failed.version
        );
    }

    #[tokio::test]
    async fn unrelated_session_history_id_cannot_retry() {
        let root = TestRoot::new();
        let history = history(root.path());
        let recording = completed_recording("Transcript");
        let unrelated = history
            .start(kcode_session_history::StartSession {
                idempotency_id: "unrelated".into(),
                started_at: recording.recorded_at.to_rfc3339(),
                session_type: "conversation".into(),
                duration_minutes: None,
                custom_prompt: None,
            })
            .await
            .unwrap()
            .value;

        let error = retry_ingress_exact(
            &history,
            std::slice::from_ref(&recording),
            "user",
            400,
            RetryIngress {
                piece_id: unrelated.id.clone(),
                expected_version: unrelated.version,
                state: None,
            },
        )
        .await
        .unwrap_err();
        assert_eq!(error.kind(), ErrorKind::NotFound);
        assert_eq!(
            history.get(&unrelated.id).await.unwrap().version,
            unrelated.version
        );
    }

    #[tokio::test]
    async fn malformed_audio_looking_id_cannot_retry() {
        let root = TestRoot::new();
        let history = history(root.path());
        let recording = completed_recording("Transcript");
        let malformed = history
            .enqueue_ingress(NewIngressSession {
                idempotency_id: "audio:not-a-uuid:0".into(),
                started_at: recording.recorded_at.to_rfc3339(),
                source_session_type: "audio".into(),
                kind: SessionKind::AudioIngress,
                effective_context_tokens: 400,
                text: "Malformed source".into(),
                metadata: json!({}),
            })
            .await
            .unwrap()
            .value;

        let error = retry_ingress_exact(
            &history,
            std::slice::from_ref(&recording),
            "user",
            400,
            RetryIngress {
                piece_id: malformed.id.clone(),
                expected_version: malformed.version,
                state: None,
            },
        )
        .await
        .unwrap_err();
        assert_eq!(error.kind(), ErrorKind::NotFound);
        assert_eq!(
            history.get(&malformed.id).await.unwrap().version,
            malformed.version
        );
    }

    #[tokio::test]
    async fn foreign_audio_piece_cannot_retry_or_mutate() {
        let root = TestRoot::new();
        let history = history(root.path());
        let own = completed_recording_for_user("own-user", "Own transcript");
        let foreign = completed_recording_for_user("foreign-user", "Foreign transcript");
        let failed = synchronized_failed_piece(&history, &foreign, "foreign-user", 400).await;

        let error = retry_ingress_exact(
            &history,
            &[own, foreign],
            "own-user",
            400,
            RetryIngress {
                piece_id: failed.id.clone(),
                expected_version: failed.version,
                state: None,
            },
        )
        .await
        .unwrap_err();
        assert_eq!(error.kind(), ErrorKind::NotFound);
        let unchanged = history.get(&failed.id).await.unwrap();
        assert_eq!(unchanged.phase, "ingress_failed");
        assert_eq!(unchanged.version, failed.version);
    }

    #[tokio::test]
    async fn retry_rejects_context_and_transcript_drift() {
        let root = TestRoot::new();
        let history = history(root.path());
        let recording = completed_recording("a".repeat(801));
        let failed = synchronized_failed_piece(&history, &recording, "user", 400).await;

        let context_error = retry_ingress_exact(
            &history,
            std::slice::from_ref(&recording),
            "user",
            800,
            RetryIngress {
                piece_id: failed.id.clone(),
                expected_version: failed.version,
                state: None,
            },
        )
        .await
        .unwrap_err();
        assert_eq!(context_error.kind(), ErrorKind::Conflict);

        let mut changed = recording.clone();
        changed.state = RecordingState::Complete {
            transcript: format!("b{}", "a".repeat(800)),
        };
        let transcript_error = retry_ingress_exact(
            &history,
            std::slice::from_ref(&changed),
            "user",
            400,
            RetryIngress {
                piece_id: failed.id.clone(),
                expected_version: failed.version,
                state: None,
            },
        )
        .await
        .unwrap_err();
        assert_eq!(transcript_error.kind(), ErrorKind::Conflict);
        assert_eq!(
            history.get(&failed.id).await.unwrap().version,
            failed.version
        );
    }
}