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;
@@ -270,6 +272,11 @@
    runtime_turn_bridge_url: Option<String>,
    runtime_turn_bridge_token: Option<String>,
    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,
@@ -332,6 +339,11 @@
    bridge_url: Option<String>,
    bridge_token: Option<String>,
    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>,
@@ -343,6 +355,11 @@
            bridge_url: config.runtime_turn_bridge_url.clone(),
            bridge_token: config.runtime_turn_bridge_token.clone(),
            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(),
@@ -374,6 +391,32 @@
                .as_deref()
                .is_some_and(|value| value.trim_end_matches('/').ends_with("/stream"))
    }

    fn is_asr_stream_ready(&self) -> bool {
        self.asr_stream_enabled
            && self
                .asr_stream_url
                .as_ref()
                .is_some_and(|value| !value.is_empty())
            && self
                .bridge_token
                .as_ref()
                .is_some_and(|value| !value.is_empty())
            && self
                .runtime_session_nonce
                .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 {
@@ -395,6 +438,14 @@
            runtime_turn_bridge_token: optional_env("CV_RUNTIME_TURN_BRIDGE_TOKEN"),
            runtime_turn_bridge_mode: env::var("CV_RUNTIME_TURN_BRIDGE_MODE")
                .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),
@@ -751,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() {
@@ -807,9 +859,13 @@
                    "turnArtifactBytes": byte_size,
                    "frameCount": turn.frame_count,
                    "sampleCount": turn.sample_count,
                    "endReason": turn.end_reason,
                    "endReason": turn.end_reason.as_str(),
                }),
            );
            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,
@@ -820,6 +876,7 @@
                    &turn,
                    &path_ref,
                    byte_size,
                    asr_result_ref.as_deref(),
                    turn_pipeline_started_at,
                )
                .await
@@ -883,6 +940,7 @@
                &turn,
                &path_ref,
                byte_size,
                asr_result_ref.as_deref(),
                turn_pipeline_started_at,
            )
            .await
@@ -1004,6 +1062,251 @@
    Ok((path_ref, byte_size))
}

async fn request_asr_result_ref(
    http: &Client,
    bridge_config: &TurnBridgeConfig,
    call_id: &str,
    trace_id: &str,
    turn: &FinishedSpeechTurn,
) -> Option<String> {
    if !bridge_config.is_asr_stream_ready() {
        return None;
    }
    match request_asr_stream(http, bridge_config, call_id, trace_id, turn).await {
        Ok(Some(asr_result_ref)) => {
            emit_activity(
                call_id,
                trace_id,
                Some(&turn.turn_id),
                "asr_stream_ref_ready",
                "ok",
                None,
                None,
                json!({
                    "asrResultRefPresent": true,
                    "format": "pcm_s16le",
                    "sampleRate": 16000,
                    "channels": 1,
                }),
            );
            Some(asr_result_ref)
        }
        Ok(None) => 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_stream_failed_fallback"
            );
            emit_activity(
                call_id,
                trace_id,
                Some(&turn.turn_id),
                "asr_stream_fallback",
                "skipped",
                Some("ASR_STREAM_INTERRUPTED"),
                Some(true),
                json!({
                    "fallbackReason": "asr_stream_request_failed",
                    "fallbackStage": "asr_stream",
                }),
            );
            None
        }
    }
}

async fn request_asr_stream(
    http: &Client,
    bridge_config: &TurnBridgeConfig,
    call_id: &str,
    trace_id: &str,
    turn: &FinishedSpeechTurn,
) -> Result<Option<String>> {
    let started_at = Instant::now();
    let ndjson = build_asr_stream_ndjson(call_id, trace_id, turn, bridge_config)?;
    let response = http
        .post(bridge_config.asr_stream_url.as_deref().unwrap_or_default())
        .header("Content-Type", "application/x-ndjson")
        .header(
            "X-CV-Runtime-Token",
            bridge_config.bridge_token.as_deref().unwrap_or_default(),
        )
        .header("X-CV-Call-Id", call_id)
        .header("X-CV-Trace-Id", trace_id)
        .header(
            "X-CV-Runtime-Session-Nonce",
            bridge_config
                .runtime_session_nonce
                .as_deref()
                .unwrap_or_default(),
        )
        .body(ndjson)
        .send()
        .await
        .context("failed to post asr stream")?;
    let status = response.status();
    if !status.is_success() {
        let body_len = response.text().await.map(|body| body.len()).unwrap_or(0);
        warn!(
            call_id = %call_id,
            trace_id = %trace_id,
            turn_id = %turn.turn_id,
            http_status = status.as_u16(),
            body_len,
            "runtime helper asr_stream_http_failed"
        );
        return Ok(None);
    }
    let body: RuntimeTurnCommonResult<RuntimeAsrStreamResp> = response
        .json()
        .await
        .context("failed to decode asr stream response")?;
    if body.code != 0 {
        warn!(
            call_id = %call_id,
            trace_id = %trace_id,
            turn_id = %turn.turn_id,
            code = body.code,
            msg_len = body.msg.as_deref().unwrap_or_default().len(),
            "runtime helper asr_stream_common_result_failed"
        );
        return Ok(None);
    }
    let Some(data) = body.data else {
        return Ok(None);
    };
    if data.status.as_deref() == Some("final") {
        info!(
            call_id = %call_id,
            trace_id = %trace_id,
            turn_id = %turn.turn_id,
            chunk_count = data.chunk_count.unwrap_or_default(),
            audio_bytes = data.audio_bytes.unwrap_or_default(),
            asr_duration_ms = data.asr_duration_ms.unwrap_or_default(),
            wall_ms = started_at.elapsed().as_millis() as u64,
            provider = %data.provider_alias.as_deref().unwrap_or("unknown"),
            text_len = data.text_len.unwrap_or_default(),
            "runtime helper asr_stream_final"
        );
        return Ok(data.asr_result_ref);
    }
    emit_activity(
        call_id,
        trace_id,
        Some(&turn.turn_id),
        "asr_stream_fallback",
        "skipped",
        None,
        Some(true),
        json!({
            "fallbackReason": data.fallback_reason,
            "fallbackStage": data.fallback_stage,
            "status": data.status,
        }),
    );
    Ok(None)
}

fn build_asr_stream_ndjson(
    call_id: &str,
    trace_id: &str,
    turn: &FinishedSpeechTurn,
    bridge_config: &TurnBridgeConfig,
) -> Result<String> {
    let chunks = asr_pcm_16k_chunks(turn)?;
    let mut seq = 1u64;
    let mut lines = Vec::with_capacity(chunks.len() + 3);
    lines.push(serde_json::to_string(&json!({
        "event": "asr_stream_started",
        "seq": seq,
        "callId": call_id,
        "traceId": trace_id,
        "turnId": turn.turn_id.as_str(),
        "tsMs": current_time_millis(),
        "payload": {
            "format": "pcm_s16le",
            "sampleRate": 16000,
            "channels": 1,
            "runtimeSessionNonce": bridge_config.runtime_session_nonce.as_deref().unwrap_or_default(),
            "providerHint": "volcengine",
        }
    }))?);
    for (index, samples) in chunks.iter().enumerate() {
        seq += 1;
        let bytes = pcm_i16_to_le_bytes(samples);
        let duration_ms = ((samples.len() as u64) * 1000 / 16_000).max(1);
        lines.push(serde_json::to_string(&json!({
            "event": "asr_audio_chunk",
            "seq": seq,
            "callId": call_id,
            "traceId": trace_id,
            "turnId": turn.turn_id.as_str(),
            "tsMs": current_time_millis(),
            "payload": {
                "chunkSeq": index + 1,
                "format": "pcm_s16le",
                "sampleRate": 16000,
                "channels": 1,
                "durationMs": duration_ms,
                "payloadBase64": general_purpose::STANDARD.encode(bytes),
            }
        }))?);
    }
    seq += 1;
    lines.push(serde_json::to_string(&json!({
        "event": "vad_speech_end",
        "seq": seq,
        "callId": call_id,
        "traceId": trace_id,
        "turnId": turn.turn_id.as_str(),
        "tsMs": current_time_millis(),
        "payload": {
            "endReason": turn.end_reason.as_str(),
            "speechDurationMs": turn.duration_ms,
        }
    }))?);
    seq += 1;
    lines.push(serde_json::to_string(&json!({
        "event": "asr_stream_finish",
        "seq": seq,
        "callId": call_id,
        "traceId": trace_id,
        "turnId": turn.turn_id.as_str(),
        "tsMs": current_time_millis(),
        "payload": {
            "finalChunkSeq": chunks.len(),
            "audioDurationMs": turn.duration_ms,
        }
    }))?);
    Ok(lines.join("\n") + "\n")
}

fn asr_pcm_16k_chunks(turn: &FinishedSpeechTurn) -> Result<Vec<Vec<i16>>> {
    if turn.samples.is_empty() {
        return Err(anyhow!("empty turn samples"));
    }
    let samples_16k: Vec<i16> = turn.samples.iter().step_by(3).copied().collect();
    if samples_16k.is_empty() {
        return Err(anyhow!("empty 16k asr samples"));
    }
    let samples_per_chunk = 320usize;
    Ok(samples_16k
        .chunks(samples_per_chunk)
        .map(|chunk| chunk.to_vec())
        .collect())
}

fn pcm_i16_to_le_bytes(samples: &[i16]) -> Vec<u8> {
    let mut bytes = Vec::with_capacity(samples.len() * 2);
    for sample in samples {
        bytes.extend_from_slice(&sample.to_le_bytes());
    }
    bytes
}

async fn request_turn_bridge_stream(
    http: &Client,
    bridge_config: &TurnBridgeConfig,
@@ -1013,6 +1316,7 @@
    turn: &FinishedSpeechTurn,
    path_ref: &str,
    byte_size: u64,
    asr_result_ref: Option<&str>,
    turn_pipeline_started_at: Instant,
) -> Result<RuntimeTurnStreamOutcome> {
    let bridge_started_at = Instant::now();
@@ -1020,7 +1324,7 @@
        call_id: call_id.to_string(),
        trace_id: trace_id.to_string(),
        turn_id: turn.turn_id.clone(),
        audio_artifact: RuntimeTurnAudioArtifact {
        audio_artifact: Some(RuntimeTurnAudioArtifact {
            artifact_type: "local_file".to_string(),
            path_ref: path_ref.to_string(),
            format: "wav".to_string(),
@@ -1028,7 +1332,8 @@
            channels: u32::from(USER_AUDIO_NUM_CHANNELS),
            duration_ms: turn.duration_ms,
            byte_size,
        },
        }),
        asr_result_ref: asr_result_ref.map(str::to_string),
    };
    let response = http
        .post(bridge_config.bridge_url.as_deref().unwrap_or_default())
@@ -1621,6 +1926,7 @@
    turn: &FinishedSpeechTurn,
    path_ref: &str,
    byte_size: u64,
    asr_result_ref: Option<&str>,
    turn_pipeline_started_at: Instant,
) -> Result<RuntimeTurnBridgeOutcome> {
    let bridge_started_at = Instant::now();
@@ -1628,7 +1934,7 @@
        call_id: call_id.to_string(),
        trace_id: trace_id.to_string(),
        turn_id: turn.turn_id.clone(),
        audio_artifact: RuntimeTurnAudioArtifact {
        audio_artifact: Some(RuntimeTurnAudioArtifact {
            artifact_type: "local_file".to_string(),
            path_ref: path_ref.to_string(),
            format: "wav".to_string(),
@@ -1636,7 +1942,8 @@
            channels: u32::from(USER_AUDIO_NUM_CHANNELS),
            duration_ms: turn.duration_ms,
            byte_size,
        },
        }),
        asr_result_ref: asr_result_ref.map(str::to_string),
    };
    let response = http
        .post(bridge_config.bridge_url.as_deref().unwrap_or_default())
@@ -1927,7 +2234,10 @@
    #[serde(rename = "turnId")]
    turn_id: String,
    #[serde(rename = "audioArtifact")]
    audio_artifact: RuntimeTurnAudioArtifact,
    #[serde(skip_serializing_if = "Option::is_none")]
    audio_artifact: Option<RuntimeTurnAudioArtifact>,
    #[serde(rename = "asrResultRef", skip_serializing_if = "Option::is_none")]
    asr_result_ref: Option<String>,
}

#[derive(Serialize)]
@@ -1953,6 +2263,20 @@
    data: Option<T>,
    stage: Option<String>,
    retryable: Option<bool>,
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct RuntimeAsrStreamResp {
    status: Option<String>,
    asr_result_ref: Option<String>,
    chunk_count: Option<u64>,
    audio_bytes: Option<u64>,
    asr_duration_ms: Option<u64>,
    provider_alias: Option<String>,
    text_len: Option<u64>,
    fallback_reason: Option<String>,
    fallback_stage: Option<String>,
}

#[derive(Default)]
@@ -2152,6 +2476,7 @@
        } else {
            None
        };
        let mut realtime_asr_upload: Option<RealtimeAsrUpload> = None;

        while let Some(frame) = stream.next().await {
            frame_count += 1;
@@ -2184,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,
@@ -2192,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,
@@ -2200,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,
@@ -2224,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!(
@@ -2242,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,