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

kcode_audio_session_ingress/
lib.rs
