kcode-audio-session-ingress 0.2.3

#![forbid(unsafe_code)]

use std::collections::HashMap;

use chrono::{DateTime, Utc};
use kcode_audio_history_handoff::{
    Error as HandoffError, Handoff, PieceProjection, RecordingProjection,
};
use kcode_audio_ingress::{
    AudioIngress, AudioInput, ConfirmationState, CorrectionPacket, ErrorKind as AudioErrorKind,
    RecordingConfirmation, RecordingState, RecordingStatus,
};
use kcode_session_history::{SessionHistory, SessionRecord};
use serde_json::Value;
use uuid::Uuid;

/// Application policy needed to coordinate audio with Session History.
#[derive(Clone, Debug)]
pub struct Config {
    /// Stable application user identifier attributed to accepted recordings.
    pub user_id: String,
    /// Effective context window available to an audio-ingress session.
    pub effective_context_tokens: u64,
}

/// Admitted recording bytes and source metadata.
#[derive(Clone, Debug)]
pub struct RecordingInput {
    /// Complete WAV bytes already accepted by the transport.
    pub bytes: Vec<u8>,
    /// Instant at which the original recording began.
    pub recorded_at: DateTime<Utc>,
    /// Original leaf filename, when known.
    pub original_filename: Option<String>,
}

/// Result of durably submitting one recording.
#[derive(Clone, Debug)]
pub struct RecordingSubmission {
    /// Current combined state after submission.
    pub recording: Recording,
    /// Whether AudioIngress already knew the same bytes.
    pub deduplicated: bool,
}

/// Combined recording and memory-ingress status.
#[derive(Clone, Debug)]
pub struct Recording {
    pub id: Uuid,
    pub sha256: String,
    pub original_filename: String,
    pub content_type: &'static str,
    pub size_bytes: u64,
    pub source_created_at: String,
    pub received_at: String,
    pub updated_at: String,
    pub status: String,
    pub transcription_model: String,
    pub reconciliation_model: String,
    pub reconciliation_reasoning: String,
    pub transcription_status: Option<Value>,
    pub attempt_count: i64,
    pub next_attempt_at: Option<String>,
    pub last_error: Option<String>,
    pub speaker_review: Option<SpeakerReview>,
    pub transcript_piece_count: usize,
    pub completed_piece_count: usize,
}

/// Bounded review summary for one classifier-aware recording.
#[derive(Clone, Debug)]
pub struct SpeakerReview {
    pub clean: bool,
    pub confirmation_state: ConfirmationState,
    pub observation_count: usize,
}

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

/// Detailed state for one recording.
#[derive(Clone, Debug)]
pub struct RecordingHistory {
    pub recording: Recording,
    pub final_transcript: Option<String>,
    pub correction_packet: Option<CorrectionPacket>,
    pub pieces: Vec<IngressPiece>,
}

/// Request to retry one transcript piece's memory ingress.
#[derive(Clone, Debug)]
pub struct RetryIngress {
    pub piece_id: String,
    pub expected_version: i64,
    /// Optional state retained for 0.2 wire compatibility. When supplied, it
    /// must exactly equal the current retained Session History state.
    pub state: Option<Value>,
}

/// Stable coordinator error category for transport mapping.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ErrorKind {
    InvalidInput,
    NotFound,
    Conflict,
    Internal,
}

/// Error returned by the audio/session-ingress coordinator.
#[derive(Debug)]
pub struct Error {
    kind: ErrorKind,
    message: String,
}

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

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

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

    fn not_found() -> Self {
        Self::new(
            ErrorKind::NotFound,
            "Audio recording or transcript piece not found.",
        )
    }

    fn internal(error: impl std::fmt::Display) -> Self {
        tracing::warn!(%error, "Audio session ingress operation failed");
        Self::new(
            ErrorKind::Internal,
            "An unexpected audio session ingress error occurred.",
        )
    }

    pub fn kind(&self) -> ErrorKind {
        self.kind
    }

    pub fn message(&self) -> &str {
        &self.message
    }
}

impl std::fmt::Display for Error {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str(&self.message)
    }
}

impl std::error::Error for Error {}

/// Cloneable typed coordinator over AudioIngress and Session History.
#[derive(Clone)]
pub struct Coordinator {
    audio: AudioIngress,
    handoff: Handoff,
    user_id: String,
}

impl Coordinator {
    pub fn new(
        audio: AudioIngress,
        history: SessionHistory,
        config: Config,
    ) -> Result<Self, Error> {
        let handoff = Handoff::new(
            history,
            config.user_id.clone(),
            config.effective_context_tokens,
        )
        .map_err(handoff_error)?;
        Ok(Self {
            audio,
            handoff,
            user_id: config.user_id,
        })
    }

    pub fn health(&self) -> Result<(), Error> {
        self.audio.status().map_err(audio_error)?;
        Ok(())
    }

    pub async fn submit(&self, input: RecordingInput) -> Result<RecordingSubmission, Error> {
        let submission = self
            .audio
            .submit(AudioInput {
                user_id: self.user_id.clone(),
                bytes: input.bytes,
                recorded_at: input.recorded_at,
                original_filename: input.original_filename,
            })
            .await
            .map_err(audio_error)?;
        let recording_status = self
            .audio
            .status()
            .map_err(audio_error)?
            .recordings
            .into_iter()
            .find(|recording| recording.id == submission.recording_id)
            .ok_or_else(Error::not_found)?;
        validate_submission_owner(&recording_status, &self.user_id)?;

        let projection = self
            .handoff
            .project(std::slice::from_ref(&recording_status))
            .await
            .map_err(handoff_error)?
            .into_iter()
            .next()
            .ok_or_else(|| Error::internal("handoff omitted recording projection"))?;
        let projection = IngressProjection::from_handoff(&recording_status, projection);
        Ok(RecordingSubmission {
            recording: Recording::from_status(recording_status, &projection),
            deduplicated: submission.deduplicated,
        })
    }

    pub async fn recordings(&self) -> Result<Vec<Recording>, Error> {
        let recordings = self
            .audio
            .status()
            .map_err(audio_error)?
            .recordings
            .into_iter()
            .filter(|recording| recording_belongs_to(recording, &self.user_id))
            .collect::<Vec<_>>();
        let projections = self
            .handoff
            .project(&recordings)
            .await
            .map_err(handoff_error)?;
        if recordings.len() != projections.len() {
            return Err(Error::internal("handoff returned incomplete projections"));
        }
        Ok(recordings
            .into_iter()
            .zip(projections)
            .map(|(recording, projection)| {
                let projection = IngressProjection::from_handoff(&recording, projection);
                Recording::from_status(recording, &projection)
            })
            .collect())
    }

    pub async fn recording_by_sha256(&self, sha256: &str) -> Result<Recording, Error> {
        if sha256.len() != 64 || !sha256.bytes().all(|byte| byte.is_ascii_hexdigit()) {
            return Err(Error::invalid(
                "audio SHA-256 must contain exactly 64 hexadecimal characters",
            ));
        }
        let normalized = sha256.to_ascii_lowercase();
        self.recordings()
            .await?
            .into_iter()
            .find(|recording| recording.sha256 == normalized)
            .ok_or_else(Error::not_found)
    }

    pub async fn recording_history(&self, recording_id: Uuid) -> Result<RecordingHistory, Error> {
        let recording_status = self
            .audio
            .status()
            .map_err(audio_error)?
            .recordings
            .into_iter()
            .find(|recording| {
                recording.id == recording_id && recording_belongs_to(recording, &self.user_id)
            })
            .ok_or_else(Error::not_found)?;
        let projection = self
            .handoff
            .project(std::slice::from_ref(&recording_status))
            .await
            .map_err(handoff_error)?
            .into_iter()
            .next()
            .ok_or_else(|| Error::internal("handoff omitted recording projection"))?;
        let projection = IngressProjection::from_handoff(&recording_status, projection);
        let final_transcript = match &recording_status.state {
            RecordingState::Complete { transcript } => Some(transcript.clone()),
            _ => None,
        };
        let correction_packet = recording_status.correction_packet.clone();
        let recording = Recording::from_status(recording_status, &projection);
        Ok(RecordingHistory {
            recording,
            final_transcript,
            correction_packet,
            pieces: projection.pieces,
        })
    }

    pub fn retry_recording(&self, recording_id: Uuid) -> Result<(), Error> {
        let owned = self
            .audio
            .status()
            .map_err(audio_error)?
            .recordings
            .into_iter()
            .any(|recording| {
                recording.id == recording_id && recording_belongs_to(&recording, &self.user_id)
            });
        if !owned {
            return Err(Error::not_found());
        }
        self.audio.retry(recording_id).map_err(audio_error)
    }

    pub async fn confirm_speakers(
        &self,
        confirmation: RecordingConfirmation,
    ) -> Result<CorrectionPacket, Error> {
        let recording_id = confirmation.recording_id;
        let recording = self
            .audio
            .status()
            .map_err(audio_error)?
            .recordings
            .into_iter()
            .find(|recording| {
                recording.id == recording_id && recording_belongs_to(recording, &self.user_id)
            })
            .ok_or_else(Error::not_found)?;

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

        let packet = self
            .audio
            .confirm_speakers(confirmation)
            .map_err(audio_error)?;
        let recording = self
            .audio
            .status()
            .map_err(audio_error)?
            .recordings
            .into_iter()
            .find(|recording| {
                recording.id == recording_id && recording_belongs_to(recording, &self.user_id)
            })
            .ok_or_else(Error::not_found)?;
        self.handoff
            .synchronize(std::slice::from_ref(&recording))
            .await
            .map_err(handoff_error)?;
        Ok(packet)
    }

    pub async fn retry_ingress(&self, input: RetryIngress) -> Result<SessionRecord, Error> {
        let recordings = self.audio.status().map_err(audio_error)?.recordings;
        self.handoff
            .retry(
                &recordings,
                &input.piece_id,
                input.expected_version,
                input.state,
            )
            .await
            .map_err(handoff_error)
    }

    pub async fn synchronize_completed_transcripts(&self) -> Result<(), Error> {
        let recordings = self.audio.status().map_err(audio_error)?.recordings;
        self.handoff
            .synchronize(&recordings)
            .await
            .map_err(handoff_error)
    }
}

fn recording_belongs_to(recording: &RecordingStatus, user_id: &str) -> bool {
    recording.user_id == user_id
}

fn validate_submission_owner(recording: &RecordingStatus, user_id: &str) -> Result<(), Error> {
    if recording_belongs_to(recording, user_id) {
        Ok(())
    } else {
        Err(Error::conflict(
            "identical audio is already attributed to another user",
        ))
    }
}

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

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

impl IngressProjection {
    fn from_handoff(recording: &RecordingStatus, projection: RecordingProjection) -> Self {
        Self {
            pieces: projection
                .pieces
                .into_iter()
                .map(|piece| IngressPiece::from_projection(recording, piece))
                .collect(),
        }
    }
}

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

impl IngressPiece {
    fn from_projection(recording: &RecordingStatus, piece: PieceProjection) -> Self {
        let record = piece.record;
        Self {
            id: record.id,
            recording_id: recording.id,
            sha256: recording.sha256.clone(),
            original_filename: recording.original_filename.clone(),
            source_created_at: recording.recorded_at.to_rfc3339(),
            piece_index: piece.piece_index,
            piece_count: piece.piece_count,
            transcript_text: piece.transcript_text,
            estimated_tokens: piece.estimated_tokens,
            phase: record.phase,
            provenance_id: record.provenance_id,
            state: record.state,
            version: record.version,
            ingress_failure_count: record.ingress_failure_count,
            ingress_failures: record.ingress_failures,
            created_at: record.started_at,
            updated_at: record.updated_at,
        }
    }
}

fn processing_stage(status: &kcode_audio_ingress::TranscriptionStatus) -> &'static str {
    let plan_complete = status.steps.iter().any(|entry| {
        entry.step == kcode_audio_ingress::Step::PlanChunks
            && entry.state == kcode_audio_ingress::StepState::Completed
    });
    if !plan_complete {
        return "chunking";
    }
    let chunks_complete = status
        .steps
        .iter()
        .filter(|entry| {
            matches!(
                entry.step,
                kcode_audio_ingress::Step::TranscribeChunk { .. }
            )
        })
        .all(|entry| entry.state == kcode_audio_ingress::StepState::Completed);
    if chunks_complete {
        let analyses_complete = status
            .steps
            .iter()
            .filter(|entry| matches!(entry.step, kcode_audio_ingress::Step::ParseChunk { .. }))
            .all(|entry| entry.state == kcode_audio_ingress::StepState::Completed);
        if !analyses_complete {
            return "analyzing_speakers";
        }
        let training_active = status.steps.iter().any(|entry| {
            entry.step == kcode_audio_ingress::Step::TrainIdentities
                && matches!(
                    entry.state,
                    kcode_audio_ingress::StepState::Running
                        | kcode_audio_ingress::StepState::Retrying
                )
        });
        if training_active {
            "training_speakers"
        } else {
            "reconciling"
        }
    } else {
        "transcribing"
    }
}

fn audio_error(error: kcode_audio_ingress::Error) -> Error {
    match error.kind() {
        AudioErrorKind::InvalidInput => Error::new(ErrorKind::InvalidInput, error.to_string()),
        AudioErrorKind::NotFound => Error::new(ErrorKind::NotFound, error.to_string()),
        AudioErrorKind::Conflict => Error::new(ErrorKind::Conflict, error.to_string()),
        AudioErrorKind::Internal => Error::internal(error),
    }
}

fn handoff_error(error: HandoffError) -> Error {
    match error {
        HandoffError::InvalidInput(message) => Error::new(ErrorKind::InvalidInput, message),
        HandoffError::NotFound(message) => Error::new(ErrorKind::NotFound, message),
        HandoffError::Conflict(message) => Error::new(ErrorKind::Conflict, message),
        HandoffError::Internal(message) => Error::new(ErrorKind::Internal, message),
    }
}

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

    use super::*;

    struct TestRoot(PathBuf);

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

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

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

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

    fn completed_recording_for_user(
        user_id: impl Into<String>,
        transcript: impl Into<String>,
    ) -> RecordingStatus {
        let now = Utc::now();
        RecordingStatus {
            id: Uuid::new_v4(),
            user_id: user_id.into(),
            sha256: "0".repeat(64),
            original_filename: "meeting.final.WAV".into(),
            size_bytes: 42,
            recorded_at: now,
            received_at: now,
            transcription_model: "transcription-model".into(),
            reconciliation_model: "reconciliation-model".into(),
            reconciliation_reasoning: "xhigh".into(),
            state: RecordingState::Complete {
                transcript: transcript.into(),
            },
            correction_packet: None,
        }
    }

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

    #[test]
    fn recording_filter_is_exactly_scoped_to_the_configured_user() {
        let own = completed_recording_for_user("own-user", "Own transcript");
        let foreign = completed_recording_for_user("foreign-user", "Foreign transcript");
        let visible = [own.clone(), foreign]
            .into_iter()
            .filter(|recording| recording_belongs_to(recording, "own-user"))
            .collect::<Vec<_>>();
        assert_eq!(visible.len(), 1);
        assert_eq!(visible[0].id, own.id);
    }

    #[test]
    fn cross_user_sha_deduplication_fails_closed() {
        let mut own = completed_recording_for_user("own-user", "Own transcript");
        own.sha256 = "a".repeat(64);
        let mut foreign = completed_recording_for_user("foreign-user", "Foreign transcript");
        foreign.sha256 = own.sha256.clone();

        assert!(validate_submission_owner(&own, "own-user").is_ok());
        let error = validate_submission_owner(&foreign, "own-user").unwrap_err();
        assert_eq!(error.kind(), ErrorKind::Conflict);
        assert_eq!(
            error.message(),
            "identical audio is already attributed to another user"
        );
    }

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

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

    #[tokio::test]
    async fn facade_adapts_handoff_projection_and_status_without_drift() {
        let root = TestRoot::new();
        let history = history(root.path());
        let recording = completed_recording_for_user("user", "Transcript");
        let handoff = Handoff::new(history.clone(), "user".into(), 400).unwrap();
        handoff
            .synchronize(std::slice::from_ref(&recording))
            .await
            .unwrap();

        let projection = handoff
            .project(std::slice::from_ref(&recording))
            .await
            .unwrap()
            .remove(0);
        let projection = IngressProjection::from_handoff(&recording, projection);
        let pending = Recording::from_status(recording.clone(), &projection);
        assert_eq!(pending.status, "ready_for_ingress");
        assert_eq!(pending.transcript_piece_count, 1);
        assert_eq!(pending.completed_piece_count, 0);
        assert_eq!(projection.pieces[0].recording_id, recording.id);
        assert_eq!(projection.pieces[0].sha256, recording.sha256);
        assert_eq!(projection.pieces[0].piece_index, 0);
        assert_eq!(projection.pieces[0].piece_count, 1);
        assert_eq!(projection.pieces[0].transcript_text, "Transcript");
        assert_eq!(projection.pieces[0].phase, "ingress_pending");

        let record = &projection.pieces[0];
        history
            .start_ingress(
                &record.id,
                kcode_session_history::StartIngress {
                    expected_version: record.version,
                    provenance_id: "test:facade-adaptation".into(),
                },
            )
            .await
            .unwrap();
        let projection = handoff
            .project(std::slice::from_ref(&recording))
            .await
            .unwrap()
            .remove(0);
        let projection = IngressProjection::from_handoff(&recording, projection);
        let ingressing = Recording::from_status(recording, &projection);
        assert_eq!(ingressing.status, "ingressing");
        assert_eq!(projection.pieces[0].phase, "ingress_in_progress");
    }

    #[test]
    fn handoff_errors_preserve_category_and_message() {
        let cases = [
            (
                HandoffError::InvalidInput("invalid".into()),
                ErrorKind::InvalidInput,
                "invalid",
            ),
            (
                HandoffError::NotFound("missing".into()),
                ErrorKind::NotFound,
                "missing",
            ),
            (
                HandoffError::Conflict("conflict".into()),
                ErrorKind::Conflict,
                "conflict",
            ),
            (
                HandoffError::Internal("raw storage message".into()),
                ErrorKind::Internal,
                "raw storage message",
            ),
        ];

        for (source, kind, message) in cases {
            let error = handoff_error(source);
            assert_eq!(error.kind(), kind);
            assert_eq!(error.message(), message);
        }
    }
}