1use std::io::Read;
2use std::time::Instant;
3
4use serde::{Deserialize, Serialize};
5use thiserror::Error;
6
7use crate::workflow::CancellationHandle;
8
9const DEFAULT_STABILITY_TOLERANCE_SECONDS: f64 = 0.4;
10pub const LIVE_PCM_SAMPLE_RATE: u32 = 16_000;
11
12#[derive(Debug, Clone, Copy, PartialEq)]
13pub struct LiveWindowingConfig {
14 pub window_seconds: f64,
15 pub hop_seconds: f64,
16 pub finalize_lag_seconds: f64,
17 pub max_buffer_lag_seconds: f64,
18 pub stability_tolerance_seconds: f64,
19}
20
21impl Default for LiveWindowingConfig {
22 fn default() -> Self {
23 Self {
24 window_seconds: 5.0,
25 hop_seconds: 2.5,
26 finalize_lag_seconds: 5.0,
27 max_buffer_lag_seconds: 30.0,
28 stability_tolerance_seconds: DEFAULT_STABILITY_TOLERANCE_SECONDS,
29 }
30 }
31}
32
33#[derive(Debug, Error, Clone, PartialEq)]
34pub enum LiveWindowingError {
35 #[error("{field} must be finite and greater than zero")]
36 InvalidPositiveSeconds { field: &'static str },
37}
38
39#[derive(Debug, Clone, PartialEq)]
40pub struct LiveAsrSegmentCandidate {
41 pub start_seconds: f64,
42 pub end_seconds: f64,
43 pub start_at_utc: String,
44 pub end_at_utc: String,
45 pub text: String,
46 pub language: Option<String>,
47}
48
49#[derive(Debug, Clone, PartialEq)]
50pub struct LiveWindowTranscriptObservation {
51 pub session_id: String,
52 pub window_start_seconds: f64,
53 pub window_end_seconds: f64,
54 pub window_start_at_utc: String,
55 pub window_end_at_utc: String,
56 pub latest_ingested_audio_seconds: f64,
57 pub segments: Vec<LiveAsrSegmentCandidate>,
58}
59
60#[derive(Debug, Clone, Copy, PartialEq)]
61pub struct LiveWindow {
62 pub start_seconds: f64,
63 pub end_seconds: f64,
64}
65
66#[non_exhaustive]
72#[derive(Debug, Clone, PartialEq)]
73pub enum LiveTranscriptionProgressEvent {
74 SessionStart {
75 session_id: String,
76 },
77 WindowStart {
78 session_id: String,
79 window_index: usize,
80 start_seconds: f64,
81 end_seconds: f64,
82 },
83 ModelResolutionStart {
84 session_id: String,
85 window_index: usize,
86 provider: String,
87 model_id: String,
88 },
89 ModelResolutionEnd {
90 session_id: String,
91 window_index: usize,
92 provider: String,
93 model_id: String,
94 source: String,
95 },
96 ModelDownloadStart {
97 session_id: String,
98 window_index: usize,
99 provider: String,
100 model_id: String,
101 },
102 ModelDownloadEnd {
103 session_id: String,
104 window_index: usize,
105 provider: String,
106 model_id: String,
107 duration_seconds: f64,
108 },
109 ModelLoadStart {
110 session_id: String,
111 window_index: usize,
112 provider: String,
113 model_id: String,
114 },
115 ModelLoadEnd {
116 session_id: String,
117 window_index: usize,
118 provider: String,
119 model_id: String,
120 duration_seconds: f64,
121 },
122 ModelReuse {
123 session_id: String,
124 window_index: usize,
125 provider: String,
126 model_id: String,
127 },
128 WindowEnd {
129 session_id: String,
130 window_index: usize,
131 start_seconds: f64,
132 end_seconds: f64,
133 duration_seconds: f64,
134 },
135 Completed {
136 session_id: String,
137 processed_audio_seconds: f64,
138 window_count: usize,
139 final_segment_count: u64,
140 duration_seconds: f64,
141 },
142 Failure {
143 session_id: String,
144 window_index: Option<usize>,
145 message: String,
146 duration_seconds: f64,
147 },
148 Cancelled {
149 session_id: String,
150 next_window_index: usize,
151 processed_audio_seconds: f64,
152 final_segment_count: u64,
153 duration_seconds: f64,
154 },
155}
156
157pub trait LiveTranscriptionProgressObserver {
158 fn observe(&mut self, event: LiveTranscriptionProgressEvent);
159}
160
161#[derive(Debug, Default)]
162pub struct NoopLiveTranscriptionProgressObserver;
163
164impl LiveTranscriptionProgressObserver for NoopLiveTranscriptionProgressObserver {
165 fn observe(&mut self, _event: LiveTranscriptionProgressEvent) {}
166}
167
168#[derive(Debug, Error, Clone, PartialEq, Eq)]
169#[error("{message}")]
170pub struct LiveWindowProcessingError {
171 message: String,
172}
173
174impl LiveWindowProcessingError {
175 pub fn new(message: impl Into<String>) -> Self {
176 Self {
177 message: message.into(),
178 }
179 }
180}
181
182impl From<LiveWindowingError> for LiveWindowProcessingError {
183 fn from(error: LiveWindowingError) -> Self {
184 Self::new(error.to_string())
185 }
186}
187
188#[derive(Debug, Clone, PartialEq)]
189pub struct LivePcmWindow {
190 pub window_index: usize,
191 pub start_seconds: f64,
192 pub end_seconds: f64,
193 pub latest_ingested_audio_seconds: f64,
194 pub samples: Vec<f32>,
195}
196
197pub trait LivePcmWindowProcessor {
198 fn process_window(
199 &mut self,
200 window: LivePcmWindow,
201 progress: &mut dyn LiveTranscriptionProgressObserver,
202 ) -> Result<Vec<LiveTranscriptEvent>, LiveWindowProcessingError>;
203}
204
205#[derive(Debug, Clone, PartialEq)]
206pub struct LivePcmIngestionReport {
207 pub processed_audio_seconds: f64,
208 pub processed_sample_count: usize,
209 pub window_count: usize,
210 pub events: Vec<LiveTranscriptEvent>,
211}
212
213impl LivePcmIngestionReport {
214 pub fn failed(&self) -> bool {
215 self.events.iter().any(|event| {
216 matches!(
217 event,
218 LiveTranscriptEvent::Error(_)
219 | LiveTranscriptEvent::SessionEnded(LiveSessionEnded {
220 reason: LiveSessionEndReason::Error,
221 ..
222 })
223 )
224 })
225 }
226}
227
228#[derive(Debug, Clone)]
229pub struct LivePcmIngestionSession {
230 session_id: String,
231 config: LiveWindowingConfig,
232 samples: Vec<f32>,
233 next_window_start_seconds: f64,
234 next_sequence: u64,
235 window_count: usize,
236 final_segment_count: u64,
237 failed: bool,
238 active_window_index: Option<usize>,
239}
240
241impl LivePcmIngestionSession {
242 pub fn new(
243 session_id: impl Into<String>,
244 config: LiveWindowingConfig,
245 ) -> Result<Self, LiveWindowingError> {
246 LiveWindowPlanner::new(config)?;
247 Ok(Self {
248 session_id: session_id.into(),
249 config,
250 samples: Vec::new(),
251 next_window_start_seconds: 0.0,
252 next_sequence: 1,
253 window_count: 0,
254 final_segment_count: 0,
255 failed: false,
256 active_window_index: None,
257 })
258 }
259
260 pub fn ingest_reader(
261 &mut self,
262 reader: &mut dyn Read,
263 processor: &mut dyn LivePcmWindowProcessor,
264 ) -> LivePcmIngestionReport {
265 let cancellation = CancellationHandle::new();
266 let mut progress = NoopLiveTranscriptionProgressObserver;
267 self.ingest_reader_with_control(
268 reader,
269 processor,
270 &mut |_| Ok(()),
271 &mut progress,
272 &cancellation,
273 )
274 }
275
276 pub fn ingest_reader_with_event_sink(
277 &mut self,
278 reader: &mut dyn Read,
279 processor: &mut dyn LivePcmWindowProcessor,
280 sink: &mut dyn FnMut(&LiveTranscriptEvent) -> Result<(), LiveWindowProcessingError>,
281 ) -> LivePcmIngestionReport {
282 let cancellation = CancellationHandle::new();
283 let mut progress = NoopLiveTranscriptionProgressObserver;
284 self.ingest_reader_with_control(reader, processor, sink, &mut progress, &cancellation)
285 }
286
287 pub fn ingest_reader_with_observer(
288 &mut self,
289 reader: &mut dyn Read,
290 processor: &mut dyn LivePcmWindowProcessor,
291 sink: &mut dyn FnMut(&LiveTranscriptEvent) -> Result<(), LiveWindowProcessingError>,
292 progress: &mut dyn LiveTranscriptionProgressObserver,
293 ) -> LivePcmIngestionReport {
294 let cancellation = CancellationHandle::new();
295 self.ingest_reader_with_control(reader, processor, sink, progress, &cancellation)
296 }
297
298 pub fn ingest_reader_with_control(
299 &mut self,
300 reader: &mut dyn Read,
301 processor: &mut dyn LivePcmWindowProcessor,
302 sink: &mut dyn FnMut(&LiveTranscriptEvent) -> Result<(), LiveWindowProcessingError>,
303 progress: &mut dyn LiveTranscriptionProgressObserver,
304 cancellation: &CancellationHandle,
305 ) -> LivePcmIngestionReport {
306 let started = Instant::now();
307 progress.observe(LiveTranscriptionProgressEvent::SessionStart {
308 session_id: self.session_id.clone(),
309 });
310
311 let mut events = Vec::new();
312 if cancellation.is_cancelled() {
313 self.finish_cancelled(&mut events, sink, progress, started);
314 return self.report(events);
315 }
316
317 let mut pending_bytes = Vec::new();
318 let mut read_buffer = [0_u8; 8192];
319 loop {
320 if cancellation.is_cancelled() {
321 break;
322 }
323 match reader.read(&mut read_buffer) {
324 Ok(0) => {
325 if !pending_bytes.is_empty() {
326 self.emit_error(
327 &mut events,
328 sink,
329 progress,
330 started,
331 format!(
332 "truncated f32le PCM frame: {} trailing byte(s)",
333 pending_bytes.len()
334 ),
335 );
336 }
337 break;
338 }
339 Ok(read_len) => {
340 let mut bytes = Vec::with_capacity(pending_bytes.len() + read_len);
341 bytes.extend_from_slice(&pending_bytes);
342 bytes.extend_from_slice(&read_buffer[..read_len]);
343 let complete_len = bytes.len() - (bytes.len() % 4);
344 pending_bytes.clear();
345 pending_bytes.extend_from_slice(&bytes[complete_len..]);
346 let ingest_result =
347 self.ingest_pcm_bytes(&bytes[..complete_len])
348 .and_then(|()| {
349 self.process_ready_windows(
350 processor,
351 &mut events,
352 sink,
353 progress,
354 cancellation,
355 started,
356 )
357 });
358 if let Err(message) = ingest_result {
359 self.emit_error(&mut events, sink, progress, started, message);
360 break;
361 }
362 }
363 Err(error) => {
364 self.emit_error(
365 &mut events,
366 sink,
367 progress,
368 started,
369 format!("live PCM reader failed: {error}"),
370 );
371 break;
372 }
373 }
374 }
375
376 if !self.failed && !cancellation.is_cancelled() {
377 if let Err(message) = self.process_ready_windows(
378 processor,
379 &mut events,
380 sink,
381 progress,
382 cancellation,
383 started,
384 ) {
385 self.emit_error(&mut events, sink, progress, started, message);
386 }
387 }
388
389 if cancellation.is_cancelled() && !self.failed {
390 self.finish_cancelled(&mut events, sink, progress, started);
391 } else {
392 self.finish(&mut events, sink, progress, started);
393 }
394 self.report(events)
395 }
396
397 fn ingest_pcm_bytes(&mut self, bytes: &[u8]) -> Result<(), String> {
398 for sample_bytes in bytes.chunks_exact(4) {
399 let sample = f32::from_le_bytes([
400 sample_bytes[0],
401 sample_bytes[1],
402 sample_bytes[2],
403 sample_bytes[3],
404 ]);
405 if !sample.is_finite() {
406 return Err(format!(
407 "non-finite f32le PCM sample at feed sample {}",
408 self.samples.len()
409 ));
410 }
411 self.samples.push(sample);
412 }
413 Ok(())
414 }
415
416 fn process_ready_windows(
417 &mut self,
418 processor: &mut dyn LivePcmWindowProcessor,
419 events: &mut Vec<LiveTranscriptEvent>,
420 sink: &mut dyn FnMut(&LiveTranscriptEvent) -> Result<(), LiveWindowProcessingError>,
421 progress: &mut dyn LiveTranscriptionProgressObserver,
422 cancellation: &CancellationHandle,
423 started: Instant,
424 ) -> Result<(), String> {
425 while self.next_window_start_seconds + self.config.window_seconds
426 <= self.processed_audio_seconds()
427 {
428 if cancellation.is_cancelled() {
429 break;
430 }
431 let window_index = self.window_count;
432 let start_seconds = self.next_window_start_seconds;
433 let end_seconds = start_seconds + self.config.window_seconds;
434 let window_started = Instant::now();
435 progress.observe(LiveTranscriptionProgressEvent::WindowStart {
436 session_id: self.session_id.clone(),
437 window_index,
438 start_seconds,
439 end_seconds,
440 });
441 self.active_window_index = Some(window_index);
442 let start_sample = seconds_to_sample_index(start_seconds);
443 let end_sample = seconds_to_sample_index(end_seconds);
444 let window_events = processor
445 .process_window(
446 LivePcmWindow {
447 window_index,
448 start_seconds,
449 end_seconds,
450 latest_ingested_audio_seconds: self.processed_audio_seconds(),
451 samples: self.samples[start_sample..end_sample].to_vec(),
452 },
453 progress,
454 )
455 .map_err(|error| format!("live PCM window processing failed: {error}"))?;
456 let cancelled_during_window = cancellation.is_cancelled();
457 self.final_segment_count += window_events
458 .iter()
459 .filter(|event| matches!(event, LiveTranscriptEvent::Final(_)))
460 .count() as u64;
461 for mut event in window_events {
462 if cancelled_during_window && matches!(event, LiveTranscriptEvent::Partial(_)) {
463 continue;
464 }
465 let error_message = match &event {
466 LiveTranscriptEvent::Error(error) => Some(error.message.clone()),
467 _ => None,
468 };
469 set_live_event_sequence(&mut event, self.next_sequence());
470 if let Some(message) = error_message {
471 self.failed = true;
472 progress.observe(LiveTranscriptionProgressEvent::Failure {
473 session_id: self.session_id.clone(),
474 window_index: self.active_window_index,
475 message,
476 duration_seconds: started.elapsed().as_secs_f64(),
477 });
478 }
479 self.push_event(events, sink, event);
480 if self.failed {
481 break;
482 }
483 }
484 if self.failed {
485 break;
486 }
487 self.window_count += 1;
488 self.next_window_start_seconds += self.config.hop_seconds;
489 progress.observe(LiveTranscriptionProgressEvent::WindowEnd {
490 session_id: self.session_id.clone(),
491 window_index,
492 start_seconds,
493 end_seconds,
494 duration_seconds: window_started.elapsed().as_secs_f64(),
495 });
496 self.active_window_index = None;
497 if self.failed || cancelled_during_window {
498 break;
499 }
500 }
501 Ok(())
502 }
503
504 fn emit_error(
505 &mut self,
506 events: &mut Vec<LiveTranscriptEvent>,
507 sink: &mut dyn FnMut(&LiveTranscriptEvent) -> Result<(), LiveWindowProcessingError>,
508 progress: &mut dyn LiveTranscriptionProgressObserver,
509 started: Instant,
510 message: String,
511 ) {
512 self.failed = true;
513 progress.observe(LiveTranscriptionProgressEvent::Failure {
514 session_id: self.session_id.clone(),
515 window_index: self.active_window_index,
516 message: message.clone(),
517 duration_seconds: started.elapsed().as_secs_f64(),
518 });
519 let sequence = self.next_sequence();
520 self.push_event(
521 events,
522 sink,
523 LiveTranscriptEvent::Error(LiveTranscriptError {
524 session_id: self.session_id.clone(),
525 sequence,
526 message,
527 recoverable: false,
528 }),
529 );
530 }
531
532 fn finish_cancelled(
533 &mut self,
534 events: &mut Vec<LiveTranscriptEvent>,
535 sink: &mut dyn FnMut(&LiveTranscriptEvent) -> Result<(), LiveWindowProcessingError>,
536 progress: &mut dyn LiveTranscriptionProgressObserver,
537 started: Instant,
538 ) {
539 progress.observe(LiveTranscriptionProgressEvent::Cancelled {
540 session_id: self.session_id.clone(),
541 next_window_index: self.window_count,
542 processed_audio_seconds: self.processed_audio_seconds(),
543 final_segment_count: self.final_segment_count,
544 duration_seconds: started.elapsed().as_secs_f64(),
545 });
546 self.push_session_end(events, sink, LiveSessionEndReason::Cancelled);
547 }
548
549 fn finish(
550 &mut self,
551 events: &mut Vec<LiveTranscriptEvent>,
552 sink: &mut dyn FnMut(&LiveTranscriptEvent) -> Result<(), LiveWindowProcessingError>,
553 progress: &mut dyn LiveTranscriptionProgressObserver,
554 started: Instant,
555 ) {
556 let reason = if self.failed {
557 LiveSessionEndReason::Error
558 } else {
559 progress.observe(LiveTranscriptionProgressEvent::Completed {
560 session_id: self.session_id.clone(),
561 processed_audio_seconds: self.processed_audio_seconds(),
562 window_count: self.window_count,
563 final_segment_count: self.final_segment_count,
564 duration_seconds: started.elapsed().as_secs_f64(),
565 });
566 LiveSessionEndReason::Completed
567 };
568 self.push_session_end(events, sink, reason);
569 }
570
571 fn push_session_end(
572 &mut self,
573 events: &mut Vec<LiveTranscriptEvent>,
574 sink: &mut dyn FnMut(&LiveTranscriptEvent) -> Result<(), LiveWindowProcessingError>,
575 reason: LiveSessionEndReason,
576 ) {
577 let sequence = self.next_sequence();
578 self.push_event(
579 events,
580 sink,
581 LiveTranscriptEvent::SessionEnded(LiveSessionEnded {
582 session_id: self.session_id.clone(),
583 sequence,
584 reason,
585 processed_audio_seconds: self.processed_audio_seconds(),
586 final_segment_count: self.final_segment_count,
587 }),
588 );
589 }
590
591 fn push_event(
592 &mut self,
593 events: &mut Vec<LiveTranscriptEvent>,
594 sink: &mut dyn FnMut(&LiveTranscriptEvent) -> Result<(), LiveWindowProcessingError>,
595 event: LiveTranscriptEvent,
596 ) {
597 let _ = sink(&event);
598 events.push(event);
599 }
600
601 fn report(&self, events: Vec<LiveTranscriptEvent>) -> LivePcmIngestionReport {
602 LivePcmIngestionReport {
603 processed_audio_seconds: self.processed_audio_seconds(),
604 processed_sample_count: self.samples.len(),
605 window_count: self.window_count,
606 events,
607 }
608 }
609
610 fn next_sequence(&mut self) -> u64 {
611 let sequence = self.next_sequence;
612 self.next_sequence += 1;
613 sequence
614 }
615
616 fn processed_audio_seconds(&self) -> f64 {
617 self.samples.len() as f64 / LIVE_PCM_SAMPLE_RATE as f64
618 }
619}
620
621fn seconds_to_sample_index(seconds: f64) -> usize {
622 (seconds * LIVE_PCM_SAMPLE_RATE as f64).round() as usize
623}
624
625fn set_live_event_sequence(event: &mut LiveTranscriptEvent, sequence: u64) {
626 match event {
627 LiveTranscriptEvent::SessionStarted(event) => event.sequence = sequence,
628 LiveTranscriptEvent::Partial(event) => event.sequence = sequence,
629 LiveTranscriptEvent::Final(event) => event.sequence = sequence,
630 LiveTranscriptEvent::Error(event) => event.sequence = sequence,
631 LiveTranscriptEvent::SessionEnded(event) => event.sequence = sequence,
632 }
633}
634
635#[derive(Debug, Clone)]
636pub struct LiveWindowPlanner {
637 config: LiveWindowingConfig,
638}
639
640impl LiveWindowPlanner {
641 pub fn new(config: LiveWindowingConfig) -> Result<Self, LiveWindowingError> {
642 validate_config(config)?;
643 Ok(Self { config })
644 }
645
646 pub fn ready_windows(&self, processed_audio_seconds: f64) -> Vec<LiveWindow> {
647 if !processed_audio_seconds.is_finite()
648 || processed_audio_seconds < self.config.window_seconds
649 {
650 return Vec::new();
651 }
652
653 let mut windows = Vec::new();
654 let mut start_seconds = 0.0;
655 while start_seconds + self.config.window_seconds <= processed_audio_seconds {
656 windows.push(LiveWindow {
657 start_seconds,
658 end_seconds: start_seconds + self.config.window_seconds,
659 });
660 start_seconds += self.config.hop_seconds;
661 }
662 windows
663 }
664}
665
666#[derive(Debug, Clone)]
667pub struct LiveWindowState {
668 config: LiveWindowingConfig,
669 next_sequence: u64,
670 final_segment_count: u64,
671 pending_segments: Vec<PendingLiveSegment>,
672 finalized_segments: Vec<FinalizedLiveSegmentKey>,
673 failed: bool,
674}
675
676impl LiveWindowState {
677 pub fn new(config: LiveWindowingConfig) -> Result<Self, LiveWindowingError> {
678 validate_config(config)?;
679
680 Ok(Self {
681 config,
682 next_sequence: 1,
683 final_segment_count: 0,
684 pending_segments: Vec::new(),
685 finalized_segments: Vec::new(),
686 failed: false,
687 })
688 }
689
690 pub fn observe_window(
691 &mut self,
692 observation: LiveWindowTranscriptObservation,
693 ) -> Result<Vec<LiveTranscriptEvent>, LiveWindowingError> {
694 let buffer_lag_seconds =
695 observation.latest_ingested_audio_seconds - observation.window_end_seconds;
696 if buffer_lag_seconds > self.config.max_buffer_lag_seconds {
697 self.failed = true;
698 return Ok(vec![LiveTranscriptEvent::Error(LiveTranscriptError {
699 session_id: observation.session_id,
700 sequence: self.next_sequence(),
701 message: format!(
702 "processing fell behind live input by {buffer_lag_seconds:.3} seconds"
703 ),
704 recoverable: false,
705 })]);
706 }
707
708 let stable_segments = self.mark_stable_segments(&observation);
709 let sequence = self.next_sequence();
710 let partial_segments = observation
711 .segments
712 .iter()
713 .map(|segment| LivePartialSegment {
714 start_seconds: segment.start_seconds,
715 end_seconds: segment.end_seconds,
716 text: segment.text.clone(),
717 language: segment.language.clone(),
718 })
719 .collect();
720 let mut events = vec![LiveTranscriptEvent::Partial(LivePartialTranscript {
721 session_id: observation.session_id.clone(),
722 sequence,
723 window_start_seconds: observation.window_start_seconds,
724 window_end_seconds: observation.window_end_seconds,
725 window_start_at_utc: observation.window_start_at_utc.clone(),
726 window_end_at_utc: observation.window_end_at_utc.clone(),
727 text: observation
728 .segments
729 .iter()
730 .map(|segment| segment.text.trim())
731 .filter(|text| !text.is_empty())
732 .collect::<Vec<_>>()
733 .join(" "),
734 segments: partial_segments,
735 })];
736
737 for stable_segment in stable_segments {
738 if stable_segment.end_seconds
739 <= observation.window_end_seconds - self.config.finalize_lag_seconds
740 {
741 events.push(LiveTranscriptEvent::Final(
742 self.finalize_segment(observation.session_id.clone(), stable_segment),
743 ));
744 }
745 }
746
747 self.pending_segments
748 .retain(|segment| !(segment.stable && segment.finalized));
749 self.add_pending_segments(observation);
750
751 Ok(events)
752 }
753
754 pub fn final_segment_count(&self) -> u64 {
755 self.final_segment_count
756 }
757
758 pub fn has_failed(&self) -> bool {
759 self.failed
760 }
761
762 fn next_sequence(&mut self) -> u64 {
763 let sequence = self.next_sequence;
764 self.next_sequence += 1;
765 sequence
766 }
767
768 fn mark_stable_segments(
769 &mut self,
770 observation: &LiveWindowTranscriptObservation,
771 ) -> Vec<PendingLiveSegment> {
772 let mut stable_segments = Vec::new();
773 let finalized_segments = &self.finalized_segments;
774 let stability_tolerance_seconds = self.config.stability_tolerance_seconds;
775
776 for pending in &mut self.pending_segments {
777 if pending.finalized
778 || segment_matches_finalized(
779 finalized_segments,
780 stability_tolerance_seconds,
781 pending.start_seconds,
782 pending.end_seconds,
783 &pending.normalized_text,
784 )
785 || !windows_overlap(
786 pending.window_start_seconds,
787 pending.window_end_seconds,
788 observation.window_start_seconds,
789 observation.window_end_seconds,
790 )
791 {
792 continue;
793 }
794
795 if observation.segments.iter().any(|candidate| {
796 normalized_text(&candidate.text) == pending.normalized_text
797 && seconds_within_tolerance(
798 candidate.start_seconds,
799 pending.start_seconds,
800 self.config.stability_tolerance_seconds,
801 )
802 && seconds_within_tolerance(
803 candidate.end_seconds,
804 pending.end_seconds,
805 self.config.stability_tolerance_seconds,
806 )
807 }) {
808 pending.stable = true;
809 stable_segments.push(pending.clone());
810 }
811 }
812
813 stable_segments
814 }
815
816 fn finalize_segment(
817 &mut self,
818 session_id: String,
819 mut stable_segment: PendingLiveSegment,
820 ) -> LiveFinalTranscriptSegment {
821 self.final_segment_count += 1;
822 stable_segment.finalized = true;
823 self.finalized_segments.push(FinalizedLiveSegmentKey {
824 start_seconds: stable_segment.start_seconds,
825 end_seconds: stable_segment.end_seconds,
826 normalized_text: stable_segment.normalized_text.clone(),
827 });
828
829 if let Some(pending) = self
830 .pending_segments
831 .iter_mut()
832 .find(|pending| pending.id == stable_segment.id)
833 {
834 pending.finalized = true;
835 }
836
837 LiveFinalTranscriptSegment {
838 session_id,
839 sequence: self.next_sequence(),
840 segment_id: format!("seg-{segment_id:06}", segment_id = self.final_segment_count),
841 start_seconds: stable_segment.start_seconds,
842 end_seconds: stable_segment.end_seconds,
843 start_at_utc: stable_segment.start_at_utc,
844 end_at_utc: stable_segment.end_at_utc,
845 text: stable_segment.normalized_text,
846 language: stable_segment.language,
847 }
848 }
849
850 fn add_pending_segments(&mut self, observation: LiveWindowTranscriptObservation) {
851 for segment in observation.segments {
852 let normalized_text = normalized_text(&segment.text);
853 if normalized_text.is_empty()
854 || self.segment_matches_finalized(
855 segment.start_seconds,
856 segment.end_seconds,
857 &normalized_text,
858 )
859 {
860 continue;
861 }
862
863 self.pending_segments.push(PendingLiveSegment {
864 id: format!(
865 "{:.3}:{:.3}:{}",
866 segment.start_seconds, segment.end_seconds, normalized_text
867 ),
868 window_start_seconds: observation.window_start_seconds,
869 window_end_seconds: observation.window_end_seconds,
870 start_seconds: segment.start_seconds,
871 end_seconds: segment.end_seconds,
872 start_at_utc: segment.start_at_utc,
873 end_at_utc: segment.end_at_utc,
874 normalized_text,
875 language: segment.language,
876 stable: false,
877 finalized: false,
878 });
879 }
880 }
881
882 fn segment_matches_finalized(
883 &self,
884 start_seconds: f64,
885 end_seconds: f64,
886 normalized_text: &str,
887 ) -> bool {
888 segment_matches_finalized(
889 &self.finalized_segments,
890 self.config.stability_tolerance_seconds,
891 start_seconds,
892 end_seconds,
893 normalized_text,
894 )
895 }
896}
897
898#[derive(Debug, Clone, PartialEq)]
899struct PendingLiveSegment {
900 id: String,
901 window_start_seconds: f64,
902 window_end_seconds: f64,
903 start_seconds: f64,
904 end_seconds: f64,
905 start_at_utc: String,
906 end_at_utc: String,
907 normalized_text: String,
908 language: Option<String>,
909 stable: bool,
910 finalized: bool,
911}
912
913#[derive(Debug, Clone, PartialEq)]
914struct FinalizedLiveSegmentKey {
915 start_seconds: f64,
916 end_seconds: f64,
917 normalized_text: String,
918}
919
920fn segment_matches_finalized(
921 finalized_segments: &[FinalizedLiveSegmentKey],
922 tolerance_seconds: f64,
923 start_seconds: f64,
924 end_seconds: f64,
925 normalized_text: &str,
926) -> bool {
927 finalized_segments.iter().any(|finalized| {
928 finalized.normalized_text == normalized_text
929 && seconds_within_tolerance(finalized.start_seconds, start_seconds, tolerance_seconds)
930 && seconds_within_tolerance(finalized.end_seconds, end_seconds, tolerance_seconds)
931 })
932}
933
934fn validate_config(config: LiveWindowingConfig) -> Result<(), LiveWindowingError> {
935 validate_positive_seconds("window_seconds", config.window_seconds)?;
936 validate_positive_seconds("hop_seconds", config.hop_seconds)?;
937 validate_positive_seconds("finalize_lag_seconds", config.finalize_lag_seconds)?;
938 validate_positive_seconds("max_buffer_lag_seconds", config.max_buffer_lag_seconds)?;
939 validate_positive_seconds(
940 "stability_tolerance_seconds",
941 config.stability_tolerance_seconds,
942 )
943}
944
945fn validate_positive_seconds(field: &'static str, value: f64) -> Result<(), LiveWindowingError> {
946 if value.is_finite() && value > 0.0 {
947 Ok(())
948 } else {
949 Err(LiveWindowingError::InvalidPositiveSeconds { field })
950 }
951}
952
953fn normalized_text(text: &str) -> String {
954 text.split_whitespace().collect::<Vec<_>>().join(" ")
955}
956
957fn seconds_within_tolerance(left: f64, right: f64, tolerance: f64) -> bool {
958 (left - right).abs() <= tolerance
959}
960
961fn windows_overlap(first_start: f64, first_end: f64, second_start: f64, second_end: f64) -> bool {
962 first_start < second_end && second_start < first_end
963}
964
965#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
966#[serde(rename_all = "camelCase", tag = "event")]
967pub enum LiveTranscriptEvent {
968 SessionStarted(LiveSessionStarted),
969 Partial(LivePartialTranscript),
970 Final(LiveFinalTranscriptSegment),
971 Error(LiveTranscriptError),
972 SessionEnded(LiveSessionEnded),
973}
974
975#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
976#[serde(rename_all = "camelCase")]
977pub struct LiveSessionStarted {
978 pub session_id: String,
979 pub sequence: u64,
980 pub source: String,
981 pub ingest_started_at_utc: String,
982 pub sample_rate: u32,
983 pub channels: u16,
984 pub model_id: String,
985 #[serde(skip_serializing_if = "Option::is_none")]
986 pub language: Option<String>,
987}
988
989#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
990#[serde(rename_all = "camelCase")]
991pub struct LivePartialTranscript {
992 pub session_id: String,
993 pub sequence: u64,
994 pub window_start_seconds: f64,
995 pub window_end_seconds: f64,
996 pub window_start_at_utc: String,
997 pub window_end_at_utc: String,
998 pub text: String,
999 pub segments: Vec<LivePartialSegment>,
1000}
1001
1002#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
1003#[serde(rename_all = "camelCase")]
1004pub struct LivePartialSegment {
1005 pub start_seconds: f64,
1006 pub end_seconds: f64,
1007 pub text: String,
1008 #[serde(skip_serializing_if = "Option::is_none")]
1009 pub language: Option<String>,
1010}
1011
1012#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
1013#[serde(rename_all = "camelCase")]
1014pub struct LiveFinalTranscriptSegment {
1015 pub session_id: String,
1016 pub sequence: u64,
1017 pub segment_id: String,
1018 pub start_seconds: f64,
1019 pub end_seconds: f64,
1020 pub start_at_utc: String,
1021 pub end_at_utc: String,
1022 pub text: String,
1023 #[serde(skip_serializing_if = "Option::is_none")]
1024 pub language: Option<String>,
1025}
1026
1027#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
1028#[serde(rename_all = "camelCase")]
1029pub struct LiveTranscriptError {
1030 pub session_id: String,
1031 pub sequence: u64,
1032 pub message: String,
1033 pub recoverable: bool,
1034}
1035
1036#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
1037#[serde(rename_all = "camelCase")]
1038pub struct LiveSessionEnded {
1039 pub session_id: String,
1040 pub sequence: u64,
1041 pub reason: LiveSessionEndReason,
1042 pub processed_audio_seconds: f64,
1043 pub final_segment_count: u64,
1044}
1045
1046#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
1047#[serde(rename_all = "camelCase")]
1048pub enum LiveSessionEndReason {
1049 Completed,
1050 Error,
1051 Cancelled,
1052}
1053
1054pub fn live_transcript_events_to_jsonl(
1055 events: &[LiveTranscriptEvent],
1056) -> Result<String, serde_json::Error> {
1057 let mut jsonl = String::new();
1058 for event in events {
1059 jsonl.push_str(&serde_json::to_string(event)?);
1060 jsonl.push('\n');
1061 }
1062 Ok(jsonl)
1063}
1064
1065#[cfg(test)]
1066mod tests {
1067 use super::*;
1068
1069 #[test]
1070 fn near_live_window_state_emits_partial_before_any_final() {
1071 let mut state = LiveWindowState::new(LiveWindowingConfig {
1072 window_seconds: 5.0,
1073 hop_seconds: 2.5,
1074 finalize_lag_seconds: 5.0,
1075 max_buffer_lag_seconds: 30.0,
1076 stability_tolerance_seconds: 0.4,
1077 })
1078 .expect("valid live config");
1079
1080 let events = state
1081 .observe_window(LiveWindowTranscriptObservation {
1082 session_id: "session-1".to_string(),
1083 window_start_seconds: 0.0,
1084 window_end_seconds: 5.0,
1085 window_start_at_utc: "2026-06-22T15:30:00Z".to_string(),
1086 window_end_at_utc: "2026-06-22T15:30:05Z".to_string(),
1087 latest_ingested_audio_seconds: 5.0,
1088 segments: vec![LiveAsrSegmentCandidate {
1089 start_seconds: 0.4,
1090 end_seconds: 1.8,
1091 start_at_utc: "2026-06-22T15:30:00.400Z".to_string(),
1092 end_at_utc: "2026-06-22T15:30:01.800Z".to_string(),
1093 text: "hello wor".to_string(),
1094 language: Some("en".to_string()),
1095 }],
1096 })
1097 .expect("window is accepted");
1098
1099 assert_eq!(events.len(), 1);
1100 assert!(matches!(events[0], LiveTranscriptEvent::Partial(_)));
1101 assert_eq!(state.final_segment_count(), 0);
1102 }
1103
1104 #[test]
1105 fn overlapping_matching_window_promotes_stable_segment_after_finalize_lag() {
1106 let mut state = LiveWindowState::new(LiveWindowingConfig::default()).expect("valid config");
1107
1108 state
1109 .observe_window(observation(
1110 "session-1",
1111 0.0,
1112 5.0,
1113 5.0,
1114 vec![candidate(0.4, 1.8, " hello world ")],
1115 ))
1116 .expect("first window accepted");
1117
1118 let events = state
1119 .observe_window(observation(
1120 "session-1",
1121 2.5,
1122 7.5,
1123 7.5,
1124 vec![candidate(0.45, 1.75, "hello world")],
1125 ))
1126 .expect("second window accepted");
1127
1128 assert_eq!(events.len(), 2);
1129 assert!(matches!(events[0], LiveTranscriptEvent::Partial(_)));
1130 assert_eq!(
1131 events[1],
1132 LiveTranscriptEvent::Final(LiveFinalTranscriptSegment {
1133 session_id: "session-1".to_string(),
1134 sequence: 3,
1135 segment_id: "seg-000001".to_string(),
1136 start_seconds: 0.4,
1137 end_seconds: 1.8,
1138 start_at_utc: "2026-06-22T15:30:00.400Z".to_string(),
1139 end_at_utc: "2026-06-22T15:30:01.800Z".to_string(),
1140 text: "hello world".to_string(),
1141 language: Some("en".to_string()),
1142 })
1143 );
1144 assert_eq!(state.final_segment_count(), 1);
1145 }
1146
1147 #[test]
1148 fn rolling_windows_use_configured_window_and_hop_seconds() {
1149 let planner = LiveWindowPlanner::new(LiveWindowingConfig {
1150 window_seconds: 4.0,
1151 hop_seconds: 1.5,
1152 finalize_lag_seconds: 5.0,
1153 max_buffer_lag_seconds: 30.0,
1154 stability_tolerance_seconds: 0.4,
1155 })
1156 .expect("valid planner");
1157
1158 assert_eq!(
1159 planner.ready_windows(8.2),
1160 vec![
1161 LiveWindow {
1162 start_seconds: 0.0,
1163 end_seconds: 4.0,
1164 },
1165 LiveWindow {
1166 start_seconds: 1.5,
1167 end_seconds: 5.5,
1168 },
1169 LiveWindow {
1170 start_seconds: 3.0,
1171 end_seconds: 7.0,
1172 },
1173 ]
1174 );
1175 }
1176
1177 #[test]
1178 fn finalized_segments_are_not_revised_by_later_windows() {
1179 let mut state = LiveWindowState::new(LiveWindowingConfig::default()).expect("valid config");
1180
1181 state
1182 .observe_window(observation(
1183 "session-1",
1184 0.0,
1185 5.0,
1186 5.0,
1187 vec![candidate(0.4, 1.8, "hello world")],
1188 ))
1189 .expect("first window accepted");
1190 let finalizing_events = state
1191 .observe_window(observation(
1192 "session-1",
1193 2.5,
1194 7.5,
1195 7.5,
1196 vec![candidate(0.45, 1.75, "hello world")],
1197 ))
1198 .expect("second window accepted");
1199 let revised_events = state
1200 .observe_window(observation(
1201 "session-1",
1202 5.0,
1203 10.0,
1204 10.0,
1205 vec![candidate(0.42, 1.78, "changed text")],
1206 ))
1207 .expect("later window accepted");
1208 let matching_events_after_final = state
1209 .observe_window(observation(
1210 "session-1",
1211 7.4,
1212 12.4,
1213 12.4,
1214 vec![candidate(0.44, 1.76, "hello world")],
1215 ))
1216 .expect("matching later window accepted");
1217
1218 let final_events = finalizing_events
1219 .iter()
1220 .chain(revised_events.iter())
1221 .chain(matching_events_after_final.iter())
1222 .filter_map(|event| match event {
1223 LiveTranscriptEvent::Final(final_segment) => Some(final_segment),
1224 _ => None,
1225 })
1226 .collect::<Vec<_>>();
1227
1228 assert_eq!(final_events.len(), 1);
1229 assert_eq!(final_events[0].text, "hello world");
1230 assert_eq!(state.final_segment_count(), 1);
1231 }
1232
1233 #[test]
1234 fn lag_failure_emits_error_instead_of_dropping_speech() {
1235 let mut state = LiveWindowState::new(LiveWindowingConfig {
1236 window_seconds: 5.0,
1237 hop_seconds: 2.5,
1238 finalize_lag_seconds: 5.0,
1239 max_buffer_lag_seconds: 3.0,
1240 stability_tolerance_seconds: 0.4,
1241 })
1242 .expect("valid config");
1243
1244 let events = state
1245 .observe_window(observation(
1246 "session-1",
1247 0.0,
1248 5.0,
1249 9.1,
1250 vec![candidate(0.4, 1.8, "hello world")],
1251 ))
1252 .expect("lag is reported as an event");
1253
1254 assert_eq!(
1255 events,
1256 vec![LiveTranscriptEvent::Error(LiveTranscriptError {
1257 session_id: "session-1".to_string(),
1258 sequence: 1,
1259 message: "processing fell behind live input by 4.100 seconds".to_string(),
1260 recoverable: false,
1261 })]
1262 );
1263 assert!(state.has_failed());
1264 }
1265
1266 fn observation(
1267 session_id: &str,
1268 window_start_seconds: f64,
1269 window_end_seconds: f64,
1270 latest_ingested_audio_seconds: f64,
1271 segments: Vec<LiveAsrSegmentCandidate>,
1272 ) -> LiveWindowTranscriptObservation {
1273 LiveWindowTranscriptObservation {
1274 session_id: session_id.to_string(),
1275 window_start_seconds,
1276 window_end_seconds,
1277 window_start_at_utc: format!("2026-06-22T15:30:{window_start_seconds:02.0}Z"),
1278 window_end_at_utc: format!("2026-06-22T15:30:{window_end_seconds:02.0}Z"),
1279 latest_ingested_audio_seconds,
1280 segments,
1281 }
1282 }
1283
1284 fn candidate(start_seconds: f64, end_seconds: f64, text: &str) -> LiveAsrSegmentCandidate {
1285 LiveAsrSegmentCandidate {
1286 start_seconds,
1287 end_seconds,
1288 start_at_utc: format!("2026-06-22T15:30:{start_seconds:06.3}Z"),
1289 end_at_utc: format!("2026-06-22T15:30:{end_seconds:06.3}Z"),
1290 text: text.to_string(),
1291 language: Some("en".to_string()),
1292 }
1293 }
1294
1295 #[test]
1296 fn session_started_serializes_camel_case_event_contract() {
1297 let event = LiveTranscriptEvent::SessionStarted(LiveSessionStarted {
1298 session_id: "session-1".to_string(),
1299 sequence: 1,
1300 source: "rtsp://camera/live".to_string(),
1301 ingest_started_at_utc: "2026-06-22T15:30:00Z".to_string(),
1302 sample_rate: 16_000,
1303 channels: 1,
1304 model_id: "tiny.en".to_string(),
1305 language: Some("en".to_string()),
1306 });
1307
1308 let json = serde_json::to_value(&event).expect("event serializes");
1309
1310 assert_eq!(
1311 json,
1312 serde_json::json!({
1313 "event": "sessionStarted",
1314 "sessionId": "session-1",
1315 "sequence": 1,
1316 "source": "rtsp://camera/live",
1317 "ingestStartedAtUtc": "2026-06-22T15:30:00Z",
1318 "sampleRate": 16000,
1319 "channels": 1,
1320 "modelId": "tiny.en",
1321 "language": "en"
1322 })
1323 );
1324 }
1325
1326 #[test]
1327 fn live_transcript_events_serialize_all_jsonl_shapes() {
1328 let events = vec![
1329 LiveTranscriptEvent::Partial(LivePartialTranscript {
1330 session_id: "session-1".to_string(),
1331 sequence: 2,
1332 window_start_seconds: 0.0,
1333 window_end_seconds: 5.0,
1334 window_start_at_utc: "2026-06-22T15:30:00Z".to_string(),
1335 window_end_at_utc: "2026-06-22T15:30:05Z".to_string(),
1336 text: "hello wor".to_string(),
1337 segments: vec![LivePartialSegment {
1338 start_seconds: 0.4,
1339 end_seconds: 1.8,
1340 text: "hello wor".to_string(),
1341 language: Some("en".to_string()),
1342 }],
1343 }),
1344 LiveTranscriptEvent::Final(LiveFinalTranscriptSegment {
1345 session_id: "session-1".to_string(),
1346 sequence: 3,
1347 segment_id: "seg-000001".to_string(),
1348 start_seconds: 0.4,
1349 end_seconds: 1.9,
1350 start_at_utc: "2026-06-22T15:30:00.400Z".to_string(),
1351 end_at_utc: "2026-06-22T15:30:01.900Z".to_string(),
1352 text: "hello world".to_string(),
1353 language: Some("en".to_string()),
1354 }),
1355 LiveTranscriptEvent::Error(LiveTranscriptError {
1356 session_id: "session-1".to_string(),
1357 sequence: 4,
1358 message: "processing fell behind live input".to_string(),
1359 recoverable: false,
1360 }),
1361 LiveTranscriptEvent::SessionEnded(LiveSessionEnded {
1362 session_id: "session-1".to_string(),
1363 sequence: 5,
1364 reason: LiveSessionEndReason::Error,
1365 processed_audio_seconds: 12.5,
1366 final_segment_count: 1,
1367 }),
1368 ];
1369
1370 let jsonl = live_transcript_events_to_jsonl(&events).expect("events serialize");
1371 let lines = jsonl.lines().collect::<Vec<_>>();
1372
1373 assert_eq!(lines.len(), 4);
1374 assert_eq!(
1375 serde_json::from_str::<serde_json::Value>(lines[0]).expect("partial json"),
1376 serde_json::json!({
1377 "event": "partial",
1378 "sessionId": "session-1",
1379 "sequence": 2,
1380 "windowStartSeconds": 0.0,
1381 "windowEndSeconds": 5.0,
1382 "windowStartAtUtc": "2026-06-22T15:30:00Z",
1383 "windowEndAtUtc": "2026-06-22T15:30:05Z",
1384 "text": "hello wor",
1385 "segments": [{
1386 "startSeconds": 0.4,
1387 "endSeconds": 1.8,
1388 "text": "hello wor",
1389 "language": "en"
1390 }]
1391 })
1392 );
1393 assert_eq!(
1394 serde_json::from_str::<serde_json::Value>(lines[1]).expect("final json"),
1395 serde_json::json!({
1396 "event": "final",
1397 "sessionId": "session-1",
1398 "sequence": 3,
1399 "segmentId": "seg-000001",
1400 "startSeconds": 0.4,
1401 "endSeconds": 1.9,
1402 "startAtUtc": "2026-06-22T15:30:00.400Z",
1403 "endAtUtc": "2026-06-22T15:30:01.900Z",
1404 "text": "hello world",
1405 "language": "en"
1406 })
1407 );
1408 assert_eq!(
1409 serde_json::from_str::<serde_json::Value>(lines[2]).expect("error json"),
1410 serde_json::json!({
1411 "event": "error",
1412 "sessionId": "session-1",
1413 "sequence": 4,
1414 "message": "processing fell behind live input",
1415 "recoverable": false
1416 })
1417 );
1418 assert_eq!(
1419 serde_json::from_str::<serde_json::Value>(lines[3]).expect("ended json"),
1420 serde_json::json!({
1421 "event": "sessionEnded",
1422 "sessionId": "session-1",
1423 "sequence": 5,
1424 "reason": "error",
1425 "processedAudioSeconds": 12.5,
1426 "finalSegmentCount": 1
1427 })
1428 );
1429 assert!(jsonl.ends_with('\n'));
1430 }
1431}