feat: stream realtime asr from helper

cai

2026-07-10 a590c735cfa4ec55ba6b32e297600d83e1a3e46a

 src/main.rs

@@ -1,3 +1,4 @@
mod asr_realtime;
mod audio;
mod service;

@@ -12,6 +13,7 @@
};

use anyhow::{Context, Result, anyhow};
use asr_realtime::{RealtimeAsrConfig, RealtimeAsrOutcome, RealtimeAsrUpload};
use audio::{AudioDiagnostics, load_pre_recorded_frames};
use base64::{Engine as _, engine::general_purpose};
use futures_util::StreamExt;
@@ -272,6 +274,9 @@
    runtime_turn_bridge_mode: String,
    runtime_asr_stream_enabled: bool,
    runtime_asr_stream_url: Option<String>,
    runtime_asr_realtime_enabled: bool,
    runtime_asr_realtime_url: Option<String>,
    runtime_asr_realtime_chunk_duration_ms: u64,
    runtime_turn_artifact_dir: Option<String>,
    runtime_session_nonce: Option<String>,
    user_audio_observer_enabled: bool,
@@ -336,6 +341,9 @@
    bridge_mode: String,
    asr_stream_enabled: bool,
    asr_stream_url: Option<String>,
    asr_realtime_enabled: bool,
    asr_realtime_url: Option<String>,
    asr_realtime_chunk_duration_ms: u64,
    artifact_dir: Option<String>,
    runtime_session_nonce: Option<String>,
    audio_debug_dump_dir: Option<String>,
@@ -349,6 +357,9 @@
            bridge_mode: config.runtime_turn_bridge_mode.clone(),
            asr_stream_enabled: config.runtime_asr_stream_enabled,
            asr_stream_url: config.runtime_asr_stream_url.clone(),
            asr_realtime_enabled: config.runtime_asr_realtime_enabled,
            asr_realtime_url: config.runtime_asr_realtime_url.clone(),
            asr_realtime_chunk_duration_ms: config.runtime_asr_realtime_chunk_duration_ms,
            artifact_dir: config.runtime_turn_artifact_dir.clone(),
            runtime_session_nonce: config.runtime_session_nonce.clone(),
            audio_debug_dump_dir: config.audio_debug_dump_dir.clone(),
@@ -396,6 +407,16 @@
                .as_ref()
                .is_some_and(|value| !value.is_empty())
    }

    fn realtime_asr_config(&self) -> RealtimeAsrConfig {
        RealtimeAsrConfig {
            enabled: self.asr_realtime_enabled,
            url: self.asr_realtime_url.clone(),
            runtime_token: self.bridge_token.clone(),
            runtime_session_nonce: self.runtime_session_nonce.clone(),
            chunk_duration_ms: self.asr_realtime_chunk_duration_ms,
        }
    }
}

impl Config {
@@ -419,6 +440,12 @@
                .unwrap_or_else(|_| "json".to_string()),
            runtime_asr_stream_enabled: bool_env("CV_RUNTIME_ASR_STREAM_ENABLED", false),
            runtime_asr_stream_url: optional_env("CV_RUNTIME_ASR_STREAM_URL"),
            runtime_asr_realtime_enabled: bool_env("CV_RUNTIME_ASR_REALTIME_ENABLED", false),
            runtime_asr_realtime_url: optional_env("CV_RUNTIME_ASR_REALTIME_URL"),
            runtime_asr_realtime_chunk_duration_ms: u64_env(
                "CV_RUNTIME_ASR_REALTIME_CHUNK_DURATION_MS",
                200,
            ),
            runtime_turn_artifact_dir: optional_env("CV_RUNTIME_TURN_ARTIFACT_DIR"),
            runtime_session_nonce: optional_env("CV_RUNTIME_SESSION_NONCE"),
            user_audio_observer_enabled: bool_env("CV_ENABLE_USER_AUDIO_OBSERVER", true),
@@ -775,6 +802,7 @@
    call_id: &str,
    trace_id: &str,
    turn: FinishedSpeechTurn,
    realtime_asr_result_ref: Option<String>,
) {
    let turn_pipeline_started_at = Instant::now();
    if !bridge_config.is_ready() {
@@ -834,8 +862,10 @@
                    "endReason": turn.end_reason.as_str(),
                }),
            );
            let asr_result_ref =
                request_asr_result_ref(http, bridge_config, call_id, trace_id, &turn).await;
            let asr_result_ref = match realtime_asr_result_ref {
                Some(value) => Some(value),
                None => request_asr_result_ref(http, bridge_config, call_id, trace_id, &turn).await,
            };
            if bridge_config.is_stream_mode() {
                match request_turn_bridge_stream(
                    http,
@@ -2446,6 +2476,7 @@
        } else {
            None
        };
        let mut realtime_asr_upload: Option<RealtimeAsrUpload> = None;

        while let Some(frame) = stream.next().await {
            frame_count += 1;
@@ -2478,7 +2509,8 @@

            if let Some(vad) = simple_vad.as_mut() {
                if vad_enabled_gate.load(Ordering::Acquire) {
                    if let Some(turn) = vad.observe_frame(
                    let was_in_speech = vad.in_speech;
                    let turn = vad.observe_frame(
                        &call_id,
                        &trace_id,
                        &participant_alias,
@@ -2486,7 +2518,65 @@
                        frame_count,
                        elapsed_ms,
                        &frame,
                    ) {
                    );
                    let is_in_speech = vad.in_speech;

                    if !was_in_speech && is_in_speech {
                        let turn_id = format!("turn-{:04}", vad.turn_index);
                        match RealtimeAsrUpload::start(
                            http.clone(),
                            turn_bridge_config.realtime_asr_config(),
                            &call_id,
                            &trace_id,
                            &turn_id,
                            &vad.speech_samples,
                        ) {
                            Ok(upload) => {
                                info!(
                                    call_id = %call_id,
                                    trace_id = %trace_id,
                                    turn_id = %turn_id,
                                    "runtime helper asr_realtime_session_started"
                                );
                                realtime_asr_upload = Some(upload);
                            }
                            Err(error) if turn_bridge_config.asr_realtime_enabled => {
                                warn!(
                                    call_id = %call_id,
                                    trace_id = %trace_id,
                                    turn_id = %turn_id,
                                    error = %safe_error(&error.to_string()),
                                    "runtime helper asr_realtime_start_failed_fallback"
                                );
                            }
                            Err(_) => {}
                        }
                    } else if was_in_speech {
                        let push_failed = realtime_asr_upload
                            .as_mut()
                            .and_then(|upload| upload.push_48k_samples(frame.data.as_ref()).err());
                        if let Some(error) = push_failed {
                            warn!(
                                call_id = %call_id,
                                trace_id = %trace_id,
                                error = %safe_error(&error.to_string()),
                                "runtime helper asr_realtime_upload_failed_fallback"
                            );
                            if let Some(upload) = realtime_asr_upload.take() {
                                tokio::spawn(async move {
                                    upload.cancel("upload_backpressure").await;
                                });
                            }
                        }
                    }

                    if let Some(turn) = turn {
                        let realtime_asr_result_ref = match realtime_asr_upload.take() {
                            Some(upload) => {
                                finish_realtime_asr_upload(upload, &call_id, &trace_id, &turn).await
                            }
                            None => None,
                        };
                        handle_finished_turn(
                            &http,
                            &turn_bridge_config,
@@ -2494,10 +2584,22 @@
                            &call_id,
                            &trace_id,
                            turn,
                            realtime_asr_result_ref,
                        )
                        .await;
                    } else if was_in_speech && !is_in_speech {
                        if let Some(upload) = realtime_asr_upload.take() {
                            tokio::spawn(async move {
                                upload.cancel("speech_too_short").await;
                            });
                        }
                    }
                } else {
                    if let Some(upload) = realtime_asr_upload.take() {
                        tokio::spawn(async move {
                            upload.cancel("vad_disabled").await;
                        });
                    }
                    vad.observe_disabled_frame(
                        &call_id,
                        &trace_id,
@@ -2518,9 +2620,26 @@
                &track_sid_alias,
                started_at.elapsed().as_millis() as u64,
            ) {
                handle_finished_turn(&http, &turn_bridge_config, &sink, &call_id, &trace_id, turn)
                    .await;
                let realtime_asr_result_ref = match realtime_asr_upload.take() {
                    Some(upload) => {
                        finish_realtime_asr_upload(upload, &call_id, &trace_id, &turn).await
                    }
                    None => None,
                };
                handle_finished_turn(
                    &http,
                    &turn_bridge_config,
                    &sink,
                    &call_id,
                    &trace_id,
                    turn,
                    realtime_asr_result_ref,
                )
                .await;
            }
        }
        if let Some(upload) = realtime_asr_upload.take() {
            upload.cancel("stream_end").await;
        }

        info!(
@@ -2536,6 +2655,58 @@
    })
}

async fn finish_realtime_asr_upload(
    upload: RealtimeAsrUpload,
    call_id: &str,
    trace_id: &str,
    turn: &FinishedSpeechTurn,
) -> Option<String> {
    match upload.finish(turn.duration_ms, &turn.end_reason).await {
        Ok(RealtimeAsrOutcome {
            status,
            asr_result_ref,
            provider_alias,
            partial_count,
            fallback_reason,
            fallback_stage,
            chunk_count,
            audio_bytes,
            wall_ms,
        }) => {
            info!(
                call_id = %call_id,
                trace_id = %trace_id,
                turn_id = %turn.turn_id,
                status = %status,
                provider_alias = ?provider_alias,
                partial_count,
                chunk_count,
                audio_bytes,
                wall_ms,
                asr_result_ref_present = asr_result_ref.is_some(),
                fallback_reason = ?fallback_reason,
                fallback_stage = ?fallback_stage,
                "runtime helper asr_realtime_finished"
            );
            if status == "final" {
                asr_result_ref
            } else {
                None
            }
        }
        Err(error) => {
            warn!(
                call_id = %call_id,
                trace_id = %trace_id,
                turn_id = %turn.turn_id,
                error = %safe_error(&error.to_string()),
                "runtime helper asr_realtime_failed_fallback"
            );
            None
        }
    }
}

#[derive(Clone)]
struct SimpleVadConfig {
    rms_threshold: f64,