kcode-audio-ingress 0.6.1

//! Durable, automatic audio transcription and speaker correction.
//!
//! [`AudioIngress`] accepts owned WAV bytes, persists them before returning,
//! and automatically transcribes, classifies, and reconciles them in the
//! background.

#![deny(missing_docs)]
#![forbid(unsafe_code)]

use std::{
    collections::{HashMap, HashSet},
    fs,
    path::{Path, PathBuf},
    sync::{Arc, Mutex, Weak},
    time::Duration,
};

use anyhow::{Context, ensure};
use chrono::{DateTime, Duration as ChronoDuration, Utc};
use kcode_speaker_system::SpeechClassifier;
use rusqlite::{Connection, OptionalExtension, params};
use serde::Serialize;
use serde_json::Value;
use sha2::{Digest, Sha256};
use tokio::io::AsyncWriteExt;
use uuid::Uuid;

pub use identity::{
    CLASSIFIER_MODEL, CLASSIFIER_PROMPT_VERSION, CLASSIFIER_PROVIDER, CLASSIFIER_SCHEMA_VERSION,
    CandidateMapping, ConfirmationState, CorrectionChunk, CorrectionObservation, CorrectionPacket,
    FeatureRow, ObservationConfirmation, ObservationKey, ParsedChunk, ParsedSpeaker,
    ParsedUtterance, RecordingConfirmation,
};
use identity::{
    ClassificationContext, apply_confirmations, classify_speakers, restore_packet_training,
    validate_parsed_chunk,
};
pub use transcribe::{
    AudioChunkCall, AudioChunkRequest, AudioTranscriber, IntelligenceError, IntelligenceFuture,
    JobState, RECONCILIATION_MODEL, RECONCILIATION_REASONING, Step, StepError, StepState,
    StepStatus, TRANSCRIPTION_MODEL, TextGenerationCall, TextGenerationRequest, TranscriptionJob,
    TranscriptionStatus,
};
use transcribe::{ChunkPlan, ChunkTranscript, PIECE_CACHE_REVISION, PieceCache, PieceSink};

mod identity;
mod transcribe;

const INITIAL_MIGRATION: &str = include_str!("../migrations/001_initial.sql");
const RELEASE_DEFERRED_INGRESS_MIGRATION: &str =
    include_str!("../migrations/002_release_deferred_ingress.sql");
const TRANSCRIPTION_STATUS_MIGRATION: &str =
    include_str!("../migrations/003_transcription_status.sql");
const RETRY_ROUNDED_WAV_INTERVALS_MIGRATION: &str =
    include_str!("../migrations/004_retry_rounded_wav_intervals.sql");
const UNIFIED_INGRESS_QUEUE_MIGRATION: &str =
    include_str!("../migrations/005_unified_ingress_queue.sql");
const STANDALONE_LIBRARY_MIGRATION: &str = include_str!("../migrations/006_standalone_library.sql");
const DURABLE_TRANSCRIPT_PIECES_MIGRATION: &str =
    include_str!("../migrations/007_durable_transcript_pieces.sql");
const UNIQUE_TRANSCRIPTION_ATTEMPTS_MIGRATION: &str =
    include_str!("../migrations/008_unique_transcription_attempts.sql");
const USAGE_USER_MIGRATION: &str = include_str!("../migrations/009_usage_user.sql");
const SPEAKER_CORRECTION_PACKETS_MIGRATION: &str =
    include_str!("../migrations/010_speaker_correction_packets.sql");

const DATABASE_FILENAME: &str = "state.sqlite3";
const ORIGINALS_DIRECTORY: &str = "originals";
const FAILURE_LIMIT: i64 = 5;
const RETRY_DELAY_SECONDS: i64 = 15;
const LATEST_SCHEMA_VERSION: i64 = 10;

const FRESH_SCHEMA: &str = r#"
CREATE TABLE audio_recordings (
    id TEXT PRIMARY KEY NOT NULL,
    user_id TEXT NOT NULL CHECK(length(user_id) > 0),
    sha256 TEXT NOT NULL UNIQUE CHECK(length(sha256) = 64),
    original_filename TEXT NOT NULL,
    content_type TEXT NOT NULL,
    size_bytes INTEGER NOT NULL CHECK(size_bytes >= 0),
    source_created_at TEXT NOT NULL,
    received_at TEXT NOT NULL,
    updated_at TEXT NOT NULL,
    original_relative_path TEXT NOT NULL,
    status TEXT NOT NULL CHECK(status IN (
        'uploaded', 'chunking', 'transcribing', 'reconciling',
        'ready_for_ingress', 'ingressing', 'ingress_failed', 'complete', 'failed'
    )),
    gemini_model TEXT NOT NULL,
    reconciliation_model TEXT NOT NULL,
    reconciliation_reasoning TEXT NOT NULL,
    final_transcript TEXT,
    attempt_count INTEGER NOT NULL DEFAULT 0 CHECK(attempt_count >= 0),
    next_attempt_at TEXT,
    last_error TEXT,
    transcription_status_json TEXT,
    failure_retryable INTEGER NOT NULL DEFAULT 1
        CHECK(failure_retryable IN (0, 1)),
    correction_packet_json TEXT
);

CREATE INDEX audio_recordings_work_queue
ON audio_recordings(status, next_attempt_at, received_at);

CREATE TABLE audio_transcript_pieces (
    recording_id TEXT NOT NULL REFERENCES audio_recordings(id) ON DELETE CASCADE,
    attempt_id TEXT NOT NULL CHECK(length(attempt_id) > 0),
    cache_revision TEXT NOT NULL CHECK(length(cache_revision) > 0),
    piece_index INTEGER NOT NULL CHECK(piece_index >= 0),
    piece_count INTEGER NOT NULL CHECK(piece_count > 0 AND piece_index < piece_count),
    audio_start_ms INTEGER NOT NULL CHECK(audio_start_ms >= 0),
    audio_end_ms INTEGER NOT NULL CHECK(audio_end_ms > audio_start_ms),
    transcript_json TEXT NOT NULL,
    raw_gemini_response TEXT NOT NULL,
    parsed_json TEXT NOT NULL,
    created_at TEXT NOT NULL,
    PRIMARY KEY(recording_id, attempt_id, piece_index)
);

CREATE INDEX audio_transcript_pieces_cache_lookup
ON audio_transcript_pieces(
    recording_id,
    cache_revision,
    piece_index,
    piece_count,
    audio_start_ms,
    audio_end_ms,
    created_at
);

PRAGMA user_version = 10;
"#;

/// Owned audio and its source metadata.
#[derive(Clone, Debug)]
pub struct AudioInput {
    /// Stable application user identifier charged for provider calls.
    pub user_id: String,
    /// Complete WAV bytes.
    pub bytes: Vec<u8>,
    /// Instant at which recording began.
    pub recorded_at: DateTime<Utc>,
    /// Original leaf filename, when known.
    pub original_filename: Option<String>,
}

/// Result of durably submitting audio.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
pub struct Submission {
    /// Stable recording identifier.
    pub recording_id: Uuid,
    /// Whether the same audio bytes were already known.
    pub deduplicated: bool,
}

/// Complete current library state.
#[derive(Clone, Debug, Serialize)]
pub struct Status {
    /// Recordings ordered by recording time, newest first.
    pub recordings: Vec<RecordingStatus>,
}

/// Current state, immutable identity, and optional completed correction packet.
#[derive(Clone, Debug, Serialize)]
pub struct RecordingStatus {
    /// Stable recording identifier.
    pub id: Uuid,
    /// Stable application user identifier charged for provider calls.
    pub user_id: String,
    /// Lowercase SHA-256 digest of the original bytes.
    pub sha256: String,
    /// Sanitized original filename.
    pub original_filename: String,
    /// Original byte length.
    pub size_bytes: u64,
    /// Instant at which recording began.
    pub recorded_at: DateTime<Utc>,
    /// Instant at which AudioIngress accepted the recording.
    pub received_at: DateTime<Utc>,
    /// Gemini speaker-analysis model attribution.
    pub transcription_model: String,
    /// GPT parsing and reconciliation model attribution.
    pub reconciliation_model: String,
    /// GPT parsing and reconciliation reasoning attribution.
    pub reconciliation_reasoning: String,
    /// Current processing state.
    pub state: RecordingState,
    /// Durable transport-neutral packet, present after successful completion.
    pub correction_packet: Option<CorrectionPacket>,
}

/// Automatic processing state of one recording.
#[derive(Clone, Debug, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum RecordingState {
    /// Persisted and waiting for automatic processing.
    Queued,
    /// An in-memory transcription attempt is active.
    Processing {
        /// One-based full-job attempt number.
        attempt: u8,
        /// Current dependency status without transcript or correction payloads.
        progress: TranscriptionStatus,
    },
    /// The canonical transcript and correction packet are durable.
    Complete {
        /// Canonical reconciled Markdown.
        transcript: String,
    },
    /// Automatic processing stopped.
    Failed {
        /// Attempts used in the current manual-attempt budget.
        attempts: u8,
        /// Concise diagnostic.
        error: String,
        /// Whether another attempt can reasonably succeed.
        retryable: bool,
    },
}

/// Stable library error category.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ErrorKind {
    /// Supplied data is invalid.
    InvalidInput,
    /// The requested recording does not exist.
    NotFound,
    /// The requested transition is not valid in the current state.
    Conflict,
    /// Persistence or internal processing failed unexpectedly.
    Internal,
}

/// Error returned by the AudioIngress API.
#[derive(Debug)]
pub struct Error {
    kind: ErrorKind,
    message: String,
}

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

    fn not_found() -> Self {
        Self {
            kind: ErrorKind::NotFound,
            message: "Audio recording not found.".into(),
        }
    }

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

    fn internal(error: impl std::fmt::Display) -> Self {
        tracing::error!(%error, "AudioIngress operation failed");
        Self {
            kind: ErrorKind::Internal,
            message: "An unexpected AudioIngress error occurred.".into(),
        }
    }

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

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

struct TemporaryUpload(PathBuf);

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

struct Inner {
    root: PathBuf,
    db: Arc<Mutex<Connection>>,
    classifier: Arc<SpeechClassifier>,
    transcriber: AudioTranscriber,
    jobs: Mutex<HashMap<Uuid, TranscriptionJob>>,
}

/// Cloneable handle to durable, automatically processed audio.
#[derive(Clone)]
pub struct AudioIngress {
    inner: Arc<Inner>,
}

impl AudioIngress {
    /// Opens the owned persistence root with a shared classifier and starts automatic processing.
    ///
    /// Ingress state is stored at `<root>/state.sqlite3`, originals remain
    /// below `<root>/originals/`, and classifier state is owned by the
    /// caller-supplied shared classifier.
    pub async fn open(
        persistence_root: impl AsRef<Path>,
        transcriber: AudioTranscriber,
        classifier: Arc<SpeechClassifier>,
    ) -> Result<Self, Error> {
        let root = persistence_root.as_ref().to_path_buf();
        ensure_private_directory(&root).map_err(Error::internal)?;
        ensure_private_directory(&root.join(ORIGINALS_DIRECTORY)).map_err(Error::internal)?;

        let connection = Connection::open(root.join(DATABASE_FILENAME)).map_err(Error::internal)?;
        connection
            .execute_batch(
                "PRAGMA foreign_keys=ON; PRAGMA journal_mode=WAL; PRAGMA busy_timeout=15000;",
            )
            .map_err(Error::internal)?;
        apply_migrations(&connection).map_err(Error::internal)?;
        recover_interrupted_attempts(&connection).map_err(Error::internal)?;

        let inner = Arc::new(Inner {
            root,
            db: Arc::new(Mutex::new(connection)),
            classifier,
            transcriber,
            jobs: Mutex::new(HashMap::new()),
        });
        tokio::spawn(worker_loop(Arc::downgrade(&inner)));
        Ok(Self { inner })
    }

    /// Durably accepts complete WAV bytes and schedules automatic processing.
    pub async fn submit(&self, input: AudioInput) -> Result<Submission, Error> {
        if input.user_id.trim().is_empty() || input.user_id.chars().count() > 256 {
            return Err(Error::invalid(
                "User ID must contain between 1 and 256 characters.",
            ));
        }
        if input.bytes.is_empty() {
            return Err(Error::invalid("Audio bytes must not be empty."));
        }
        let size_bytes = i64::try_from(input.bytes.len())
            .map_err(|_| Error::invalid("Audio is too large for this platform."))?;
        let sha256 = format!("{:x}", Sha256::digest(&input.bytes));
        if let Some(id) = self.recording_id_by_sha(&sha256)? {
            return Ok(Submission {
                recording_id: id,
                deduplicated: true,
            });
        }

        let upload_id = Uuid::new_v4();
        let temporary = TemporaryUpload(self.inner.root.join(format!(".upload-{upload_id}.tmp")));
        let mut file = tokio::fs::OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(&temporary.0)
            .await
            .map_err(Error::internal)?;
        set_private_file(&temporary.0).map_err(Error::internal)?;
        file.write_all(&input.bytes)
            .await
            .map_err(Error::internal)?;
        file.sync_all().await.map_err(Error::internal)?;
        drop(file);

        let relative_path = format!("{ORIGINALS_DIRECTORY}/{sha256}.wav");
        let final_path = self.inner.root.join(&relative_path);
        if final_path.exists() {
            tokio::fs::remove_file(&temporary.0)
                .await
                .map_err(Error::internal)?;
        } else {
            tokio::fs::rename(&temporary.0, &final_path)
                .await
                .map_err(Error::internal)?;
        }
        set_private_file(&final_path).map_err(Error::internal)?;
        sync_file(&final_path).map_err(Error::internal)?;
        sync_directory(final_path.parent().unwrap_or(&self.inner.root)).map_err(Error::internal)?;
        sync_directory(&self.inner.root).map_err(Error::internal)?;

        let id = Uuid::new_v4();
        let now = Utc::now().to_rfc3339();
        let filename = safe_filename(input.original_filename.as_deref());
        let insert = {
            let db = self.inner.db.lock().map_err(Error::internal)?;
            db.execute(
                "INSERT INTO audio_recordings(
                    id,user_id,sha256,original_filename,content_type,size_bytes,
                    source_created_at,received_at,updated_at,original_relative_path,
                    status,gemini_model,reconciliation_model,reconciliation_reasoning
                 ) VALUES(?1,?2,?3,?4,'audio/wav',?5,?6,?7,?7,?8,'uploaded',?9,?10,?11)",
                params![
                    id.to_string(),
                    input.user_id,
                    sha256,
                    filename,
                    size_bytes,
                    input.recorded_at.to_rfc3339(),
                    now,
                    relative_path,
                    TRANSCRIPTION_MODEL,
                    RECONCILIATION_MODEL,
                    RECONCILIATION_REASONING,
                ],
            )
        };
        if let Err(error) = insert {
            if let Some(existing) = self.recording_id_by_sha(&sha256)? {
                return Ok(Submission {
                    recording_id: existing,
                    deduplicated: true,
                });
            }
            return Err(Error::internal(error));
        }
        tracing::info!(recording_id=%id, %sha256, bytes=size_bytes, "Durably accepted audio");
        Ok(Submission {
            recording_id: id,
            deduplicated: false,
        })
    }

    /// Returns all current recording states and completed correction packets.
    pub fn status(&self) -> Result<Status, Error> {
        let db = self.inner.db.lock().map_err(Error::internal)?;
        let mut statement = db
            .prepare(
                "SELECT id,user_id,sha256,original_filename,size_bytes,source_created_at,received_at,
                        status,gemini_model,reconciliation_model,reconciliation_reasoning,
                        attempt_count,last_error,failure_retryable,transcription_status_json,
                        final_transcript,correction_packet_json
                 FROM audio_recordings
                 ORDER BY datetime(source_created_at) DESC,datetime(received_at) DESC,id DESC",
            )
            .map_err(Error::internal)?;
        let recordings = statement
            .query_map([], row_recording_status)
            .map_err(Error::internal)?
            .collect::<Result<Vec<_>, _>>()
            .map_err(Error::internal)?;
        Ok(Status { recordings })
    }

    /// Gives a failed recording a fresh five-attempt processing budget.
    pub fn retry(&self, recording_id: Uuid) -> Result<(), Error> {
        let db = self.inner.db.lock().map_err(Error::internal)?;
        let changed = db
            .execute(
                "UPDATE audio_recordings
                 SET status='uploaded',attempt_count=0,next_attempt_at=NULL,last_error=NULL,
                     transcription_status_json=NULL,failure_retryable=1,updated_at=?1
                 WHERE id=?2 AND status='failed'",
                params![Utc::now().to_rfc3339(), recording_id.to_string()],
            )
            .map_err(Error::internal)?;
        if changed == 1 {
            return Ok(());
        }
        let exists = db
            .query_row(
                "SELECT 1 FROM audio_recordings WHERE id=?1",
                [recording_id.to_string()],
                |row| row.get::<_, i64>(0),
            )
            .optional()
            .map_err(Error::internal)?
            .is_some();
        if exists {
            Err(Error::conflict("Only a failed recording can be retried."))
        } else {
            Err(Error::not_found())
        }
    }

    /// Applies exact observation-level full-name confirmations to one completed recording.
    ///
    /// The confirmation must cover every deterministic observation key in the
    /// packet exactly once. Classifier updates are idempotent. The corrected
    /// packet is committed before this method returns and is also visible
    /// through [`AudioIngress::status`].
    pub fn confirm_speakers(
        &self,
        confirmation: RecordingConfirmation,
    ) -> Result<CorrectionPacket, Error> {
        let recording_id = confirmation.recording_id;
        let db = self.inner.db.lock().map_err(Error::internal)?;
        let stored = db
            .query_row(
                "SELECT status,correction_packet_json
                 FROM audio_recordings WHERE id=?1",
                [recording_id.to_string()],
                |row| Ok((row.get::<_, String>(0)?, row.get::<_, Option<String>>(1)?)),
            )
            .optional()
            .map_err(Error::internal)?;
        let Some((status, packet_json)) = stored else {
            return Err(Error::not_found());
        };
        if !matches!(
            status.as_str(),
            "ready_for_ingress" | "ingressing" | "ingress_failed" | "complete"
        ) {
            return Err(Error::conflict(
                "Speaker confirmation requires a completed recording.",
            ));
        }
        let packet_json = packet_json
            .ok_or_else(|| Error::conflict("The completed recording has no correction packet."))?;
        let mut stored = StoredCorrectionPacket::decode(&packet_json).map_err(Error::internal)?;
        identity::validate_confirmation_coverage(&stored.packet, &confirmation)
            .map_err(Error::invalid)?;
        let legacy_observation_keys = stored.legacy_observation_keys().map_err(Error::internal)?;

        let previous = stored.packet.clone();
        apply_confirmations(
            &self.inner.classifier,
            &mut stored.packet,
            &confirmation,
            &legacy_observation_keys,
        )
        .map_err(Error::internal)?;
        let updated_json = stored.encode().map_err(Error::internal)?;
        if let Err(error) = db.execute(
            "UPDATE audio_recordings
             SET correction_packet_json=?1,updated_at=?2
             WHERE id=?3",
            params![
                updated_json,
                Utc::now().to_rfc3339(),
                recording_id.to_string()
            ],
        ) {
            let rollback_errors = restore_packet_training(
                &self.inner.classifier,
                &previous,
                &legacy_observation_keys,
            );
            if !rollback_errors.is_empty() {
                tracing::error!(
                    recording_id=%recording_id,
                    errors=?rollback_errors,
                    "Could not fully restore classifier state after packet persistence failed"
                );
            }
            return Err(Error::internal(error));
        }
        Ok(stored.packet)
    }

    fn recording_id_by_sha(&self, sha256: &str) -> Result<Option<Uuid>, Error> {
        let db = self.inner.db.lock().map_err(Error::internal)?;
        db.query_row(
            "SELECT id FROM audio_recordings WHERE sha256=?1",
            [sha256],
            |row| row.get::<_, String>(0),
        )
        .optional()
        .map_err(Error::internal)?
        .map(|id| Uuid::parse_str(&id).map_err(Error::internal))
        .transpose()
    }
}

fn row_recording_status(row: &rusqlite::Row<'_>) -> rusqlite::Result<RecordingStatus> {
    let id: String = row.get(0)?;
    let recorded_at: String = row.get(5)?;
    let received_at: String = row.get(6)?;
    let durable_status: String = row.get(7)?;
    let attempts: i64 = row.get(11)?;
    let last_error: Option<String> = row.get(12)?;
    let retryable: i64 = row.get(13)?;
    let progress_json: Option<String> = row.get(14)?;
    let transcript: Option<String> = row.get(15)?;
    let correction_packet_json: Option<String> = row.get(16)?;
    let parse_time = |index, value: &str| {
        DateTime::parse_from_rfc3339(value)
            .map(|value| value.with_timezone(&Utc))
            .map_err(|error| {
                rusqlite::Error::FromSqlConversionFailure(
                    index,
                    rusqlite::types::Type::Text,
                    Box::new(error),
                )
            })
    };
    let state = match durable_status.as_str() {
        "uploaded" => RecordingState::Queued,
        "chunking" | "transcribing" | "reconciling" => {
            let progress = progress_json
                .as_deref()
                .map(serde_json::from_str)
                .transpose()
                .map_err(|error| {
                    rusqlite::Error::FromSqlConversionFailure(
                        14,
                        rusqlite::types::Type::Text,
                        Box::new(error),
                    )
                })?
                .unwrap_or_else(initial_progress);
            RecordingState::Processing {
                attempt: attempts.clamp(0, i64::from(u8::MAX)) as u8,
                progress: without_results(progress),
            }
        }
        "ready_for_ingress" | "ingressing" | "ingress_failed" | "complete" => {
            RecordingState::Complete {
                transcript: transcript.unwrap_or_default(),
            }
        }
        "failed" => RecordingState::Failed {
            attempts: attempts.clamp(0, i64::from(u8::MAX)) as u8,
            error: last_error.unwrap_or_else(|| "Audio processing failed.".into()),
            retryable: retryable != 0,
        },
        other => {
            return Err(rusqlite::Error::FromSqlConversionFailure(
                7,
                rusqlite::types::Type::Text,
                format!("unknown audio status {other:?}").into(),
            ));
        }
    };
    let correction_packet = correction_packet_json
        .as_deref()
        .map(StoredCorrectionPacket::decode)
        .transpose()
        .map_err(|error| {
            rusqlite::Error::FromSqlConversionFailure(
                16,
                rusqlite::types::Type::Text,
                Box::new(error),
            )
        })?
        .map(|stored| stored.packet);
    Ok(RecordingStatus {
        id: Uuid::parse_str(&id).map_err(|error| {
            rusqlite::Error::FromSqlConversionFailure(
                0,
                rusqlite::types::Type::Text,
                Box::new(error),
            )
        })?,
        user_id: row.get(1)?,
        sha256: row.get(2)?,
        original_filename: row.get(3)?,
        size_bytes: u64::try_from(row.get::<_, i64>(4)?).map_err(|error| {
            rusqlite::Error::FromSqlConversionFailure(
                4,
                rusqlite::types::Type::Integer,
                Box::new(error),
            )
        })?,
        recorded_at: parse_time(5, &recorded_at)?,
        received_at: parse_time(6, &received_at)?,
        transcription_model: row.get(8)?,
        reconciliation_model: row.get(9)?,
        reconciliation_reasoning: row.get(10)?,
        state,
        correction_packet,
    })
}

#[derive(Clone, Debug)]
struct LegacyFeatureRow {
    chunk_position: usize,
    speaker_position: usize,
    value: Value,
}

#[derive(Clone, Debug)]
struct StoredCorrectionPacket {
    packet: CorrectionPacket,
    legacy_feature_rows: Vec<LegacyFeatureRow>,
}

impl StoredCorrectionPacket {
    fn decode(serialized: &str) -> serde_json::Result<Self> {
        match serde_json::from_str(serialized) {
            Ok(packet) => Ok(Self {
                packet,
                legacy_feature_rows: Vec::new(),
            }),
            Err(current_error) => {
                let mut value: Value = serde_json::from_str(serialized)?;
                let Some(chunks) = value.get_mut("chunks").and_then(Value::as_array_mut) else {
                    return Err(current_error);
                };
                let mut legacy_feature_rows = Vec::new();
                for (chunk_position, chunk) in chunks.iter_mut().enumerate() {
                    let Some(speakers) = chunk
                        .pointer_mut("/parsed/speakers")
                        .and_then(Value::as_array_mut)
                    else {
                        continue;
                    };
                    for (speaker_position, speaker) in speakers.iter_mut().enumerate() {
                        let Some(feature_row) = speaker.get_mut("feature_row") else {
                            continue;
                        };
                        if is_legacy_feature_row(feature_row) {
                            legacy_feature_rows.push(LegacyFeatureRow {
                                chunk_position,
                                speaker_position,
                                value: feature_row.take(),
                            });
                        }
                    }
                }
                if legacy_feature_rows.is_empty() {
                    return Err(current_error);
                }
                let packet = serde_json::from_value(value)?;
                Ok(Self {
                    packet,
                    legacy_feature_rows,
                })
            }
        }
    }

    fn encode(&self) -> serde_json::Result<String> {
        let mut value = serde_json::to_value(&self.packet)?;
        for legacy in &self.legacy_feature_rows {
            let feature_row = value
                .get_mut("chunks")
                .and_then(Value::as_array_mut)
                .and_then(|chunks| chunks.get_mut(legacy.chunk_position))
                .and_then(|chunk| chunk.pointer_mut("/parsed/speakers"))
                .and_then(Value::as_array_mut)
                .and_then(|speakers| speakers.get_mut(legacy.speaker_position))
                .and_then(|speaker| speaker.get_mut("feature_row"))
                .expect("decoded correction packet retains its speaker positions");
            *feature_row = legacy.value.clone();
        }
        serde_json::to_string(&value)
    }

    fn legacy_observation_keys(&self) -> anyhow::Result<HashSet<(String, u32)>> {
        let mut keys = HashSet::new();
        for legacy in &self.legacy_feature_rows {
            let chunk = self
                .packet
                .chunks
                .get(legacy.chunk_position)
                .context("legacy feature row is outside its correction packet")?;
            let speaker = chunk
                .parsed
                .speakers
                .get(legacy.speaker_position)
                .context("legacy feature row is outside its parsed speakers")?;
            let observation = chunk
                .observations
                .iter()
                .find(|observation| {
                    observation.speaker_ordinal as usize == legacy.speaker_position
                        && observation.local_label == speaker.local_label
                })
                .context("legacy feature row has no matching correction observation")?;
            ensure!(
                keys.insert((
                    observation.observation_key.object_id.clone(),
                    observation.observation_key.piece_index,
                )),
                "legacy correction packet repeats an observation key"
            );
        }
        Ok(keys)
    }
}

fn is_legacy_feature_row(value: &Value) -> bool {
    const FIELDS: [&str; 24] = [
        "accent_variety",
        "articulation_rate_syllables_per_second",
        "breathiness",
        "cefr",
        "consonant_cluster_reduction_percent",
        "creaky_phonation_percent",
        "f0_pitch_span_semitones",
        "filled_pauses_per_100_words",
        "foreign_accentedness",
        "formant_dispersion_hz",
        "hypernasality",
        "lateral_realization",
        "lexical_stress_accuracy_percent",
        "median_f0_hz",
        "monophthongization_percent",
        "npvi_v",
        "perceived_age",
        "rhotic_realization",
        "roughness",
        "s_realization",
        "unstressed_vowel_reduction_percent",
        "vai",
        "vocal_gender_presentation",
        "word_initial_stressed_prevocalic_t_vot_ms",
    ];
    value.as_object().is_some_and(|object| {
        object.len() == FIELDS.len() && FIELDS.iter().all(|field| object.contains_key(*field))
    })
}

async fn worker_loop(inner: Weak<Inner>) {
    loop {
        let Some(inner) = inner.upgrade() else {
            return;
        };
        let worked = match process_next_recording(&inner).await {
            Ok(worked) => worked,
            Err(error) => {
                tracing::error!(error=%error, "AudioIngress worker iteration failed");
                false
            }
        };
        drop(inner);
        tokio::time::sleep(if worked {
            Duration::from_millis(100)
        } else {
            Duration::from_secs(5)
        })
        .await;
    }
}

#[derive(Debug)]
struct WorkRecording {
    id: Uuid,
    user_id: String,
    sha256: String,
    original_filename: String,
    size_bytes: u64,
    recorded_at: DateTime<Utc>,
    original_relative_path: String,
    attempt_count: i64,
}

async fn process_next_recording(inner: &Inner) -> anyhow::Result<bool> {
    let recording = {
        let db = inner
            .db
            .lock()
            .map_err(|_| anyhow::anyhow!("AudioIngress database lock was poisoned"))?;
        fetch_work_recording(&db)?
    };
    let Some(recording) = recording else {
        return Ok(false);
    };
    poll_transcription(inner, recording).await?;
    Ok(true)
}

fn fetch_work_recording(db: &Connection) -> anyhow::Result<Option<WorkRecording>> {
    db.query_row(
        "SELECT id,user_id,sha256,original_filename,size_bytes,source_created_at,
                original_relative_path,attempt_count
         FROM audio_recordings
         WHERE status IN ('uploaded','chunking','transcribing','reconciling')
           AND (next_attempt_at IS NULL OR datetime(next_attempt_at)<=datetime('now'))
         ORDER BY datetime(received_at),id
         LIMIT 1",
        [],
        |row| {
            Ok((
                row.get::<_, String>(0)?,
                row.get::<_, String>(1)?,
                row.get::<_, String>(2)?,
                row.get::<_, String>(3)?,
                row.get::<_, i64>(4)?,
                row.get::<_, String>(5)?,
                row.get::<_, String>(6)?,
                row.get::<_, i64>(7)?,
            ))
        },
    )
    .optional()?
    .map(
        |(
            id,
            user_id,
            sha256,
            original_filename,
            size_bytes,
            recorded_at,
            original_relative_path,
            attempt_count,
        )| {
            Ok(WorkRecording {
                id: Uuid::parse_str(&id)?,
                user_id,
                sha256,
                original_filename,
                size_bytes: u64::try_from(size_bytes).context("stored audio size is negative")?,
                recorded_at: DateTime::parse_from_rfc3339(&recorded_at)
                    .context("stored recording time is invalid")?
                    .with_timezone(&Utc),
                original_relative_path,
                attempt_count,
            })
        },
    )
    .transpose()
}

async fn poll_transcription(inner: &Inner, mut recording: WorkRecording) -> anyhow::Result<()> {
    let existing = inner
        .jobs
        .lock()
        .map_err(|_| anyhow::anyhow!("audio transcription job lock was poisoned"))?
        .get(&recording.id)
        .cloned();
    let job = if let Some(job) = existing {
        job
    } else {
        if recording.attempt_count >= FAILURE_LIMIT {
            mark_failed(
                inner,
                recording.id,
                recording.attempt_count,
                true,
                "Audio transcription exhausted its five automatic attempts.",
                None,
            )?;
            return Ok(());
        }
        let source = inner.root.join(&recording.original_relative_path);
        let audio = tokio::fs::read(&source)
            .await
            .with_context(|| format!("reading retained audio {}", source.display()))?;
        recording.attempt_count += 1;
        {
            let db = inner
                .db
                .lock()
                .map_err(|_| anyhow::anyhow!("AudioIngress database lock was poisoned"))?;
            db.execute(
                "UPDATE audio_recordings
                 SET status='chunking',attempt_count=?1,next_attempt_at=NULL,last_error=NULL,
                     failure_retryable=1,updated_at=?2
                 WHERE id=?3",
                params![
                    recording.attempt_count,
                    Utc::now().to_rfc3339(),
                    recording.id.to_string()
                ],
            )?;
        }

        let classification = ClassificationContext {
            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,
            classifier: inner.classifier.clone(),
        };

        let cache_db = inner.db.clone();
        let cache_recording_id = recording.id;
        let cache_classification = classification.clone();
        let piece_cache: PieceCache = Arc::new(move |plan| {
            load_cached_transcript_piece(&cache_db, cache_recording_id, plan, &cache_classification)
        });

        let sink_db = inner.db.clone();
        let sink_recording_id = recording.id;
        let attempt_id = Uuid::new_v4().to_string();
        let piece_sink: PieceSink = Arc::new(move |piece| {
            persist_transcript_piece(&sink_db, sink_recording_id, &attempt_id, piece)
        });

        let job = inner.transcriber.transcribe_durably(
            recording.user_id.clone(),
            audio,
            piece_cache,
            piece_sink,
            classification,
        );
        inner
            .jobs
            .lock()
            .map_err(|_| anyhow::anyhow!("audio transcription job lock was poisoned"))?
            .insert(recording.id, job.clone());
        job
    };

    let snapshot = job.status();
    persist_progress(inner, recording.id, &snapshot)?;
    match snapshot.state {
        JobState::Queued | JobState::Running => Ok(()),
        JobState::Completed => {
            let transcript = snapshot
                .transcript
                .clone()
                .context("completed transcription omitted its transcript")?;
            ensure!(
                !transcript.trim().is_empty(),
                "completed transcription is empty"
            );
            let packet = snapshot
                .correction_packet
                .clone()
                .context("completed durable transcription omitted its correction packet")?;
            ensure!(
                packet.recording_id == recording.id,
                "completed correction packet belongs to another recording"
            );
            let packet_json = serde_json::to_string(&packet)?;
            let progress = serde_json::to_string(&without_results(snapshot))?;
            {
                let db = inner
                    .db
                    .lock()
                    .map_err(|_| anyhow::anyhow!("AudioIngress database lock was poisoned"))?;
                db.execute(
                    "UPDATE audio_recordings
                     SET status='ready_for_ingress',final_transcript=?1,
                         correction_packet_json=?2,transcription_status_json=?3,
                         next_attempt_at=NULL,last_error=NULL,failure_retryable=1,updated_at=?4
                     WHERE id=?5",
                    params![
                        transcript.trim(),
                        packet_json,
                        progress,
                        Utc::now().to_rfc3339(),
                        recording.id.to_string()
                    ],
                )?;
            }
            remove_job(inner, recording.id)?;
            tracing::info!(
                recording_id=%recording.id,
                clean=packet.clean,
                "Audio transcript and correction packet completed"
            );
            Ok(())
        }
        JobState::Failed => {
            let error = snapshot
                .steps
                .iter()
                .find(|step| step.state == StepState::Failed)
                .and_then(|step| step.error.as_ref());
            let message = error
                .map(|error| error.message.clone())
                .unwrap_or_else(|| "Audio transcription failed without detail.".into());
            let retryable = error.is_none_or(|error| error.retryable);
            record_attempt_failure(
                inner,
                recording.id,
                recording.attempt_count,
                retryable,
                &message,
                Some(snapshot),
            )?;
            remove_job(inner, recording.id)
        }
    }
}

fn record_attempt_failure(
    inner: &Inner,
    id: Uuid,
    attempts: i64,
    retryable: bool,
    message: &str,
    progress: Option<TranscriptionStatus>,
) -> anyhow::Result<()> {
    if !retryable || attempts >= FAILURE_LIMIT {
        return mark_failed(inner, id, attempts, retryable, message, progress);
    }
    let db = inner
        .db
        .lock()
        .map_err(|_| anyhow::anyhow!("AudioIngress database lock was poisoned"))?;
    db.execute(
        "UPDATE audio_recordings
         SET status='uploaded',next_attempt_at=?1,last_error=?2,failure_retryable=1,
             transcription_status_json=?3,updated_at=?4
         WHERE id=?5",
        params![
            (Utc::now() + ChronoDuration::seconds(RETRY_DELAY_SECONDS)).to_rfc3339(),
            concise(message, 2_000),
            progress
                .map(without_results)
                .map(|progress| serde_json::to_string(&progress))
                .transpose()?,
            Utc::now().to_rfc3339(),
            id.to_string()
        ],
    )?;
    tracing::warn!(recording_id=%id, attempt=attempts, "Audio transcription will retry");
    Ok(())
}

fn mark_failed(
    inner: &Inner,
    id: Uuid,
    attempts: i64,
    retryable: bool,
    message: &str,
    progress: Option<TranscriptionStatus>,
) -> anyhow::Result<()> {
    let db = inner
        .db
        .lock()
        .map_err(|_| anyhow::anyhow!("AudioIngress database lock was poisoned"))?;
    db.execute(
        "UPDATE audio_recordings
         SET status='failed',attempt_count=?1,next_attempt_at=NULL,last_error=?2,
             failure_retryable=?3,transcription_status_json=?4,updated_at=?5
         WHERE id=?6",
        params![
            attempts,
            concise(message, 2_000),
            i64::from(retryable),
            progress
                .map(without_results)
                .map(|progress| serde_json::to_string(&progress))
                .transpose()?,
            Utc::now().to_rfc3339(),
            id.to_string()
        ],
    )?;
    tracing::error!(recording_id=%id, attempts, retryable, "Audio transcription stopped");
    Ok(())
}

fn persist_progress(inner: &Inner, id: Uuid, snapshot: &TranscriptionStatus) -> anyhow::Result<()> {
    let durable_status = transcription_stage(snapshot);
    let serialized = serde_json::to_string(&without_results(snapshot.clone()))?;
    let db = inner
        .db
        .lock()
        .map_err(|_| anyhow::anyhow!("AudioIngress database lock was poisoned"))?;
    db.execute(
        "UPDATE audio_recordings
         SET status=?1,transcription_status_json=?2,updated_at=?3
         WHERE id=?4 AND (status<>?1 OR COALESCE(transcription_status_json,'')<>?2)",
        params![
            durable_status,
            serialized,
            Utc::now().to_rfc3339(),
            id.to_string()
        ],
    )?;
    Ok(())
}

fn load_cached_transcript_piece(
    db: &Mutex<Connection>,
    recording_id: Uuid,
    plan: ChunkPlan,
    classification: &ClassificationContext,
) -> anyhow::Result<Option<ChunkTranscript>> {
    ensure!(
        recording_id == classification.recording_id,
        "cache classification context belongs to another recording"
    );
    let piece_index = i64::try_from(plan.index).context("piece index exceeds SQLite limits")?;
    let piece_count = i64::try_from(plan.total).context("piece count exceeds SQLite limits")?;
    let audio_start_ms =
        i64::try_from(plan.start_ms).context("piece start exceeds SQLite limits")?;
    let audio_end_ms = i64::try_from(plan.end_ms).context("piece end exceeds SQLite limits")?;
    let stored = {
        let db = db
            .lock()
            .map_err(|_| anyhow::anyhow!("AudioIngress database lock was poisoned"))?;
        db.query_row(
            "SELECT raw_gemini_response,parsed_json
             FROM audio_transcript_pieces
             WHERE recording_id=?1
               AND cache_revision=?2
               AND piece_index=?3
               AND piece_count=?4
               AND audio_start_ms=?5
               AND audio_end_ms=?6
             ORDER BY datetime(created_at) DESC,attempt_id DESC
             LIMIT 1",
            params![
                recording_id.to_string(),
                PIECE_CACHE_REVISION,
                piece_index,
                piece_count,
                audio_start_ms,
                audio_end_ms,
            ],
            |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
        )
        .optional()?
    };
    let Some((raw_gemini_response, parsed_json)) = stored else {
        return Ok(None);
    };
    ensure!(
        !raw_gemini_response.trim().is_empty(),
        "cached piece omitted its raw Gemini response"
    );
    let duration_seconds = (plan.end_ms - plan.start_ms) as f64 / 1_000.0;
    let mut parsed: ParsedChunk = serde_json::from_str(&parsed_json)
        .context("cached parsed speaker analysis is not valid JSON")?;
    validate_parsed_chunk(&mut parsed, duration_seconds)
        .context("cached parsed speaker analysis is invalid")?;
    let (observations, clean) = classify_speakers(classification, plan.index, &parsed)
        .context("reclassifying cached speaker rows failed")?;
    Ok(Some(ChunkTranscript {
        plan,
        raw_gemini_response,
        parsed,
        observations,
        clean,
    }))
}

fn persist_transcript_piece(
    db: &Mutex<Connection>,
    recording_id: Uuid,
    attempt_id: &str,
    piece: &ChunkTranscript,
) -> anyhow::Result<()> {
    ensure!(!attempt_id.is_empty(), "piece attempt identity is empty");
    let piece_index =
        i64::try_from(piece.plan.index).context("piece index exceeds SQLite limits")?;
    let piece_count =
        i64::try_from(piece.plan.total).context("piece count exceeds SQLite limits")?;
    let audio_start_ms =
        i64::try_from(piece.plan.start_ms).context("piece start exceeds SQLite limits")?;
    let audio_end_ms =
        i64::try_from(piece.plan.end_ms).context("piece end exceeds SQLite limits")?;
    let parsed_json = serde_json::to_string(&piece.parsed)?;
    let db = db
        .lock()
        .map_err(|_| anyhow::anyhow!("AudioIngress database lock was poisoned"))?;
    db.execute(
        "INSERT INTO audio_transcript_pieces(
            recording_id,attempt_id,cache_revision,piece_index,piece_count,
            audio_start_ms,audio_end_ms,transcript_json,raw_gemini_response,
            parsed_json,created_at
         ) VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9,?8,?10)
         ON CONFLICT(recording_id,attempt_id,piece_index) DO UPDATE SET
            cache_revision=excluded.cache_revision,
            piece_count=excluded.piece_count,
            audio_start_ms=excluded.audio_start_ms,
            audio_end_ms=excluded.audio_end_ms,
            transcript_json=excluded.transcript_json,
            raw_gemini_response=excluded.raw_gemini_response,
            parsed_json=excluded.parsed_json",
        params![
            recording_id.to_string(),
            attempt_id,
            PIECE_CACHE_REVISION,
            piece_index,
            piece_count,
            audio_start_ms,
            audio_end_ms,
            parsed_json,
            piece.raw_gemini_response,
            Utc::now().to_rfc3339(),
        ],
    )?;
    Ok(())
}

fn remove_job(inner: &Inner, id: Uuid) -> anyhow::Result<()> {
    inner
        .jobs
        .lock()
        .map_err(|_| anyhow::anyhow!("audio transcription job lock was poisoned"))?
        .remove(&id);
    Ok(())
}

fn transcription_stage(snapshot: &TranscriptionStatus) -> &'static str {
    let plan_complete = snapshot
        .steps
        .iter()
        .any(|entry| entry.step == Step::PlanChunks && entry.state == StepState::Completed);
    if !plan_complete {
        return "chunking";
    }
    let chunks_complete = snapshot
        .steps
        .iter()
        .filter(|entry| {
            matches!(
                entry.step,
                Step::TranscribeChunk { .. } | Step::ParseChunk { .. }
            )
        })
        .all(|entry| entry.state == StepState::Completed);
    if chunks_complete {
        "reconciling"
    } else {
        "transcribing"
    }
}

fn without_results(mut status: TranscriptionStatus) -> TranscriptionStatus {
    status.transcript = None;
    status.correction_packet = None;
    status
}

fn initial_progress() -> TranscriptionStatus {
    TranscriptionStatus {
        state: JobState::Queued,
        steps: Vec::new(),
        transcript: None,
        correction_packet: None,
    }
}

fn recover_interrupted_attempts(connection: &Connection) -> rusqlite::Result<()> {
    connection.execute(
        "UPDATE audio_recordings
         SET status=CASE WHEN attempt_count>=?1 THEN 'failed' ELSE 'uploaded' END,
             next_attempt_at=NULL,
             last_error=CASE WHEN attempt_count>=?1
                 THEN 'Audio transcription stopped after its fifth attempt was interrupted.'
                 ELSE 'Audio transcription was interrupted and will restart automatically.'
             END,
             failure_retryable=1,
             updated_at=?2
         WHERE status IN ('chunking','transcribing','reconciling')",
        params![FAILURE_LIMIT, Utc::now().to_rfc3339()],
    )?;
    Ok(())
}

fn apply_migrations(connection: &Connection) -> anyhow::Result<()> {
    let version: i64 = connection.query_row("PRAGMA user_version", [], |row| row.get(0))?;
    ensure!(
        version <= LATEST_SCHEMA_VERSION,
        "audio-ingress database schema version {version} is newer than supported version {LATEST_SCHEMA_VERSION}"
    );
    if version == 0 {
        let has_recordings = connection.query_row(
            "SELECT EXISTS(
                SELECT 1 FROM sqlite_schema
                WHERE type='table' AND name='audio_recordings'
             )",
            [],
            |row| row.get::<_, i64>(0),
        )? == 1;
        if !has_recordings {
            connection.execute_batch(FRESH_SCHEMA)?;
            return Ok(());
        }
    }
    if version < 1 {
        connection.execute_batch(INITIAL_MIGRATION)?;
    }
    if version < 2 {
        connection.execute_batch(RELEASE_DEFERRED_INGRESS_MIGRATION)?;
    }
    if version < 3 {
        connection.execute_batch(TRANSCRIPTION_STATUS_MIGRATION)?;
    }
    if version < 4 {
        connection.execute_batch(RETRY_ROUNDED_WAV_INTERVALS_MIGRATION)?;
    }
    if version < 5 {
        connection.execute_batch(UNIFIED_INGRESS_QUEUE_MIGRATION)?;
    }
    if version < 6 {
        connection.execute_batch(STANDALONE_LIBRARY_MIGRATION)?;
    }
    if version < 7 {
        connection.execute_batch(DURABLE_TRANSCRIPT_PIECES_MIGRATION)?;
    }
    if version < 8 {
        connection.execute_batch(UNIQUE_TRANSCRIPTION_ATTEMPTS_MIGRATION)?;
    }
    if version < 9 {
        connection.execute_batch(USAGE_USER_MIGRATION)?;
    }
    if version < 10 {
        connection.execute_batch(SPEAKER_CORRECTION_PACKETS_MIGRATION)?;
    }
    Ok(())
}

fn ensure_private_directory(path: &Path) -> anyhow::Result<()> {
    fs::create_dir_all(path).with_context(|| format!("creating {}", path.display()))?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        fs::set_permissions(path, fs::Permissions::from_mode(0o700))
            .with_context(|| format!("setting private permissions on {}", path.display()))?;
    }
    Ok(())
}

fn set_private_file(path: &Path) -> anyhow::Result<()> {
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        fs::set_permissions(path, fs::Permissions::from_mode(0o600))
            .with_context(|| format!("setting private permissions on {}", path.display()))?;
    }
    Ok(())
}

fn sync_file(path: &Path) -> anyhow::Result<()> {
    fs::OpenOptions::new()
        .read(true)
        .write(true)
        .open(path)
        .with_context(|| format!("opening {} for sync", path.display()))?
        .sync_all()
        .with_context(|| format!("syncing {}", path.display()))
}

fn sync_directory(path: &Path) -> anyhow::Result<()> {
    #[cfg(unix)]
    fs::File::open(path)
        .with_context(|| format!("opening directory {} for sync", path.display()))?
        .sync_all()
        .with_context(|| format!("syncing directory {}", path.display()))?;
    Ok(())
}

fn safe_filename(value: Option<&str>) -> String {
    let name = value
        .and_then(|value| Path::new(value).file_name())
        .and_then(|value| value.to_str())
        .unwrap_or("audio.wav");
    let clean = name
        .chars()
        .map(|character| {
            if character.is_ascii_alphanumeric() || matches!(character, '.' | '-' | '_') {
                character
            } else {
                '_'
            }
        })
        .take(200)
        .collect::<String>();
    if clean.is_empty() {
        "audio.wav".into()
    } else {
        clean
    }
}

fn concise(value: &str, limit: usize) -> String {
    let normalized = value.split_whitespace().collect::<Vec<_>>().join(" ");
    let bounded = normalized.chars().take(limit).collect::<String>();
    if bounded.is_empty() {
        "Audio transcription failed without an error message.".into()
    } else {
        bounded
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicU64, Ordering};

    static NEXT_PATH: AtomicU64 = AtomicU64::new(0);

    fn database() -> Connection {
        let connection = Connection::open_in_memory().unwrap();
        apply_migrations(&connection).unwrap();
        connection
    }

    fn classifier_path(label: &str) -> PathBuf {
        std::env::temp_dir().join(format!(
            "kcode-audio-ingress-lib-{}-{label}-{}.sqlite3",
            std::process::id(),
            NEXT_PATH.fetch_add(1, Ordering::Relaxed)
        ))
    }

    fn root_path(label: &str) -> PathBuf {
        std::env::temp_dir().join(format!(
            "kcode-audio-ingress-lib-{}-{label}-{}",
            std::process::id(),
            NEXT_PATH.fetch_add(1, Ordering::Relaxed)
        ))
    }

    fn remove_database(path: &Path) {
        for suffix in ["", "-wal", "-shm"] {
            let mut value = path.as_os_str().to_os_string();
            value.push(suffix);
            let _ = fs::remove_file(PathBuf::from(value));
        }
    }

    fn transcriber() -> AudioTranscriber {
        let audio: AudioChunkCall = Arc::new(|_| {
            Box::pin(async { Err(IntelligenceError::new("unexpected audio call", false)) })
        });
        let text: TextGenerationCall = Arc::new(|_| {
            Box::pin(async { Err(IntelligenceError::new("unexpected text call", false)) })
        });
        AudioTranscriber::new(audio, text)
    }

    fn insert_recording(connection: &Connection, id: Uuid, status: &str) {
        let has_user_id = connection
            .prepare("SELECT 1 FROM pragma_table_info('audio_recordings') WHERE name='user_id'")
            .unwrap()
            .exists([])
            .unwrap();
        let sql = if has_user_id {
            "INSERT INTO audio_recordings(
                id,user_id,sha256,original_filename,content_type,size_bytes,source_created_at,
                received_at,updated_at,original_relative_path,status,gemini_model,
                reconciliation_model,reconciliation_reasoning
             ) VALUES(?1,'test-user',?2,'note.wav','audio/wav',4,?3,?3,?3,?4,?5,?6,?7,?8)"
        } else {
            "INSERT INTO audio_recordings(
                id,sha256,original_filename,content_type,size_bytes,source_created_at,
                received_at,updated_at,original_relative_path,status,gemini_model,
                reconciliation_model,reconciliation_reasoning
             ) VALUES(?1,?2,'note.wav','audio/wav',4,?3,?3,?3,?4,?5,?6,?7,?8)"
        };
        connection
            .execute(
                sql,
                params![
                    id.to_string(),
                    format!("{:064x}", 1),
                    "2026-01-01T00:00:00Z",
                    format!("originals/{id}.wav"),
                    status,
                    TRANSCRIPTION_MODEL,
                    RECONCILIATION_MODEL,
                    RECONCILIATION_REASONING,
                ],
            )
            .unwrap();
    }

    fn sample_row() -> FeatureRow {
        FeatureRow::new(std::array::from_fn(|index| {
            u8::try_from(index * 3 + 10).unwrap()
        }))
        .unwrap()
    }

    fn legacy_feature_row() -> Value {
        serde_json::json!({
            "accent_variety":"general_american",
            "articulation_rate_syllables_per_second":4.1,
            "breathiness":0.1,
            "cefr":"c2",
            "consonant_cluster_reduction_percent":0.0,
            "creaky_phonation_percent":0.0,
            "f0_pitch_span_semitones":8.0,
            "filled_pauses_per_100_words":1.0,
            "foreign_accentedness":0.0,
            "formant_dispersion_hz":1100.0,
            "hypernasality":0.0,
            "lateral_realization":"standard",
            "lexical_stress_accuracy_percent":100.0,
            "median_f0_hz":150.0,
            "monophthongization_percent":0.0,
            "npvi_v":50.0,
            "perceived_age":40.0,
            "rhotic_realization":"rhotic",
            "roughness":0.1,
            "s_realization":"standard",
            "unstressed_vowel_reduction_percent":50.0,
            "vai":1.0,
            "vocal_gender_presentation":50.0,
            "word_initial_stressed_prevocalic_t_vot_ms":60.0
        })
    }

    fn legacy_packet_json(id: Uuid) -> String {
        let packet = CorrectionPacket {
            recording_id: id,
            user_id: "test-user".into(),
            sha256: format!("{:064x}", 1),
            original_filename: "note.wav".into(),
            size_bytes: 4,
            recorded_at: DateTime::parse_from_rfc3339("2026-01-01T00:00:00Z")
                .unwrap()
                .with_timezone(&Utc),
            clean: false,
            chunk_count: 1,
            chunks: vec![CorrectionChunk {
                chunk_index: 0,
                chunk_count: 1,
                audio_start_ms: 0,
                audio_end_ms: 1_000,
                raw_gemini_response: "legacy raw response".into(),
                parsed: sample_parsed(),
                observations: vec![CorrectionObservation {
                    local_label: "Speaker A".into(),
                    speaker_ordinal: 0,
                    observation_key: ObservationKey {
                        object_id: format!("kcode-audio-ingress/recording/{id}/chunk/0"),
                        piece_index: 0,
                    },
                    candidate: None,
                    identified_full_name: None,
                    confirmed_full_name: None,
                }],
                clean: false,
            }],
            confirmation_state: ConfirmationState::Unconfirmed,
        };
        let mut value = serde_json::to_value(packet).unwrap();
        value["chunks"][0]["parsed"]["speakers"][0]["feature_row"] = legacy_feature_row();
        serde_json::to_string(&value).unwrap()
    }

    fn sample_parsed() -> ParsedChunk {
        ParsedChunk {
            utterances: vec![ParsedUtterance {
                speaker: "Speaker A".into(),
                language: "eng".into(),
                original_text: "Hello.".into(),
                english_translation: String::new(),
                corrected_natural_text: None,
                coaching: Vec::new(),
                annotations: Vec::new(),
            }],
            notes: vec!["Clear recording.".into()],
            clip_valid: true,
            clip_validity_reason: None,
            speakers: vec![ParsedSpeaker {
                local_label: "Speaker A".into(),
                primary_language: Some("eng".into()),
                feature_row: Some(sample_row()),
            }],
        }
    }

    fn sample_piece() -> ChunkTranscript {
        ChunkTranscript {
            plan: ChunkPlan {
                index: 0,
                total: 2,
                start_ms: 0,
                end_ms: 120_000,
            },
            raw_gemini_response: "complete raw Gemini response".into(),
            parsed: sample_parsed(),
            observations: Vec::new(),
            clean: false,
        }
    }

    fn classification_context(
        recording_id: Uuid,
        classifier: Arc<SpeechClassifier>,
    ) -> ClassificationContext {
        ClassificationContext {
            recording_id,
            user_id: "test-user".into(),
            sha256: format!("{:064x}", 1),
            original_filename: "note.wav".into(),
            size_bytes: 4,
            recorded_at: DateTime::parse_from_rfc3339("2026-01-01T00:00:00Z")
                .unwrap()
                .with_timezone(&Utc),
            classifier,
        }
    }

    #[test]
    fn fresh_schema_contains_recordings_durable_pieces_and_packet_columns() {
        let connection = database();
        let tables = connection
            .prepare(
                "SELECT name FROM sqlite_schema
                 WHERE type='table' AND name NOT LIKE 'sqlite_%'
                 ORDER BY name",
            )
            .unwrap()
            .query_map([], |row| row.get::<_, String>(0))
            .unwrap()
            .collect::<Result<Vec<_>, _>>()
            .unwrap();
        let version: i64 = connection
            .query_row("PRAGMA user_version", [], |row| row.get(0))
            .unwrap();
        let piece_columns = connection
            .prepare("SELECT name FROM pragma_table_info('audio_transcript_pieces')")
            .unwrap()
            .query_map([], |row| row.get::<_, String>(0))
            .unwrap()
            .collect::<Result<Vec<_>, _>>()
            .unwrap();
        let recording_columns = connection
            .prepare("SELECT name FROM pragma_table_info('audio_recordings')")
            .unwrap()
            .query_map([], |row| row.get::<_, String>(0))
            .unwrap()
            .collect::<Result<Vec<_>, _>>()
            .unwrap();
        assert_eq!(tables, vec!["audio_recordings", "audio_transcript_pieces"]);
        assert!(piece_columns.contains(&"raw_gemini_response".into()));
        assert!(piece_columns.contains(&"parsed_json".into()));
        assert!(recording_columns.contains(&"correction_packet_json".into()));
        assert_eq!(version, LATEST_SCHEMA_VERSION);
    }

    #[tokio::test]
    async fn open_uses_the_injected_classifier_without_creating_a_private_database() {
        let root = root_path("shared-classifier-root");
        let classifier_path = classifier_path("shared-classifier");
        let classifier = Arc::new(SpeechClassifier::open(&classifier_path).unwrap());

        let ingress = AudioIngress::open(&root, transcriber(), classifier.clone())
            .await
            .unwrap();

        assert!(Arc::ptr_eq(&ingress.inner.classifier, &classifier));
        assert!(!root.join("speaker-classification.sqlite3").exists());

        drop(ingress);
        tokio::task::yield_now().await;
        let _ = fs::remove_dir_all(&root);
        drop(classifier);
        remove_database(&classifier_path);
    }

    #[test]
    fn version_five_databases_upgrade_without_losing_legacy_queue_data() {
        let connection = Connection::open_in_memory().unwrap();
        connection.execute_batch(INITIAL_MIGRATION).unwrap();
        connection
            .execute_batch(RELEASE_DEFERRED_INGRESS_MIGRATION)
            .unwrap();
        connection
            .execute_batch(TRANSCRIPTION_STATUS_MIGRATION)
            .unwrap();
        connection
            .execute_batch(RETRY_ROUNDED_WAV_INTERVALS_MIGRATION)
            .unwrap();
        connection
            .execute_batch(UNIFIED_INGRESS_QUEUE_MIGRATION)
            .unwrap();
        let id = Uuid::new_v4();
        insert_recording(&connection, id, "failed");

        apply_migrations(&connection).unwrap();

        let version: i64 = connection
            .query_row("PRAGMA user_version", [], |row| row.get(0))
            .unwrap();
        let retryable: i64 = connection
            .query_row(
                "SELECT failure_retryable FROM audio_recordings WHERE id=?1",
                [id.to_string()],
                |row| row.get(0),
            )
            .unwrap();
        let legacy_queue_exists: i64 = connection
            .query_row(
                "SELECT EXISTS(
                    SELECT 1 FROM sqlite_schema
                    WHERE type='table' AND name='audio_ingress_pieces'
                 )",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(version, LATEST_SCHEMA_VERSION);
        assert_eq!(retryable, 1);
        assert_eq!(legacy_queue_exists, 1);
    }

    #[test]
    fn version_seven_piece_rows_migrate_but_cannot_hit_the_new_cache() {
        let connection = Connection::open_in_memory().unwrap();
        connection.execute_batch(INITIAL_MIGRATION).unwrap();
        connection
            .execute_batch(RELEASE_DEFERRED_INGRESS_MIGRATION)
            .unwrap();
        connection
            .execute_batch(TRANSCRIPTION_STATUS_MIGRATION)
            .unwrap();
        connection
            .execute_batch(RETRY_ROUNDED_WAV_INTERVALS_MIGRATION)
            .unwrap();
        connection
            .execute_batch(UNIFIED_INGRESS_QUEUE_MIGRATION)
            .unwrap();
        connection
            .execute_batch(STANDALONE_LIBRARY_MIGRATION)
            .unwrap();
        connection
            .execute_batch(DURABLE_TRANSCRIPT_PIECES_MIGRATION)
            .unwrap();
        let id = Uuid::new_v4();
        insert_recording(&connection, id, "transcribing");
        connection
            .execute(
                "INSERT INTO audio_transcript_pieces(
                    recording_id,attempt,piece_index,piece_count,audio_start_ms,
                    audio_end_ms,transcript_json,created_at
                 ) VALUES(?1,1,0,2,0,120000,?2,?3)",
                params![
                    id.to_string(),
                    r#"{"utterances":[{"original_text":"hello"}]}"#,
                    "2026-01-01T00:00:00Z",
                ],
            )
            .unwrap();

        apply_migrations(&connection).unwrap();

        let migrated: (String, String, String, String) = connection
            .query_row(
                "SELECT attempt_id,cache_revision,raw_gemini_response,parsed_json
                 FROM audio_transcript_pieces
                 WHERE recording_id=?1 AND piece_index=0",
                [id.to_string()],
                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
            )
            .unwrap();
        assert_eq!(migrated.0, "version-7-attempt-1");
        assert_eq!(migrated.1, "gemini-3.1-pro-transcription-v1");
        assert_ne!(migrated.1, PIECE_CACHE_REVISION);
        assert_eq!(migrated.2, "");
        assert_eq!(migrated.3, "");
    }

    #[test]
    fn status_exposes_a_completed_correction_packet() {
        let connection = database();
        let id = Uuid::new_v4();
        insert_recording(&connection, id, "ready_for_ingress");
        let packet = CorrectionPacket {
            recording_id: id,
            user_id: "test-user".into(),
            sha256: format!("{:064x}", 1),
            original_filename: "note.wav".into(),
            size_bytes: 4,
            recorded_at: DateTime::parse_from_rfc3339("2026-01-01T00:00:00Z")
                .unwrap()
                .with_timezone(&Utc),
            clean: false,
            chunk_count: 1,
            chunks: Vec::new(),
            confirmation_state: ConfirmationState::Unconfirmed,
        };
        connection
            .execute(
                "UPDATE audio_recordings
                 SET final_transcript='hello',correction_packet_json=?1 WHERE id=?2",
                params![serde_json::to_string(&packet).unwrap(), id.to_string()],
            )
            .unwrap();
        let status = connection
            .query_row(
                "SELECT id,user_id,sha256,original_filename,size_bytes,source_created_at,received_at,
                        status,gemini_model,reconciliation_model,reconciliation_reasoning,
                        attempt_count,last_error,failure_retryable,transcription_status_json,
                        final_transcript,correction_packet_json
                 FROM audio_recordings WHERE id=?1",
                [id.to_string()],
                row_recording_status,
            )
            .unwrap();
        assert!(matches!(
            status.state,
            RecordingState::Complete { ref transcript } if transcript == "hello"
        ));
        assert_eq!(status.correction_packet.unwrap().recording_id, id);
    }

    #[test]
    fn legacy_correction_packets_remain_reviewable_without_translating_features() {
        let connection = database();
        let id = Uuid::new_v4();
        insert_recording(&connection, id, "ready_for_ingress");
        let serialized = legacy_packet_json(id);
        connection
            .execute(
                "UPDATE audio_recordings
                 SET final_transcript='hello',correction_packet_json=?1 WHERE id=?2",
                params![serialized, id.to_string()],
            )
            .unwrap();

        let status = connection
            .query_row(
                "SELECT id,user_id,sha256,original_filename,size_bytes,source_created_at,received_at,
                        status,gemini_model,reconciliation_model,reconciliation_reasoning,
                        attempt_count,last_error,failure_retryable,transcription_status_json,
                        final_transcript,correction_packet_json
                 FROM audio_recordings WHERE id=?1",
                [id.to_string()],
                row_recording_status,
            )
            .unwrap();
        let packet = status.correction_packet.unwrap();
        assert_eq!(packet.confirmation_state, ConfirmationState::Unconfirmed);
        assert_eq!(packet.chunks[0].observations.len(), 1);
        assert!(packet.chunks[0].parsed.speakers[0].feature_row.is_none());
    }

    #[test]
    fn legacy_confirmation_preserves_obsolete_features_without_training_them() {
        use kcode_speaker_system::DeleteOutcome;

        let id = Uuid::new_v4();
        let original = legacy_packet_json(id);
        let mut stored = StoredCorrectionPacket::decode(&original).unwrap();
        let keys = stored.legacy_observation_keys().unwrap();
        let observation_key = stored.packet.chunks[0].observations[0]
            .observation_key
            .clone();
        let confirmation = RecordingConfirmation {
            recording_id: id,
            observations: vec![ObservationConfirmation {
                observation_key: observation_key.clone(),
                confirmed_full_name: "David Example".into(),
            }],
        };
        let path = classifier_path("legacy-confirmation");
        let classifier = SpeechClassifier::open(&path).unwrap();

        apply_confirmations(&classifier, &mut stored.packet, &confirmation, &keys).unwrap();
        let encoded = stored.encode().unwrap();
        let encoded_value: Value = serde_json::from_str(&encoded).unwrap();
        assert_eq!(
            encoded_value.pointer("/chunks/0/parsed/speakers/0/feature_row"),
            Some(&legacy_feature_row())
        );
        let decoded = StoredCorrectionPacket::decode(&encoded).unwrap().packet;
        assert_eq!(decoded.confirmation_state, ConfirmationState::Confirmed);
        assert_eq!(
            decoded.chunks[0].observations[0]
                .confirmed_full_name
                .as_deref(),
            Some("David Example")
        );
        assert_eq!(
            classifier.delete(observation_key).unwrap(),
            DeleteOutcome::NotFound
        );

        drop(classifier);
        remove_database(&path);
    }

    #[test]
    fn arbitrary_feature_objects_are_not_treated_as_legacy_packets() {
        let id = Uuid::new_v4();
        let mut value: Value = serde_json::from_str(&legacy_packet_json(id)).unwrap();
        value["chunks"][0]["parsed"]["speakers"][0]["feature_row"] =
            serde_json::json!({"unknown":1});
        assert!(StoredCorrectionPacket::decode(&value.to_string()).is_err());
    }

    #[test]
    fn transcript_pieces_preserve_raw_and_parsed_data_attempt_scoped() {
        let connection = database();
        let id = Uuid::new_v4();
        insert_recording(&connection, id, "transcribing");
        let db = Mutex::new(connection);
        let piece = sample_piece();
        persist_transcript_piece(&db, id, "attempt-a", &piece).unwrap();
        persist_transcript_piece(&db, id, "attempt-b", &piece).unwrap();

        let db = db.lock().unwrap();
        let rows: i64 = db
            .query_row(
                "SELECT COUNT(*) FROM audio_transcript_pieces WHERE recording_id=?1",
                [id.to_string()],
                |row| row.get(0),
            )
            .unwrap();
        let stored: (String, String) = db
            .query_row(
                "SELECT raw_gemini_response,parsed_json
                 FROM audio_transcript_pieces
                 WHERE recording_id=?1 AND attempt_id='attempt-a' AND piece_index=0",
                [id.to_string()],
                |row| Ok((row.get(0)?, row.get(1)?)),
            )
            .unwrap();
        assert_eq!(rows, 2);
        assert_eq!(stored.0, piece.raw_gemini_response);
        assert_eq!(
            serde_json::from_str::<ParsedChunk>(&stored.1).unwrap(),
            piece.parsed
        );
    }

    #[test]
    fn cache_reuses_only_exact_matching_new_revision_piece_plans() {
        let connection = database();
        let id = Uuid::new_v4();
        insert_recording(&connection, id, "transcribing");
        let db = Mutex::new(connection);
        let piece = sample_piece();
        persist_transcript_piece(&db, id, "attempt-a", &piece).unwrap();

        let path = classifier_path("cache");
        let classifier = Arc::new(SpeechClassifier::open(&path).unwrap());
        let context = classification_context(id, classifier);
        let exact = load_cached_transcript_piece(&db, id, piece.plan, &context).unwrap();
        let changed = load_cached_transcript_piece(
            &db,
            id,
            ChunkPlan {
                end_ms: piece.plan.end_ms + 1,
                ..piece.plan
            },
            &context,
        )
        .unwrap();

        assert_eq!(
            exact.unwrap().raw_gemini_response,
            piece.raw_gemini_response
        );
        assert!(changed.is_none());
        drop(context);
        remove_database(&path);
    }

    #[test]
    fn future_schema_versions_are_rejected() {
        let connection = Connection::open_in_memory().unwrap();
        connection
            .execute_batch("PRAGMA user_version = 11;")
            .unwrap();
        let error = apply_migrations(&connection).unwrap_err().to_string();
        assert!(error.contains("newer than supported"));
    }

    #[test]
    fn interrupted_attempts_consume_the_fixed_budget() {
        let connection = database();
        let id = Uuid::new_v4();
        insert_recording(&connection, id, "transcribing");
        connection
            .execute(
                "UPDATE audio_recordings SET attempt_count=5 WHERE id=?1",
                [id.to_string()],
            )
            .unwrap();
        recover_interrupted_attempts(&connection).unwrap();
        let state: String = connection
            .query_row(
                "SELECT status FROM audio_recordings WHERE id=?1",
                [id.to_string()],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(state, "failed");
    }

    #[test]
    fn filenames_cannot_escape_the_persistence_root() {
        assert_eq!(safe_filename(Some("../../secret.wav")), "secret.wav");
        assert_eq!(safe_filename(Some("meeting note.wav")), "meeting_note.wav");
        assert_eq!(safe_filename(None), "audio.wav");
    }
}