kcode_audio_session_ingress/
lib.rs

1#![forbid(unsafe_code)]
2
3use chrono::{DateTime, Utc};
4use kcode_audio_history_handoff::{Error as HandoffError, Handoff};
5pub use kcode_audio_ingress::SpeakerReviewAudio;
6use kcode_audio_ingress::{
7    AudioIngress, AudioInput, ChunkConfirmation, CorrectionPacket, ErrorKind as AudioErrorKind,
8    RecordingState, RecordingStatus,
9};
10pub use kcode_audio_session_view::{IngressPiece, Recording, SpeakerReview};
11use kcode_session_history::{SessionHistory, SessionRecord};
12use serde_json::Value;
13use uuid::Uuid;
14
15mod legacy_review;
16
17/// Application policy needed to coordinate audio with Session History.
18#[derive(Clone, Debug)]
19pub struct Config {
20    /// Stable application user identifier attributed to accepted recordings.
21    pub user_id: String,
22    /// Effective context window available to an audio-ingress session.
23    pub effective_context_tokens: u64,
24}
25
26/// Admitted recording bytes and source metadata.
27#[derive(Clone, Debug)]
28pub struct RecordingInput {
29    /// Complete WAV bytes already accepted by the transport.
30    pub bytes: Vec<u8>,
31    /// Instant at which the original recording began.
32    pub recorded_at: DateTime<Utc>,
33    /// Original leaf filename, when known.
34    pub original_filename: Option<String>,
35}
36
37/// Result of durably submitting one recording.
38#[derive(Clone, Debug)]
39pub struct RecordingSubmission {
40    /// Current combined state after submission.
41    pub recording: Recording,
42    /// Whether AudioIngress already knew the same bytes.
43    pub deduplicated: bool,
44}
45
46/// Detailed state for one recording.
47#[derive(Clone, Debug)]
48pub struct RecordingHistory {
49    pub recording: Recording,
50    pub final_transcript: Option<String>,
51    pub correction_packet: Option<CorrectionPacket>,
52    pub pieces: Vec<IngressPiece>,
53}
54
55/// Request to retry one transcript piece's memory ingress.
56#[derive(Clone, Debug)]
57pub struct RetryIngress {
58    pub piece_id: String,
59    pub expected_version: i64,
60    /// Optional state retained for 0.2 wire compatibility. When supplied, it
61    /// must exactly equal the current retained Session History state.
62    pub state: Option<Value>,
63}
64
65/// Stable coordinator error category for transport mapping.
66#[derive(Clone, Copy, Debug, Eq, PartialEq)]
67pub enum ErrorKind {
68    InvalidInput,
69    NotFound,
70    Conflict,
71    Internal,
72}
73
74/// Error returned by the audio/session-ingress coordinator.
75#[derive(Debug)]
76pub struct Error {
77    kind: ErrorKind,
78    message: String,
79}
80
81impl Error {
82    fn new(kind: ErrorKind, message: impl Into<String>) -> Self {
83        Self {
84            kind,
85            message: message.into(),
86        }
87    }
88
89    fn invalid(message: impl Into<String>) -> Self {
90        Self::new(ErrorKind::InvalidInput, message)
91    }
92
93    fn conflict(message: impl Into<String>) -> Self {
94        Self::new(ErrorKind::Conflict, message)
95    }
96
97    fn not_found() -> Self {
98        Self::new(
99            ErrorKind::NotFound,
100            "Audio recording or transcript piece not found.",
101        )
102    }
103
104    fn internal(error: impl std::fmt::Display) -> Self {
105        tracing::warn!(%error, "Audio session ingress operation failed");
106        Self::new(
107            ErrorKind::Internal,
108            "An unexpected audio session ingress error occurred.",
109        )
110    }
111
112    pub fn kind(&self) -> ErrorKind {
113        self.kind
114    }
115
116    pub fn message(&self) -> &str {
117        &self.message
118    }
119}
120
121impl std::fmt::Display for Error {
122    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
123        formatter.write_str(&self.message)
124    }
125}
126
127impl std::error::Error for Error {}
128
129/// Cloneable typed coordinator over AudioIngress and Session History.
130#[derive(Clone)]
131pub struct Coordinator {
132    audio: AudioIngress,
133    handoff: Handoff,
134    user_id: String,
135}
136
137impl Coordinator {
138    pub fn new(
139        audio: AudioIngress,
140        history: SessionHistory,
141        config: Config,
142    ) -> Result<Self, Error> {
143        let handoff = Handoff::new(
144            history,
145            config.user_id.clone(),
146            config.effective_context_tokens,
147        )
148        .map_err(handoff_error)?;
149        Ok(Self {
150            audio,
151            handoff,
152            user_id: config.user_id,
153        })
154    }
155
156    pub fn health(&self) -> Result<(), Error> {
157        self.audio.status().map_err(audio_error)?;
158        Ok(())
159    }
160
161    pub async fn submit(&self, input: RecordingInput) -> Result<RecordingSubmission, Error> {
162        let submission = self
163            .audio
164            .submit(AudioInput {
165                user_id: self.user_id.clone(),
166                bytes: input.bytes,
167                recorded_at: input.recorded_at,
168                original_filename: input.original_filename,
169            })
170            .await
171            .map_err(audio_error)?;
172        let (recordings, projections) = self.prepared_recordings().await?;
173        let (recording_status, projection) = recordings
174            .into_iter()
175            .zip(projections)
176            .find(|(recording, _)| recording.id == submission.recording_id)
177            .ok_or_else(Error::not_found)?;
178        validate_submission_owner(&recording_status, &self.user_id)?;
179        let view = kcode_audio_session_view::render(recording_status, projection);
180        Ok(RecordingSubmission {
181            recording: view.recording,
182            deduplicated: submission.deduplicated,
183        })
184    }
185
186    pub async fn recordings(&self) -> Result<Vec<Recording>, Error> {
187        let (recordings, projections) = self.prepared_recordings().await?;
188        Ok(recordings
189            .into_iter()
190            .zip(projections)
191            .map(|(recording, projection)| {
192                kcode_audio_session_view::render(recording, projection).recording
193            })
194            .collect())
195    }
196
197    pub async fn recording_by_sha256(&self, sha256: &str) -> Result<Recording, Error> {
198        if sha256.len() != 64 || !sha256.bytes().all(|byte| byte.is_ascii_hexdigit()) {
199            return Err(Error::invalid(
200                "audio SHA-256 must contain exactly 64 hexadecimal characters",
201            ));
202        }
203        let normalized = sha256.to_ascii_lowercase();
204        self.recordings()
205            .await?
206            .into_iter()
207            .find(|recording| recording.sha256 == normalized)
208            .ok_or_else(Error::not_found)
209    }
210
211    pub async fn recording_history(&self, recording_id: Uuid) -> Result<RecordingHistory, Error> {
212        let (recordings, projections) = self.prepared_recordings().await?;
213        let (recording_status, projection) = recordings
214            .into_iter()
215            .zip(projections)
216            .find(|(recording, _)| {
217                recording.id == recording_id && recording_belongs_to(recording, &self.user_id)
218            })
219            .ok_or_else(Error::not_found)?;
220        let final_transcript = match &recording_status.state {
221            RecordingState::Complete { transcript } => Some(transcript.clone()),
222            _ => None,
223        };
224        let correction_packet = recording_status.correction_packet.clone();
225        let view = kcode_audio_session_view::render(recording_status, projection);
226        Ok(RecordingHistory {
227            recording: view.recording,
228            final_transcript,
229            correction_packet,
230            pieces: view.pieces,
231        })
232    }
233
234    /// Returns one authorized correction-packet chunk as an exact WAV interval.
235    pub async fn speaker_review_audio(
236        &self,
237        recording_id: Uuid,
238        chunk_index: usize,
239    ) -> Result<SpeakerReviewAudio, Error> {
240        let owned = self
241            .audio
242            .status()
243            .map_err(audio_error)?
244            .recordings
245            .into_iter()
246            .any(|recording| {
247                recording.id == recording_id && recording_belongs_to(&recording, &self.user_id)
248            });
249        if !owned {
250            return Err(Error::not_found());
251        }
252        let audio = self.audio.clone();
253        tokio::task::spawn_blocking(move || audio.speaker_review_audio(recording_id, chunk_index))
254            .await
255            .map_err(Error::internal)?
256            .map_err(audio_error)
257    }
258
259    /// Returns the known-speaker dropdown choices for this coordinator.
260    pub fn known_speakers(&self) -> Result<Vec<String>, Error> {
261        self.audio.known_speakers().map_err(audio_error)
262    }
263
264    pub fn retry_recording(&self, recording_id: Uuid) -> Result<(), Error> {
265        let owned = self
266            .audio
267            .status()
268            .map_err(audio_error)?
269            .recordings
270            .into_iter()
271            .any(|recording| {
272                recording.id == recording_id && recording_belongs_to(&recording, &self.user_id)
273            });
274        if !owned {
275            return Err(Error::not_found());
276        }
277        self.audio.retry(recording_id).map_err(audio_error)
278    }
279
280    pub async fn confirm_speakers(
281        &self,
282        confirmation: ChunkConfirmation,
283    ) -> Result<CorrectionPacket, Error> {
284        let recording_id = confirmation.recording_id;
285        let (recordings, _) = self.prepared_recordings().await?;
286        let recording = recordings
287            .into_iter()
288            .find(|recording| {
289                recording.id == recording_id && recording_belongs_to(recording, &self.user_id)
290            })
291            .ok_or_else(Error::not_found)?;
292
293        if let Some(packet) = recording
294            .correction_packet
295            .as_ref()
296            .filter(|packet| legacy_review::confirmation_matches(packet, &confirmation))
297        {
298            self.handoff
299                .synchronize(std::slice::from_ref(&recording))
300                .await
301                .map_err(handoff_error)?;
302            return Ok(packet.clone());
303        }
304        if self
305            .handoff
306            .has_ingress(recording_id)
307            .await
308            .map_err(handoff_error)?
309        {
310            return Err(Error::conflict(
311                "speaker labels are already bound to accepted transcript ingress",
312            ));
313        }
314
315        let packet = self
316            .audio
317            .confirm_speakers(confirmation)
318            .map_err(audio_error)?;
319        let recording = self
320            .audio
321            .status()
322            .map_err(audio_error)?
323            .recordings
324            .into_iter()
325            .find(|recording| {
326                recording.id == recording_id && recording_belongs_to(recording, &self.user_id)
327            })
328            .ok_or_else(Error::not_found)?;
329        self.handoff
330            .synchronize(std::slice::from_ref(&recording))
331            .await
332            .map_err(handoff_error)?;
333        Ok(packet)
334    }
335
336    pub async fn retry_ingress(&self, input: RetryIngress) -> Result<SessionRecord, Error> {
337        let recordings = self.audio.status().map_err(audio_error)?.recordings;
338        self.handoff
339            .retry(
340                &recordings,
341                &input.piece_id,
342                input.expected_version,
343                input.state,
344            )
345            .await
346            .map_err(handoff_error)
347    }
348
349    pub async fn synchronize_completed_transcripts(&self) -> Result<(), Error> {
350        let mut recordings = self.owned_recordings()?;
351        if legacy_review::has_candidates(&recordings) {
352            let projections = self.project_recordings(&recordings).await?;
353            if self
354                .resolve_legacy_reviews(&recordings, &projections)
355                .await?
356            {
357                recordings = self.owned_recordings()?;
358            }
359        }
360        self.handoff
361            .synchronize(&recordings)
362            .await
363            .map_err(handoff_error)
364    }
365}
366
367fn recording_belongs_to(recording: &RecordingStatus, user_id: &str) -> bool {
368    recording.user_id == user_id
369}
370
371fn validate_submission_owner(recording: &RecordingStatus, user_id: &str) -> Result<(), Error> {
372    if recording_belongs_to(recording, user_id) {
373        Ok(())
374    } else {
375        Err(Error::conflict(
376            "identical audio is already attributed to another user",
377        ))
378    }
379}
380
381fn audio_error(error: kcode_audio_ingress::Error) -> Error {
382    match error.kind() {
383        AudioErrorKind::InvalidInput => Error::new(ErrorKind::InvalidInput, error.to_string()),
384        AudioErrorKind::NotFound => Error::new(ErrorKind::NotFound, error.to_string()),
385        AudioErrorKind::Conflict => Error::new(ErrorKind::Conflict, error.to_string()),
386        AudioErrorKind::Internal => Error::internal(error),
387    }
388}
389
390fn handoff_error(error: HandoffError) -> Error {
391    match error {
392        HandoffError::InvalidInput(message) => Error::new(ErrorKind::InvalidInput, message),
393        HandoffError::NotFound(message) => Error::new(ErrorKind::NotFound, message),
394        HandoffError::Conflict(message) => Error::new(ErrorKind::Conflict, message),
395        HandoffError::Internal(message) => Error::new(ErrorKind::Internal, message),
396    }
397}
398
399#[cfg(test)]
400mod tests {
401    use super::*;
402    use kcode_audio_ingress::{
403        ChunkConfirmation, ConfirmationState, ObservationConfirmation, SpeakerResolution,
404    };
405
406    fn completed_recording_for_user(
407        user_id: impl Into<String>,
408        transcript: impl Into<String>,
409    ) -> RecordingStatus {
410        let now = Utc::now();
411        RecordingStatus {
412            id: Uuid::new_v4(),
413            user_id: user_id.into(),
414            sha256: "0".repeat(64),
415            original_filename: "meeting.final.WAV".into(),
416            size_bytes: 42,
417            recorded_at: now,
418            received_at: now,
419            transcription_model: "transcription-model".into(),
420            reconciliation_model: "reconciliation-model".into(),
421            reconciliation_reasoning: "xhigh".into(),
422            state: RecordingState::Complete {
423                transcript: transcript.into(),
424            },
425            correction_packet: None,
426        }
427    }
428
429    fn with_speaker_packet(
430        mut recording: RecordingStatus,
431        confirmed_name: Option<&str>,
432    ) -> RecordingStatus {
433        recording.correction_packet = Some(CorrectionPacket {
434            recording_id: recording.id,
435            user_id: recording.user_id.clone(),
436            sha256: recording.sha256.clone(),
437            original_filename: recording.original_filename.clone(),
438            size_bytes: recording.size_bytes,
439            recorded_at: recording.recorded_at,
440            chunk_count: 1,
441            chunks: vec![kcode_audio_ingress::CorrectionChunk {
442                chunk_index: 0,
443                chunk_count: 1,
444                audio_start_ms: 0,
445                audio_end_ms: 1_000,
446                raw_gemini_response: "raw".into(),
447                parsed: kcode_audio_ingress::ParsedChunk {
448                    clip_valid: true,
449                    clip_validity_reason: None,
450                    speakers: Vec::new(),
451                },
452                observations: vec![kcode_audio_ingress::CorrectionObservation {
453                    local_label: "Speaker A".into(),
454                    speaker_ordinal: 0,
455                    observation_key: kcode_audio_ingress::ObservationKey {
456                        object_id: format!(
457                            "kcode-audio-ingress/recording/{}/chunk/0",
458                            recording.id
459                        ),
460                        piece_index: 0,
461                    },
462                    candidate: None,
463                    resolution: confirmed_name.map(|name| SpeakerResolution::Known {
464                        full_name: name.into(),
465                    }),
466                }],
467                signed_off: true,
468            }],
469            confirmation_state: ConfirmationState::Confirmed,
470        });
471        recording
472    }
473
474    #[test]
475    fn recording_filter_is_exactly_scoped_to_the_configured_user() {
476        let own = completed_recording_for_user("own-user", "Own transcript");
477        let foreign = completed_recording_for_user("foreign-user", "Foreign transcript");
478        let visible = [own.clone(), foreign]
479            .into_iter()
480            .filter(|recording| recording_belongs_to(recording, "own-user"))
481            .collect::<Vec<_>>();
482        assert_eq!(visible.len(), 1);
483        assert_eq!(visible[0].id, own.id);
484    }
485
486    #[test]
487    fn cross_user_sha_deduplication_fails_closed() {
488        let mut own = completed_recording_for_user("own-user", "Own transcript");
489        own.sha256 = "a".repeat(64);
490        let mut foreign = completed_recording_for_user("foreign-user", "Foreign transcript");
491        foreign.sha256 = own.sha256.clone();
492
493        assert!(validate_submission_owner(&own, "own-user").is_ok());
494        let error = validate_submission_owner(&foreign, "own-user").unwrap_err();
495        assert_eq!(error.kind(), ErrorKind::Conflict);
496        assert_eq!(
497            error.message(),
498            "identical audio is already attributed to another user"
499        );
500    }
501
502    #[test]
503    fn confirmed_label_retries_must_match_the_complete_mapping() {
504        let recording = with_speaker_packet(
505            completed_recording_for_user("user", "Transcript"),
506            Some("Human Choice"),
507        );
508        let packet = recording.correction_packet.as_ref().unwrap();
509        let key = packet.chunks[0].observations[0].observation_key.clone();
510        let matching = ChunkConfirmation {
511            recording_id: recording.id,
512            chunk_index: 0,
513            observations: vec![ObservationConfirmation {
514                observation_key: key.clone(),
515                resolution: SpeakerResolution::Known {
516                    full_name: "Human Choice".into(),
517                },
518            }],
519        };
520        let conflicting = ChunkConfirmation {
521            recording_id: recording.id,
522            chunk_index: 0,
523            observations: vec![ObservationConfirmation {
524                observation_key: key,
525                resolution: SpeakerResolution::Known {
526                    full_name: "Different Choice".into(),
527                },
528            }],
529        };
530        let incomplete = ChunkConfirmation {
531            recording_id: recording.id,
532            chunk_index: 0,
533            observations: Vec::new(),
534        };
535
536        assert!(legacy_review::confirmation_matches(packet, &matching));
537        assert!(!legacy_review::confirmation_matches(packet, &conflicting));
538        assert!(!legacy_review::confirmation_matches(packet, &incomplete));
539    }
540
541    #[test]
542    fn handoff_errors_preserve_category_and_message() {
543        let cases = [
544            (
545                HandoffError::InvalidInput("invalid".into()),
546                ErrorKind::InvalidInput,
547                "invalid",
548            ),
549            (
550                HandoffError::NotFound("missing".into()),
551                ErrorKind::NotFound,
552                "missing",
553            ),
554            (
555                HandoffError::Conflict("conflict".into()),
556                ErrorKind::Conflict,
557                "conflict",
558            ),
559            (
560                HandoffError::Internal("raw storage message".into()),
561                ErrorKind::Internal,
562                "raw storage message",
563            ),
564        ];
565
566        for (source, kind, message) in cases {
567            let error = handoff_error(source);
568            assert_eq!(error.kind(), kind);
569            assert_eq!(error.message(), message);
570        }
571    }
572}
kcode_audio_session_ingress/lib.rs

kcode_audio_session_ingress/
lib.rs
