kcode-audio-ingress 0.7.2

//! 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, ChunkConfirmation, ConfirmationState, CorrectionChunk, CorrectionObservation,
    CorrectionPacket, FeatureRow, ObservationConfirmation, ObservationKey, ParsedChunk,
    ParsedSpeaker, SpeakerResolution,
};
use identity::{ClassificationContext, apply_confirmations, restore_packet_training};
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, PIECE_CACHE_REVISION, PieceCache, PieceSink};

mod identity;
mod legacy_review;
mod transcribe;
mod wav_slice;

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 LEGACY_REVIEW_ARCHIVE_MIGRATION: &str =
    include_str!("../migrations/011_legacy_review_archive.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 = 11;

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
);

CREATE TABLE audio_legacy_review_archive (
    recording_id TEXT PRIMARY KEY NOT NULL,
    final_transcript TEXT NOT NULL CHECK(length(trim(final_transcript)) > 0),
    correction_packet_json TEXT NOT NULL CHECK(length(correction_packet_json) > 0),
    archived_at TEXT NOT NULL
);

PRAGMA user_version = 11;
"#;

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

/// One exact source interval encoded for human speaker review.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct SpeakerReviewAudio {
    /// Browser-playable media type.
    pub content_type: &'static str,
    /// Safe presentation filename.
    pub filename: String,
    /// Complete WAV interval bytes in the retained source format.
    pub bytes: Vec<u8>,
}

/// Durable resolution for a finalized recording with an obsolete, unsigned packet.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum LegacyReviewDisposition {
    /// Archive the obsolete result and queue the retained WAV for current analysis.
    Reprocess,
    /// Preserve the accepted result and stop exposing its obsolete packet for review.
    Complete,
}

/// 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,
    },
    /// Gemini transcript and speaker candidates await per-chunk human signoff.
    AwaitingReview,
    /// 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 })
    }

    /// Reads the exact retained source interval for one correction-packet chunk.
    pub fn speaker_review_audio(
        &self,
        recording_id: Uuid,
        chunk_index: usize,
    ) -> Result<SpeakerReviewAudio, Error> {
        let stored = {
            let db = self.inner.db.lock().map_err(Error::internal)?;
            db.query_row(
                "SELECT original_relative_path,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((relative_path, packet_json)) = stored else {
            return Err(Error::not_found());
        };
        let packet = packet_json
            .ok_or_else(|| Error::conflict("Speaker review audio is not ready."))
            .and_then(|value| StoredCorrectionPacket::decode(&value).map_err(Error::internal))?;
        let chunk = packet
            .packet
            .chunks
            .iter()
            .find(|chunk| chunk.chunk_index == chunk_index)
            .ok_or_else(|| Error::invalid("Speaker review chunk does not exist."))?;
        let original =
            fs::File::open(self.inner.root.join(relative_path)).map_err(Error::internal)?;
        let bytes = wav_slice::interval(original, chunk.audio_start_ms, chunk.audio_end_ms)
            .map_err(Error::internal)?;
        Ok(SpeakerReviewAudio {
            content_type: "audio/wav",
            filename: format!(
                "speaker-review-{}-chunk-{}.wav",
                recording_id,
                chunk_index + 1
            ),
            bytes,
        })
    }

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

    /// Returns all distinct known speaker names in deterministic order.
    pub fn known_speakers(&self) -> Result<Vec<String>, Error> {
        self.inner
            .classifier
            .known_speakers()
            .map_err(Error::internal)
    }

    /// Resolves one finalized recording whose obsolete packet was never signed.
    ///
    /// The caller owns the downstream-ingress decision. Reprocessing archives
    /// the exact old transcript and packet before it clears the active result
    /// and queues the retained WAV. Completing preserves both payloads and
    /// marks the recording as an immutable legacy result.
    pub fn resolve_legacy_review(
        &self,
        recording_id: Uuid,
        disposition: LegacyReviewDisposition,
    ) -> Result<(), Error> {
        let mut db = self.inner.db.lock().map_err(Error::internal)?;
        match legacy_review::resolve(&mut db, recording_id, disposition, Utc::now())
            .map_err(Error::internal)?
        {
            legacy_review::Outcome::Applied | legacy_review::Outcome::Unchanged => Ok(()),
            legacy_review::Outcome::Missing => Err(Error::not_found()),
            legacy_review::Outcome::Ineligible => Err(Error::conflict(
                "Recording is not an unresolved finalized legacy review.",
            )),
        }
    }

    /// Applies exact known-or-unknown resolutions and signs off one review chunk.
    ///
    /// 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: ChunkConfirmation,
    ) -> 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,final_transcript
                 FROM audio_recordings WHERE id=?1",
                [recording_id.to_string()],
                |row| {
                    Ok((
                        row.get::<_, String>(0)?,
                        row.get::<_, Option<String>>(1)?,
                        row.get::<_, Option<String>>(2)?,
                    ))
                },
            )
            .optional()
            .map_err(Error::internal)?;
        let Some((status, packet_json, transcript)) = stored else {
            return Err(Error::not_found());
        };
        if status != "ready_for_ingress" || transcript.is_some() {
            return Err(Error::conflict(
                "Speaker confirmation requires a review-ready 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)?;
        let all_signed = stored.packet.confirmation_state == ConfirmationState::Confirmed;
        if let Err(error) = db.execute(
            "UPDATE audio_recordings
             SET correction_packet_json=?1,
                 status=CASE WHEN ?2 THEN 'reconciling' ELSE status END,
                 attempt_count=CASE WHEN ?2 THEN 0 ELSE attempt_count END,
                 next_attempt_at=NULL,last_error=NULL,updated_at=?3
             WHERE id=?4",
            params![
                updated_json,
                all_signed,
                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" => {
            match transcript.filter(|value| !value.trim().is_empty()) {
                Some(transcript) => RecordingState::Complete { transcript },
                None => RecordingState::AwaitingReview,
            }
        }
        "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 = if durable_status == "complete" {
        None
    } else {
        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> {
        let mut value: Value = serde_json::from_str(serialized)?;
        let confirmed = matches!(
            value.get("confirmation_state").and_then(Value::as_str),
            Some("confirmed" | "automatically_trained")
        );
        if let Some(object) = value.as_object_mut() {
            object.remove("clean");
        }
        let mut legacy_feature_rows = Vec::new();
        if let Some(chunks) = value.get_mut("chunks").and_then(Value::as_array_mut) {
            for (chunk_position, chunk) in chunks.iter_mut().enumerate() {
                if let Some(object) = chunk.as_object_mut() {
                    object.remove("clean");
                    object
                        .entry("signed_off")
                        .or_insert_with(|| Value::Bool(confirmed));
                }
                if let Some(observations) =
                    chunk.get_mut("observations").and_then(Value::as_array_mut)
                {
                    for observation in observations {
                        let Some(object) = observation.as_object_mut() else {
                            continue;
                        };
                        if !object.contains_key("resolution") {
                            let name = object
                                .get("confirmed_full_name")
                                .or_else(|| object.get("identified_full_name"))
                                .and_then(Value::as_str)
                                .filter(|_| confirmed)
                                .map(str::to_owned);
                            object.insert(
                                "resolution".into(),
                                name.map_or(Value::Null, |full_name| {
                                    serde_json::json!({"kind":"known","full_name":full_name})
                                }),
                            );
                        }
                        object.remove("confirmed_full_name");
                        object.remove("identified_full_name");
                        if let Some(candidate) =
                            object.get_mut("candidate").and_then(Value::as_object_mut)
                        {
                            if !candidate.contains_key("score") {
                                let score = candidate
                                    .remove("cost")
                                    .and_then(|value| value.as_f64())
                                    .map(|cost| Value::from(-cost));
                                if let Some(score) = score {
                                    candidate.insert("score".into(), score);
                                }
                            }
                            if !candidate.contains_key("runner_up_score") {
                                let runner = candidate
                                    .remove("runner_up_cost")
                                    .and_then(|value| value.as_f64())
                                    .map(|cost| Value::from(-cost))
                                    .unwrap_or(Value::Null);
                                candidate.insert("runner_up_score".into(), runner);
                            }
                            candidate.remove("confidence");
                            candidate.remove("runner_up_full_name");
                            candidate.remove("background_population_cost");
                        }
                    }
                }
                if let Some(speakers) = chunk
                    .pointer_mut("/parsed/speakers")
                    .and_then(Value::as_array_mut)
                {
                    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(),
                            });
                        }
                    }
                }
            }
        }
        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,
    correction_packet_json: Option<String>,
}

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,correction_packet_json
         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)?,
                row.get::<_, Option<String>>(8)?,
            ))
        },
    )
    .optional()?
    .map(
        |(
            id,
            user_id,
            sha256,
            original_filename,
            size_bytes,
            recorded_at,
            original_relative_path,
            attempt_count,
            correction_packet_json,
        )| {
            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,
                correction_packet_json,
            })
        },
    )
    .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 final_packet = recording
            .correction_packet_json
            .as_deref()
            .map(StoredCorrectionPacket::decode)
            .transpose()?
            .map(|stored| stored.packet)
            .filter(|packet| packet.confirmation_state == ConfirmationState::Confirmed);
        if let Some(packet) = final_packet {
            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='reconciling',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 job = inner
                .transcriber
                .finalize_durably(recording.user_id.clone(), packet);
            inner
                .jobs
                .lock()
                .map_err(|_| anyhow::anyhow!("audio transcription job lock was poisoned"))?
                .insert(recording.id, job.clone());
            job
        } else {
            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 piece_cache: PieceCache = Arc::new(move |plan| {
                load_cached_transcript_piece(&cache_db, cache_recording_id, plan)
            });

            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 |plan, raw| {
                persist_transcript_piece(&sink_db, sink_recording_id, &attempt_id, plan, raw)
            });

            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 progress = serde_json::to_string(&without_results(snapshot.clone()))?;
            let db = inner
                .db
                .lock()
                .map_err(|_| anyhow::anyhow!("AudioIngress database lock was poisoned"))?;
            if let Some(transcript) = snapshot.transcript.as_deref() {
                ensure!(
                    !transcript.trim().is_empty(),
                    "completed transcript is empty"
                );
                db.execute(
                    "UPDATE audio_recordings
                     SET status='ready_for_ingress',final_transcript=?1,
                         transcription_status_json=?2,next_attempt_at=NULL,last_error=NULL,
                         failure_retryable=1,updated_at=?3 WHERE id=?4",
                    params![
                        transcript.trim(),
                        progress,
                        Utc::now().to_rfc3339(),
                        recording.id.to_string()
                    ],
                )?;
                tracing::info!(recording_id=%recording.id, "Final audio transcript completed");
            } else {
                let packet = snapshot
                    .correction_packet
                    .as_ref()
                    .context("completed analysis omitted its correction packet")?;
                ensure!(
                    packet.recording_id == recording.id,
                    "correction packet belongs to another recording"
                );
                db.execute(
                    "UPDATE audio_recordings
                     SET status='ready_for_ingress',final_transcript=NULL,
                         correction_packet_json=?1,transcription_status_json=?2,
                         next_attempt_at=NULL,last_error=NULL,failure_retryable=1,updated_at=?3
                     WHERE id=?4",
                    params![
                        serde_json::to_string(packet)?,
                        progress,
                        Utc::now().to_rfc3339(),
                        recording.id.to_string()
                    ],
                )?;
                tracing::info!(recording_id=%recording.id, "Audio chunks are ready for speaker review");
            }
            remove_job(inner, recording.id)?;
            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=CASE
                 WHEN status='reconciling' AND correction_packet_json IS NOT NULL
                      AND final_transcript IS NULL THEN 'reconciling'
                 ELSE 'uploaded'
             END,
             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,
) -> anyhow::Result<Option<String>> {
    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
             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| row.get::<_, String>(0),
        )
        .optional()?
    };
    let Some(raw_gemini_response) = stored else {
        return Ok(None);
    };
    ensure!(
        !raw_gemini_response.trim().is_empty(),
        "cached piece omitted its raw Gemini response"
    );
    Ok(Some(raw_gemini_response))
}

fn persist_transcript_piece(
    db: &Mutex<Connection>,
    recording_id: Uuid,
    attempt_id: &str,
    plan: ChunkPlan,
    raw_gemini_response: &str,
) -> anyhow::Result<()> {
    ensure!(!attempt_id.is_empty(), "piece attempt identity is empty");
    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 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,
            raw_gemini_response,
            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 {
    if snapshot.steps.len() == 1 && snapshot.steps[0].step == Step::ReconcileTranscript {
        return "reconciling";
    }
    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 { .. }))
        .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 status='reconciling' AND correction_packet_json IS NOT NULL
                      AND final_transcript IS NULL AND attempt_count<?1 THEN 'reconciling'
                 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)?;
    }
    if version < 11 {
        connection.execute_batch(LEGACY_REVIEW_ARCHIVE_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
    }
}