feat: upload user turns to asr stream

cai

2026-07-09 58a968bc9e632fc48a140afba514edf642e93fa6

feat: upload user turns to asr stream

 src/main.rs

@@ -270,6 +270,8 @@
    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_turn_artifact_dir: Option<String>,
    runtime_session_nonce: Option<String>,
    user_audio_observer_enabled: bool,
@@ -332,6 +334,8 @@
    bridge_url: Option<String>,
    bridge_token: Option<String>,
    bridge_mode: String,
    asr_stream_enabled: bool,
    asr_stream_url: Option<String>,
    artifact_dir: Option<String>,
    runtime_session_nonce: Option<String>,
    audio_debug_dump_dir: Option<String>,
@@ -343,6 +347,8 @@
            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(),
            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 +380,22 @@
                .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())
    }
}

impl Config {
@@ -395,6 +417,8 @@
            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_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),
@@ -807,9 +831,11 @@
                    "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 =
                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 +846,7 @@
                    &turn,
                    &path_ref,
                    byte_size,
                    asr_result_ref.as_deref(),
                    turn_pipeline_started_at,
                )
                .await
@@ -883,6 +910,7 @@
                &turn,
                &path_ref,
                byte_size,
                asr_result_ref.as_deref(),
                turn_pipeline_started_at,
            )
            .await
@@ -1004,6 +1032,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 +1286,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 +1294,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 +1302,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 +1896,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 +1904,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 +1912,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 +2204,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)]
@@ -1955,6 +2235,20 @@
    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)]
struct RuntimeTurnBridgeOutcome {
    reply_audio_artifact: Option<RuntimeTurnReplyAudioArtifact>,

 src/service.rs

@@ -404,6 +404,12 @@
        if let Some(value) = runtime.vad_post_greeting_delay_ms {
            command.env("CV_VAD_POST_GREETING_DELAY_MS", value.to_string());
        }
        if let Some(value) = runtime.asr_streaming_enabled {
            command.env("CV_RUNTIME_ASR_STREAM_ENABLED", value.to_string());
        }
        if let Some(value) = &runtime.asr_stream_url {
            command.env("CV_RUNTIME_ASR_STREAM_URL", value);
        }
        if let Some(true) = runtime.audio_debug_dump_enabled {
            if let Some(value) = &config.audio_debug_dump_dir {
                command.env("CV_AUDIO_DEBUG_DUMP_DIR", value);
@@ -968,6 +974,8 @@
    vad_initial_ignore_ms: Option<u64>,
    vad_post_greeting_delay_ms: Option<u64>,
    audio_debug_dump_enabled: Option<bool>,
    asr_streaming_enabled: Option<bool>,
    asr_stream_url: Option<String>,
}

#[derive(Default, Deserialize)]