kcode-audio-session-ingress 0.2.0

#![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_session_history::{
    NewIngressSession, RetryIngress as HistoryRetryIngress, SessionHistory, SessionRecord,
    chatend::SessionKind,
};
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
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 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>,
    /// Current classifier-aware human-review state, absent for legacy recordings.
    pub speaker_review: Option<SpeakerReview>,
    /// Total number of pieces already represented in Session History.
    pub transcript_piece_count: usize,
    /// Number of represented pieces whose Session History ingress is complete.
    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 replacement state retained for wire compatibility with the
    /// pre-extraction Kennedy endpoint.
    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.",
        )
    }

    /// 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 AudioIngress can read its current state.
    pub fn health(&self) -> Result<(), Error> {
        self.audio.status().map_err(audio_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)?;
        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 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()
            .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 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 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)
            .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 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,
        })
    }

    /// Gives one failed recording a fresh processing budget.
    pub fn retry_recording(&self, recording_id: Uuid) -> Result<(), Error> {
        self.audio.retry(recording_id).map_err(audio_error)
    }

    /// Applies the complete human-confirmed speaker mapping and immediately
    /// attempts the now-authorized Session History handoff.
    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)
            .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.effective_context_tokens,
                self.maximum_piece_characters,
            )
            .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)
            .ok_or_else(Error::not_found)?;
        synchronize_recordings(
            &self.history,
            std::slice::from_ref(&recording),
            self.effective_context_tokens,
            self.maximum_piece_characters,
        )
        .await?;
        Ok(packet)
    }

    /// Retries one Session History ingress record.
    pub async fn retry_ingress(&self, input: RetryIngress) -> Result<SessionRecord, Error> {
        let current = self
            .history
            .get(&input.piece_id)
            .await
            .map_err(history_error)?;
        self.history
            .retry_ingress(
                &input.piece_id,
                HistoryRetryIngress {
                    expected_version: input.expected_version,
                    state: input.state.unwrap_or(current.state),
                },
            )
            .await
            .map_err(history_error)
    }

    /// Submits every missing 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()
            .collect::<Vec<_>>();
        synchronize_recordings(
            &self.history,
            &recordings,
            self.effective_context_tokens,
            self.maximum_piece_characters,
        )
        .await
    }
}

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)]
struct PieceSpec {
    id: String,
    index: u32,
    count: u32,
    text: String,
}

#[derive(Clone, Debug, Eq, PartialEq)]
struct SegmentationParameters {
    version: u64,
    effective_context_tokens: u64,
    maximum_piece_characters: usize,
    transcript_sha256: String,
    fingerprint: String,
}

#[derive(Debug)]
struct SegmentationPlan {
    parameters: SegmentationParameters,
    specs: Vec<PieceSpec>,
}

#[derive(Debug)]
struct PersistedPieceIdentity {
    index: u32,
    count: u32,
    piece_characters: usize,
    piece_text_sha256: String,
    ingress_text_sha256: String,
    parameters: SegmentationParameters,
}

#[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, 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)?;
    let mut existing = histories
        .iter()
        .filter_map(ingress_source_id)
        .map(str::to_owned)
        .collect::<HashSet<_>>();
    for recording in recordings {
        if !matches!(&recording.state, RecordingState::Complete { .. })
            || !speaker_labels_authorized(recording)
        {
            continue;
        }
        let plan = segmentation_plan(
            recording,
            &histories,
            effective_context_tokens,
            maximum_piece_characters,
        )?;
        for spec in &plan.specs {
            if existing.contains(&spec.id) {
                continue;
            }
            let text = format_ingress_piece(recording, spec)?;
            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: plan.parameters.effective_context_tokens,
                    text: text.clone(),
                    metadata: audio_piece_metadata(recording, spec, &plan.parameters, &text),
                })
                .await
                .map_err(history_error)?;
            if !created.created {
                validate_persisted_record(recording, &plan.parameters, spec, &created.value)?;
            }
            existing.insert(spec.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,
    maximum_piece_characters: usize,
) -> Result<IngressProjection, Error> {
    if !matches!(&recording.state, RecordingState::Complete { .. }) {
        return Ok(IngressProjection::default());
    }
    let plan = segmentation_plan(
        recording,
        histories,
        effective_context_tokens,
        maximum_piece_characters,
    )?;
    let mut pieces = Vec::with_capacity(plan.specs.len());
    for spec in &plan.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 { pieces })
}

fn segmentation_plan(
    recording: &RecordingStatus,
    histories: &[SessionRecord],
    effective_context_tokens: u64,
    current_maximum_piece_characters: usize,
) -> Result<SegmentationPlan, Error> {
    let persisted = persisted_audio_records(recording.id, histories)?;
    if persisted.is_empty() {
        return new_segmentation_plan(
            recording,
            effective_context_tokens,
            current_maximum_piece_characters,
        );
    }

    let mut identities = Vec::with_capacity(persisted.len());
    for (index, record) in persisted {
        identities.push((record, persisted_piece_identity(recording, index, record)?));
    }
    let parameters = identities[0].1.parameters.clone();
    if identities
        .iter()
        .any(|(_, identity)| identity.parameters != parameters)
    {
        return Err(Error::conflict(
            "persisted audio pieces disagree about authoritative segmentation",
        ));
    }
    let expected_maximum = maximum_piece_characters(parameters.effective_context_tokens)
        .map_err(|_| Error::conflict("persisted audio segmentation parameters are invalid"))?;
    if expected_maximum != parameters.maximum_piece_characters {
        return Err(Error::conflict(
            "persisted audio segmentation parameters are inconsistent",
        ));
    }
    if parameters.version != SEGMENTATION_VERSION
        || parameters.fingerprint != segmentation_fingerprint(&parameters)
    {
        return Err(Error::conflict(
            "persisted audio segmentation authority is unsupported or invalid",
        ));
    }

    let transcript = completed_transcript(recording)?;
    if sha256_text(transcript.trim()) != parameters.transcript_sha256 {
        return Err(Error::conflict(
            "completed audio transcript conflicts with persisted ingress identity",
        ));
    }
    let specs = transcript_piece_specs(recording, parameters.maximum_piece_characters)?;
    for (record, identity) in &identities {
        let spec = specs.get(identity.index as usize).ok_or_else(|| {
            Error::conflict("persisted audio piece index is outside authoritative segmentation")
        })?;
        validate_persisted_identity(recording, &parameters, spec, identity)?;
        if ingress_source_id(record) != Some(spec.id.as_str()) {
            return Err(Error::conflict(
                "persisted audio piece identity does not match its segmentation index",
            ));
        }
    }
    Ok(SegmentationPlan { parameters, specs })
}

fn new_segmentation_plan(
    recording: &RecordingStatus,
    effective_context_tokens: u64,
    current_maximum_piece_characters: usize,
) -> Result<SegmentationPlan, Error> {
    if maximum_piece_characters(effective_context_tokens)? != current_maximum_piece_characters {
        return Err(Error::internal(
            "audio ingress context and piece limit are inconsistent",
        ));
    }
    let transcript = completed_transcript(recording)?;
    let mut parameters = SegmentationParameters {
        version: SEGMENTATION_VERSION,
        effective_context_tokens,
        maximum_piece_characters: current_maximum_piece_characters,
        transcript_sha256: sha256_text(transcript.trim()),
        fingerprint: String::new(),
    };
    parameters.fingerprint = segmentation_fingerprint(&parameters);
    let specs = transcript_piece_specs(recording, current_maximum_piece_characters)?;
    Ok(SegmentationPlan { parameters, specs })
}

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 index = suffix.parse::<u32>().map_err(|_| {
            Error::conflict("persisted audio piece has an invalid deterministic identity")
        })?;
        if audio_piece_id(recording_id, index) != source_id {
            return Err(Error::conflict(
                "persisted audio piece has a noncanonical deterministic identity",
            ));
        }
        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_identity(
    recording: &RecordingStatus,
    id_index: u32,
    record: &SessionRecord,
) -> Result<PersistedPieceIdentity, Error> {
    let metadata = record
        .state
        .pointer("/ingressSource/metadata")
        .and_then(Value::as_object)
        .ok_or_else(|| {
            Error::conflict("persisted audio piece lacks authoritative segmentation metadata")
        })?;
    let recording_id = recording.id.to_string();
    let source_created_at = recording.recorded_at.to_rfc3339();
    if 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("originalFilename").and_then(Value::as_str)
            != Some(recording.original_filename.as_str())
        || metadata.get("sizeBytes").and_then(Value::as_u64) != Some(recording.size_bytes)
        || metadata.get("sourceCreatedAt").and_then(Value::as_str)
            != Some(source_created_at.as_str())
    {
        return Err(Error::conflict(
            "persisted audio piece metadata conflicts with its recording",
        ));
    }

    let version = required_metadata_u64(metadata, "segmentationVersion")?;
    let effective_context_tokens = required_metadata_u64(metadata, "effectiveContextTokens")?;
    let maximum_piece_characters =
        usize::try_from(required_metadata_u64(metadata, "maximumPieceCharacters")?)
            .map_err(|_| Error::conflict("persisted audio piece limit exceeds platform bounds"))?;
    let index = u32::try_from(required_metadata_u64(metadata, "pieceIndex")?)
        .map_err(|_| Error::conflict("persisted audio piece index exceeds u32"))?;
    let count = u32::try_from(required_metadata_u64(metadata, "pieceCount")?)
        .map_err(|_| Error::conflict("persisted audio piece count exceeds u32"))?;
    let piece_characters = usize::try_from(required_metadata_u64(metadata, "pieceCharacters")?)
        .map_err(|_| Error::conflict("persisted audio piece length exceeds platform bounds"))?;
    if index != id_index || count == 0 || index >= count {
        return Err(Error::conflict(
            "persisted audio piece index or count conflicts with its identity",
        ));
    }

    let transcript_sha256 = required_metadata_hash(metadata, "transcriptSha256")?;
    let fingerprint = required_metadata_hash(metadata, "segmentationFingerprint")?;
    let piece_text_sha256 = required_metadata_hash(metadata, "pieceTextSha256")?;
    let ingress_text_sha256 = required_metadata_hash(metadata, "ingressTextSha256")?;
    Ok(PersistedPieceIdentity {
        index,
        count,
        piece_characters,
        piece_text_sha256,
        ingress_text_sha256,
        parameters: SegmentationParameters {
            version,
            effective_context_tokens,
            maximum_piece_characters,
            transcript_sha256,
            fingerprint,
        },
    })
}

fn validate_persisted_record(
    recording: &RecordingStatus,
    parameters: &SegmentationParameters,
    spec: &PieceSpec,
    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 != spec.id {
        return Err(Error::conflict(
            "replayed audio ingress returned a different deterministic identity",
        ));
    }
    let identity = persisted_piece_identity(recording, spec.index, record)?;
    validate_persisted_identity(recording, parameters, spec, &identity)
}

fn validate_persisted_identity(
    recording: &RecordingStatus,
    parameters: &SegmentationParameters,
    spec: &PieceSpec,
    identity: &PersistedPieceIdentity,
) -> Result<(), Error> {
    if &identity.parameters != parameters
        || identity.index != spec.index
        || identity.count != spec.count
        || identity.piece_characters != spec.text.chars().count()
        || identity.piece_text_sha256 != sha256_text(&spec.text)
    {
        return Err(Error::conflict(
            "persisted audio piece conflicts with authoritative transcript segmentation",
        ));
    }
    let ingress_text = format_ingress_piece(recording, spec)?;
    if identity.ingress_text_sha256 != sha256_text(&ingress_text) {
        return Err(Error::conflict(
            "persisted audio ingress text conflicts with its authoritative identity",
        ));
    }
    Ok(())
}

fn required_metadata_u64(
    metadata: &serde_json::Map<String, Value>,
    field: &str,
) -> Result<u64, Error> {
    metadata.get(field).and_then(Value::as_u64).ok_or_else(|| {
        Error::conflict(format!(
            "persisted audio piece lacks valid {field} metadata"
        ))
    })
}

fn required_metadata_hash(
    metadata: &serde_json::Map<String, Value>,
    field: &str,
) -> Result<String, Error> {
    let value = metadata.get(field).and_then(Value::as_str).ok_or_else(|| {
        Error::conflict(format!(
            "persisted audio piece lacks valid {field} metadata"
        ))
    })?;
    if value.len() != 64
        || !value
            .bytes()
            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
    {
        return Err(Error::conflict(format!(
            "persisted audio piece has invalid {field} metadata"
        )));
    }
    Ok(value.to_owned())
}

fn completed_transcript(recording: &RecordingStatus) -> Result<&str, Error> {
    let RecordingState::Complete { transcript } = &recording.state else {
        return Err(Error::internal(
            "audio segmentation requested for an incomplete recording",
        ));
    };
    Ok(transcript)
}

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,
                text,
            })
        })
        .collect()
}

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

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,
    parameters: &SegmentationParameters,
    ingress_text: &str,
) -> 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(),
        "pieceIndex":spec.index,
        "pieceCount":spec.count,
        "pieceCharacters":spec.text.chars().count(),
        "segmentationVersion":parameters.version,
        "effectiveContextTokens":parameters.effective_context_tokens,
        "maximumPieceCharacters":parameters.maximum_piece_characters,
        "transcriptSha256":parameters.transcript_sha256,
        "segmentationFingerprint":parameters.fingerprint,
        "pieceTextSha256":sha256_text(&spec.text),
        "ingressTextSha256":sha256_text(ingress_text),
        "speakerConfirmationState":recording.correction_packet.as_ref().map(|packet| match packet.confirmation_state {
            ConfirmationState::Unconfirmed => "unconfirmed",
            ConfirmationState::AutomaticallyTrained => "automatically_trained",
            ConfirmationState::Confirmed => "confirmed",
        }),
    })
}

fn segmentation_fingerprint(parameters: &SegmentationParameters) -> String {
    sha256_text(&format!(
        "kcode-audio-session-ingress-segmentation\n{}\n{}\n{}\n{}",
        parameters.version,
        parameters.effective_context_tokens,
        parameters.maximum_piece_characters,
        parameters.transcript_sha256,
    ))
}

fn sha256_text(value: &str) -> String {
    format!("{:x}", Sha256::digest(value.as_bytes()))
}

fn format_ingress_piece(recording: &RecordingStatus, spec: &PieceSpec) -> Result<String, Error> {
    let speaker_mapping = confirmed_speaker_mapping(recording)?;
    Ok(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,
        speaker_mapping,
        spec.text,
    ))
}

fn confirmed_speaker_mapping(recording: &RecordingStatus) -> Result<String, Error> {
    let Some(packet) = recording.correction_packet.as_ref() else {
        return Ok(String::new());
    };
    if packet.confirmation_state != ConfirmationState::Confirmed {
        return Err(Error::conflict(
            "audio speaker labels require exact human confirmation before ingress",
        ));
    }
    let mut lines = Vec::new();
    for chunk in &packet.chunks {
        for observation in &chunk.observations {
            let confirmed = observation.confirmed_full_name.as_deref().ok_or_else(|| {
                Error::internal("confirmed audio packet omitted an observation label")
            })?;
            let local_label =
                serde_json::to_string(&observation.local_label).map_err(Error::internal)?;
            let confirmed = serde_json::to_string(confirmed).map_err(Error::internal)?;
            let candidate = observation
                .identified_full_name
                .as_deref()
                .map(serde_json::to_string)
                .transpose()
                .map_err(Error::internal)?
                .unwrap_or_else(|| "null".into());
            lines.push(format!(
                "- chunk {}/{}, source {:.3}-{:.3}s: localLabel={}, confirmedFullName={}, classifierCandidate={}",
                chunk.chunk_index + 1,
                chunk.chunk_count,
                chunk.audio_start_ms as f64 / 1_000.0,
                chunk.audio_end_ms as f64 / 1_000.0,
                local_label,
                confirmed,
                candidate,
            ));
        }
    }
    if lines.is_empty() {
        return Err(Error::internal(
            "confirmed audio packet contains no speaker observations",
        ));
    }
    Ok(format!(
        "\n\nHuman-confirmed speaker-label data (authoritative; do not infer alternatives):\n{}",
        lines.join("\n")
    ))
}

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

    #[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 confirmed_label_retries_must_match_the_complete_bound_mapping() {
        let recording = with_speaker_packet(
            completed_recording("Transcript"),
            ConfirmationState::Confirmed,
            Some("Human Choice"),
        );
        let packet = recording.correction_packet.as_ref().unwrap();
        let observation_key = packet.chunks[0].observations[0].observation_key.clone();
        let matching = RecordingConfirmation {
            recording_id: recording.id,
            observations: vec![kcode_audio_ingress::ObservationConfirmation {
                observation_key: observation_key.clone(),
                confirmed_full_name: " Human Choice ".into(),
            }],
        };
        let conflicting = RecordingConfirmation {
            recording_id: recording.id,
            observations: vec![kcode_audio_ingress::ObservationConfirmation {
                observation_key,
                confirmed_full_name: "Different Choice".into(),
            }],
        };

        assert!(confirmation_matches(packet, &matching));
        assert!(!confirmation_matches(packet, &conflicting));
    }

    #[test]
    fn ingress_exposes_the_complete_file_and_segmentation_contract() {
        let recording = completed_recording("Transcript");
        let plan = new_segmentation_plan(&recording, 400, 400).unwrap();
        let spec = &plan.specs[0];
        let text = format_ingress_piece(&recording, spec).unwrap();
        let metadata = audio_piece_metadata(&recording, spec, &plan.parameters, &text);
        assert_eq!(metadata["kind"], "audio-transcript");
        assert_eq!(metadata["recordingId"], recording.id.to_string());
        assert_eq!(metadata["sha256"], recording.sha256);
        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["sourceCreatedAt"],
            recording.recorded_at.to_rfc3339()
        );
        assert_eq!(metadata["pieceIndex"], 0);
        assert_eq!(metadata["pieceCount"], 1);
        assert_eq!(metadata["pieceCharacters"], 10);
        assert_eq!(metadata["segmentationVersion"], SEGMENTATION_VERSION);
        assert_eq!(metadata["effectiveContextTokens"], 400);
        assert_eq!(metadata["maximumPieceCharacters"], 400);
        assert_eq!(metadata["pieceTextSha256"], sha256_text("Transcript"));
        assert_eq!(metadata["ingressTextSha256"], sha256_text(&text));
        assert_eq!(metadata["speakerConfirmationState"], Value::Null);
        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_only_existing_pieces_before_worker_synchronization() {
        let recording = completed_recording("a".repeat(801));
        let projection = ingress_projection(&recording, &[], 400, 400).unwrap();
        assert!(projection.pieces.is_empty());
        let combined = Recording::from_status(recording, &projection);
        assert_eq!(combined.status, "ready_for_ingress");
        assert_eq!(combined.attempt_count, 0);
        assert_eq!(combined.transcript_piece_count, 0);
        assert_eq!(combined.completed_piece_count, 0);
    }

    #[tokio::test]
    async fn classifier_aware_recordings_wait_for_exact_human_review() {
        let root = TestRoot::new();
        let history = history(root.path());
        let recording = with_speaker_packet(
            completed_recording("Transcript"),
            ConfirmationState::AutomaticallyTrained,
            None,
        );

        let combined = Recording::from_status(recording.clone(), &IngressProjection::default());
        assert_eq!(combined.status, "speaker_review");
        assert_eq!(combined.speaker_review.unwrap().observation_count, 1);

        synchronize_recordings(&history, &[recording], 400, 400)
            .await
            .unwrap();
        assert!(history.list().await.unwrap().is_empty());
    }

    #[tokio::test]
    async fn confirmed_labels_authorize_ingress_and_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), 400, 400)
            .await
            .unwrap();
        let records = history.list().await.unwrap();
        assert_eq!(records.len(), 1);
        assert_eq!(
            records[0]
                .state
                .pointer("/ingressSource/metadata/speakerConfirmationState")
                .and_then(Value::as_str),
            Some("confirmed")
        );
        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 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)
                && record
                    .state
                    .pointer("/ingressSource/metadata/segmentationVersion")
                    .and_then(Value::as_u64)
                    == Some(SEGMENTATION_VERSION)
        }));
    }

    #[tokio::test]
    async fn ambiguous_legacy_metadata_is_rejected_instead_of_remapping_identity() {
        let root = TestRoot::new();
        let history = history(root.path());
        let recording = completed_recording("Transcript");
        let id = audio_piece_id(recording.id, 0);
        history
            .enqueue_ingress(NewIngressSession {
                idempotency_id: id,
                started_at: recording.recorded_at.to_rfc3339(),
                source_session_type: "audio".into(),
                kind: SessionKind::AudioIngress,
                effective_context_tokens: 400,
                text: "Previously ingressed".into(),
                metadata: json!({
                    "kind":"audio-transcript",
                    "recordingId":recording.id.to_string(),
                    "sha256":recording.sha256,
                    "originalFilename":recording.original_filename,
                    "sizeBytes":recording.size_bytes,
                    "sourceCreatedAt":recording.recorded_at.to_rfc3339(),
                    "pieceIndex":0,
                    "pieceCount":1,
                }),
            })
            .await
            .unwrap();

        let error = synchronize_recordings(&history, std::slice::from_ref(&recording), 400, 400)
            .await
            .unwrap_err();
        assert_eq!(error.kind(), ErrorKind::Conflict);
        assert_eq!(history.list().await.unwrap().len(), 1);
    }

    #[tokio::test]
    async fn context_changes_preserve_authoritative_piece_text_counts_and_ids() {
        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 before_records = history.list().await.unwrap();
        let before = ingress_projection(&recording, &before_records, 400, 400)
            .unwrap()
            .pieces
            .into_iter()
            .map(|piece| (piece.piece_index, piece.piece_count, piece.transcript_text))
            .collect::<Vec<_>>();
        let before_ids = before_records
            .iter()
            .filter_map(ingress_source_id)
            .map(str::to_owned)
            .collect::<HashSet<_>>();
        assert_eq!(before.len(), 3);

        synchronize_recordings(&history, std::slice::from_ref(&recording), 800, 800)
            .await
            .unwrap();
        let after_records = history.list().await.unwrap();
        let after = ingress_projection(&recording, &after_records, 800, 800)
            .unwrap()
            .pieces
            .into_iter()
            .map(|piece| (piece.piece_index, piece.piece_count, piece.transcript_text))
            .collect::<Vec<_>>();
        let after_ids = after_records
            .iter()
            .filter_map(ingress_source_id)
            .map(str::to_owned)
            .collect::<HashSet<_>>();

        assert_eq!(after, before);
        assert_eq!(after_ids, before_ids);
        assert_eq!(after_records.len(), 3);
        assert!(after_records.iter().all(|record| {
            record
                .state
                .pointer("/ingressSource/metadata/effectiveContextTokens")
                .and_then(Value::as_u64)
                == Some(400)
        }));
    }

    #[tokio::test]
    async fn partial_synchronization_uses_the_original_authoritative_segmentation() {
        let root = TestRoot::new();
        let history = history(root.path());
        let recording = completed_recording("a".repeat(801));
        let plan = new_segmentation_plan(&recording, 400, 400).unwrap();
        let first = &plan.specs[0];
        let first_text = format_ingress_piece(&recording, first).unwrap();
        history
            .enqueue_ingress(NewIngressSession {
                idempotency_id: first.id.clone(),
                started_at: recording.recorded_at.to_rfc3339(),
                source_session_type: "audio".into(),
                kind: SessionKind::AudioIngress,
                effective_context_tokens: plan.parameters.effective_context_tokens,
                text: first_text.clone(),
                metadata: audio_piece_metadata(&recording, first, &plan.parameters, &first_text),
            })
            .await
            .unwrap();

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

        let records = history.list().await.unwrap();
        let projection = ingress_projection(&recording, &records, 800, 800).unwrap();
        assert_eq!(projection.pieces.len(), 3);
        assert_eq!(
            projection
                .pieces
                .iter()
                .map(|piece| piece.transcript_text.clone())
                .collect::<Vec<_>>(),
            plan.specs
                .iter()
                .map(|spec| spec.text.clone())
                .collect::<Vec<_>>()
        );
        assert!(records.iter().all(|record| {
            record
                .state
                .pointer("/ingressSource/metadata/maximumPieceCharacters")
                .and_then(Value::as_u64)
                == Some(400)
                && record
                    .state
                    .pointer("/ingressSource/metadata/pieceCount")
                    .and_then(Value::as_u64)
                    == Some(3)
        }));
    }

    #[tokio::test]
    async fn changed_transcript_conflicts_with_persisted_piece_identity() {
        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 mut changed = recording.clone();
        changed.state = RecordingState::Complete {
            transcript: format!("{}b", "a".repeat(800)),
        };
        let error = synchronize_recordings(&history, std::slice::from_ref(&changed), 400, 400)
            .await
            .unwrap_err();

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

    #[test]
    fn complete_recording_projection_preserves_pre_extraction_fields() {
        let recording = completed_recording("Transcript");
        let received_at = recording.received_at.to_rfc3339();
        let combined = Recording::from_status(recording, &IngressProjection::default());

        assert_eq!(combined.received_at, received_at);
        assert_eq!(combined.updated_at, received_at);
        assert_eq!(combined.status, "ready_for_ingress");
        assert!(combined.transcription_status.is_none());
        assert_eq!(combined.attempt_count, 0);
        assert!(combined.next_attempt_at.is_none());
        assert!(combined.last_error.is_none());
        assert_eq!(combined.transcript_piece_count, 0);
    }
}