kcode-audio-session-view 0.1.0

#![forbid(unsafe_code)]

//! Deterministic adaptation of Audio Ingress and History Handoff projections.

use kcode_audio_history_handoff::{PieceProjection, RecordingProjection};
use kcode_audio_ingress::{
    ConfirmationState, RecordingState, RecordingStatus, Step, StepState, TranscriptionStatus,
};
use serde_json::Value;
use uuid::Uuid;

/// Established facade view of one audio recording.
#[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,
}

/// Complete deterministic facade view for one recording and its projected pieces.
#[derive(Clone, Debug)]
pub struct RecordingView {
    pub recording: Recording,
    pub pieces: Vec<IngressPiece>,
}

/// Adapts one Audio Ingress status and its matching History Handoff projection.
pub fn render(recording: RecordingStatus, projection: RecordingProjection) -> RecordingView {
    let RecordingProjection {
        recording_id: _,
        pieces: projected_pieces,
    } = projection;
    let pieces = projected_pieces
        .into_iter()
        .map(|piece| adapt_piece(&recording, piece))
        .collect::<Vec<_>>();
    let recording = adapt_recording(recording, &pieces);
    RecordingView { recording, pieces }
}

fn adapt_recording(recording: RecordingStatus, pieces: &[IngressPiece]) -> Recording {
    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 !pieces.is_empty() {
        status = if pieces.iter().all(|piece| piece.phase == "complete") {
            "complete".into()
        } else if pieces.iter().any(|piece| piece.phase == "ingress_failed") {
            "ingress_failed".into()
        } else if pieces
            .iter()
            .any(|piece| piece.phase == "ingress_in_progress")
        {
            "ingressing".into()
        } else {
            "ready_for_ingress".into()
        };
    }
    let completed_piece_count = pieces
        .iter()
        .filter(|piece| piece.phase == "complete")
        .count();

    Recording {
        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: pieces.len(),
        completed_piece_count,
    }
}

fn adapt_piece(recording: &RecordingStatus, piece: PieceProjection) -> IngressPiece {
    let record = piece.record;
    IngressPiece {
        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: &TranscriptionStatus) -> &'static str {
    let plan_complete = status
        .steps
        .iter()
        .any(|entry| entry.step == Step::PlanChunks && entry.state == StepState::Completed);
    if !plan_complete {
        return "chunking";
    }

    let chunks_complete = status
        .steps
        .iter()
        .filter(|entry| matches!(entry.step, Step::TranscribeChunk { .. }))
        .all(|entry| entry.state == StepState::Completed);
    if !chunks_complete {
        return "transcribing";
    }

    let analyses_complete = status
        .steps
        .iter()
        .filter(|entry| matches!(entry.step, Step::ParseChunk { .. }))
        .all(|entry| entry.state == StepState::Completed);
    if !analyses_complete {
        return "analyzing_speakers";
    }

    let training_active = status.steps.iter().any(|entry| {
        entry.step == Step::TrainIdentities
            && matches!(entry.state, StepState::Running | StepState::Retrying)
    });
    if training_active {
        "training_speakers"
    } else {
        "reconciling"
    }
}

#[cfg(test)]
mod tests {
    use chrono::{DateTime, Utc};
    use kcode_audio_history_handoff::{PieceProjection, RecordingProjection};
    use kcode_audio_ingress::{
        ConfirmationState, CorrectionChunk, CorrectionObservation, CorrectionPacket, JobState,
        ObservationKey, ParsedChunk, RecordingState, RecordingStatus, Step, StepState, StepStatus,
        TranscriptionStatus,
    };
    use kcode_session_history::SessionRecord;
    use serde_json::json;
    use uuid::Uuid;

    use super::render;

    fn timestamp(value: &str) -> DateTime<Utc> {
        DateTime::parse_from_rfc3339(value)
            .unwrap()
            .with_timezone(&Utc)
    }

    fn recording(state: RecordingState) -> RecordingStatus {
        RecordingStatus {
            id: Uuid::parse_str("b067a460-69bb-4c49-a9d7-037715d3137d").unwrap(),
            user_id: "user".into(),
            sha256: "ab".repeat(32),
            original_filename: "meeting.final.WAV".into(),
            size_bytes: 42,
            recorded_at: timestamp("2026-01-02T03:04:05Z"),
            received_at: timestamp("2026-01-02T03:05:06Z"),
            transcription_model: "transcription-model".into(),
            reconciliation_model: "reconciliation-model".into(),
            reconciliation_reasoning: "xhigh".into(),
            state,
            correction_packet: None,
        }
    }

    fn with_speaker_packet(
        mut recording: RecordingStatus,
        confirmation_state: ConfirmationState,
    ) -> RecordingStatus {
        let observations = (0..2)
            .map(|speaker_ordinal| CorrectionObservation {
                local_label: format!("Speaker {speaker_ordinal}"),
                speaker_ordinal,
                observation_key: ObservationKey {
                    object_id: format!("observation-{speaker_ordinal}"),
                    piece_index: speaker_ordinal,
                },
                candidate: None,
                identified_full_name: None,
                confirmed_full_name: None,
            })
            .collect();
        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![CorrectionChunk {
                chunk_index: 0,
                chunk_count: 1,
                audio_start_ms: 0,
                audio_end_ms: 1_000,
                raw_gemini_response: "raw".into(),
                parsed: ParsedChunk {
                    utterances: Vec::new(),
                    notes: Vec::new(),
                    clip_valid: true,
                    clip_validity_reason: None,
                    speakers: Vec::new(),
                },
                observations,
                clean: true,
            }],
            confirmation_state,
        });
        recording
    }

    fn projection(recording_id: Uuid, phases: &[&str]) -> RecordingProjection {
        let piece_count = u32::try_from(phases.len()).unwrap();
        RecordingProjection {
            recording_id,
            pieces: phases
                .iter()
                .enumerate()
                .map(|(index, phase)| PieceProjection {
                    piece_index: u32::try_from(index).unwrap(),
                    piece_count,
                    transcript_text: format!("Transcript piece {index}"),
                    estimated_tokens: 10 + u64::try_from(index).unwrap(),
                    record: SessionRecord {
                        id: format!("session-{index}"),
                        phase: (*phase).into(),
                        started_at: format!("2026-01-02T03:0{index}:00Z"),
                        updated_at: format!("2026-01-02T04:0{index}:00Z"),
                        state: json!({"piece":index}),
                        provenance_id: Some(format!("provenance-{index}")),
                        version: 7 + i64::try_from(index).unwrap(),
                        last_user_message_at: None,
                        ended_at: None,
                        ingress_failure_count: 2 + i64::try_from(index).unwrap(),
                        ingress_failures: json!([{"piece":index,"message":"retained"}]),
                        ingress_next_attempt_at: None,
                        summary: false,
                    },
                })
                .collect(),
        }
    }

    fn empty_projection(recording_id: Uuid) -> RecordingProjection {
        projection(recording_id, &[])
    }

    #[test]
    fn queued_failed_complete_and_speaker_review_states_are_preserved() {
        let queued = recording(RecordingState::Queued);
        let queued_id = queued.id;
        let queued = render(queued, empty_projection(queued_id)).recording;
        assert_eq!(queued.status, "uploaded");
        assert_eq!(queued.attempt_count, 0);
        assert!(queued.last_error.is_none());

        let failed = recording(RecordingState::Failed {
            attempts: 3,
            error: "provider stopped".into(),
            retryable: true,
        });
        let failed_id = failed.id;
        let failed = render(failed, empty_projection(failed_id)).recording;
        assert_eq!(failed.status, "failed");
        assert_eq!(failed.attempt_count, 3);
        assert_eq!(failed.last_error.as_deref(), Some("provider stopped"));

        let complete = recording(RecordingState::Complete {
            transcript: "Transcript".into(),
        });
        let complete_id = complete.id;
        let complete = render(complete, projection(complete_id, &["complete"])).recording;
        assert_eq!(complete.status, "complete");

        let awaiting = with_speaker_packet(
            recording(RecordingState::Complete {
                transcript: "Transcript".into(),
            }),
            ConfirmationState::Unconfirmed,
        );
        let awaiting_id = awaiting.id;
        let awaiting = render(awaiting, empty_projection(awaiting_id)).recording;
        assert_eq!(awaiting.status, "speaker_review");
        let review = awaiting.speaker_review.unwrap();
        assert!(review.clean);
        assert_eq!(review.confirmation_state, ConfirmationState::Unconfirmed);
        assert_eq!(review.observation_count, 2);

        let confirmed = with_speaker_packet(
            recording(RecordingState::Complete {
                transcript: "Transcript".into(),
            }),
            ConfirmationState::Confirmed,
        );
        let confirmed_id = confirmed.id;
        let confirmed = render(confirmed, empty_projection(confirmed_id)).recording;
        assert_eq!(confirmed.status, "ready_for_ingress");
    }

    #[test]
    fn piece_phase_priority_and_completed_counts_are_exact() {
        let cases = [
            (vec!["complete", "complete"], "complete", 2),
            (
                vec!["complete", "ingress_in_progress", "ingress_failed"],
                "ingress_failed",
                1,
            ),
            (
                vec!["complete", "ingress_pending", "ingress_in_progress"],
                "ingressing",
                1,
            ),
            (vec!["complete", "ingress_pending"], "ready_for_ingress", 1),
        ];

        for (phases, expected_status, expected_completed) in cases {
            let recording = recording(RecordingState::Complete {
                transcript: "Transcript".into(),
            });
            let recording_id = recording.id;
            let view = render(recording, projection(recording_id, &phases));
            assert_eq!(view.recording.status, expected_status);
            assert_eq!(view.recording.completed_piece_count, expected_completed);
            assert_eq!(view.recording.transcript_piece_count, phases.len());
        }
    }

    #[test]
    fn recording_and_piece_fields_are_adapted_without_drift() {
        let recording = recording(RecordingState::Complete {
            transcript: "Transcript".into(),
        });
        let recording_id = recording.id;
        let projection_id = Uuid::parse_str("2456325d-b033-4caf-92f0-e044080ed9b8").unwrap();
        let view = render(recording, projection(projection_id, &["ingress_failed"]));

        let adapted = &view.recording;
        assert_eq!(adapted.id, recording_id);
        assert_eq!(adapted.sha256, "ab".repeat(32));
        assert_eq!(adapted.original_filename, "meeting.final.WAV");
        assert_eq!(adapted.content_type, "audio/wav");
        assert_eq!(adapted.size_bytes, 42);
        assert_eq!(adapted.source_created_at, "2026-01-02T03:04:05+00:00");
        assert_eq!(adapted.received_at, "2026-01-02T03:05:06+00:00");
        assert_eq!(adapted.updated_at, adapted.received_at);
        assert_eq!(adapted.status, "ingress_failed");
        assert_eq!(adapted.transcription_model, "transcription-model");
        assert_eq!(adapted.reconciliation_model, "reconciliation-model");
        assert_eq!(adapted.reconciliation_reasoning, "xhigh");
        assert!(adapted.transcription_status.is_none());
        assert_eq!(adapted.attempt_count, 0);
        assert!(adapted.next_attempt_at.is_none());
        assert!(adapted.last_error.is_none());
        assert!(adapted.speaker_review.is_none());
        assert_eq!(adapted.transcript_piece_count, 1);
        assert_eq!(adapted.completed_piece_count, 0);

        let piece = &view.pieces[0];
        assert_eq!(piece.id, "session-0");
        assert_eq!(piece.recording_id, recording_id);
        assert_eq!(piece.sha256, "ab".repeat(32));
        assert_eq!(piece.original_filename, "meeting.final.WAV");
        assert_eq!(piece.source_created_at, "2026-01-02T03:04:05+00:00");
        assert_eq!(piece.piece_index, 0);
        assert_eq!(piece.piece_count, 1);
        assert_eq!(piece.transcript_text, "Transcript piece 0");
        assert_eq!(piece.estimated_tokens, 10);
        assert_eq!(piece.phase, "ingress_failed");
        assert_eq!(piece.provenance_id.as_deref(), Some("provenance-0"));
        assert_eq!(piece.state, json!({"piece":0}));
        assert_eq!(piece.version, 7);
        assert_eq!(piece.ingress_failure_count, 2);
        assert_eq!(
            piece.ingress_failures,
            json!([{"piece":0,"message":"retained"}])
        );
        assert_eq!(piece.created_at, "2026-01-02T03:00:00Z");
        assert_eq!(piece.updated_at, "2026-01-02T04:00:00Z");
    }

    fn step(step: Step, state: StepState) -> StepStatus {
        StepStatus {
            step,
            state,
            attempts: 1,
            retry_after: None,
            error: None,
        }
    }

    fn progress(steps: Vec<StepStatus>) -> TranscriptionStatus {
        TranscriptionStatus {
            state: JobState::Running,
            steps,
            transcript: None,
            correction_packet: None,
        }
    }

    #[test]
    fn processing_stage_mapping_and_status_json_are_preserved() {
        let cases = [
            (
                "chunking",
                progress(vec![step(Step::PlanChunks, StepState::Running)]),
            ),
            (
                "transcribing",
                progress(vec![
                    step(Step::PlanChunks, StepState::Completed),
                    step(
                        Step::TranscribeChunk { index: 0, total: 1 },
                        StepState::Running,
                    ),
                ]),
            ),
            (
                "analyzing_speakers",
                progress(vec![
                    step(Step::PlanChunks, StepState::Completed),
                    step(
                        Step::TranscribeChunk { index: 0, total: 1 },
                        StepState::Completed,
                    ),
                    step(Step::ParseChunk { index: 0, total: 1 }, StepState::Running),
                ]),
            ),
            (
                "training_speakers",
                progress(vec![
                    step(Step::PlanChunks, StepState::Completed),
                    step(
                        Step::TranscribeChunk { index: 0, total: 1 },
                        StepState::Completed,
                    ),
                    step(
                        Step::ParseChunk { index: 0, total: 1 },
                        StepState::Completed,
                    ),
                    step(Step::TrainIdentities, StepState::Retrying),
                ]),
            ),
            (
                "reconciling",
                progress(vec![
                    step(Step::PlanChunks, StepState::Completed),
                    step(
                        Step::TranscribeChunk { index: 0, total: 1 },
                        StepState::Completed,
                    ),
                    step(
                        Step::ParseChunk { index: 0, total: 1 },
                        StepState::Completed,
                    ),
                    step(Step::TrainIdentities, StepState::Completed),
                ]),
            ),
        ];

        for (expected_stage, progress) in cases {
            let expected_json = serde_json::to_value(&progress).unwrap();
            let recording = recording(RecordingState::Processing {
                attempt: 4,
                progress,
            });
            let recording_id = recording.id;
            let adapted = render(recording, empty_projection(recording_id)).recording;
            assert_eq!(adapted.status, expected_stage);
            assert_eq!(adapted.attempt_count, 4);
            assert_eq!(adapted.transcription_status, Some(expected_json));
        }
    }
}