kcode_audio_session_ingress/
lib.rs

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

kcode_audio_session_ingress/
lib.rs
