feat: stream realtime asr from helper

cai

2026-07-10 a590c735cfa4ec55ba6b32e297600d83e1a3e46a

feat: stream realtime asr from helper

 Cargo.lock

@@ -240,6 +240,7 @@
 "reqwest",
 "serde",
 "serde_json",
 "sha2",
 "tokio",
 "tracing",
 "tracing-subscriber",

 Cargo.toml

@@ -15,6 +15,7 @@
reqwest = { version = "0.12.24", default-features = false, features = ["rustls-tls", "json", "stream"] }
serde = { version = "1.0.228", features = ["derive"] }
serde_json = "1.0.145"
sha2 = "0.10.9"
tokio = { version = "1.48.0", features = ["macros", "rt-multi-thread", "signal", "time"] }
tracing = "0.1.41"
tracing-subscriber = { version = "0.3.20", features = ["env-filter", "fmt"] }

 README.md

@@ -54,9 +54,12 @@
```bash
node tools/validate-turn-stream-fixture.mjs fixtures/turn-stream-happy.ndjson
node tools/validate-turn-stream-fixture.mjs fixtures/turn-stream-mp3-chunks.ndjson
node tools/validate-asr-realtime-fixture.mjs fixtures/asr-realtime-happy.ndjson
```

这条快速线只校验 NDJSON contract、事件顺序和 `pcm_s16le` / `mp3` chunk 基本约束，用于提前发现 `replyPlaybackMode`、`reply_state`、`reply_audio_chunk`、`turn_completed` 等字段破坏。它不能替代 Docker 镜像构建、真实 LiveKit smoke 或 iPhone 真机验收。

ASR realtime fixture 额外校验 `session_start -> audio_chunk -> vad_speech_end -> finish`、连续 `chunkSeq`、`pcm_s16le / 16000Hz / mono` 与 `24KiB` 单行上限。它只验证 helper 到 Java 的 wire contract，不代表 provider partial/final 已通过。

## 本机运行

@@ -98,6 +101,8 @@

`sessions/start` 收到 Java 传入的 LiveKit bot 入房材料后，会拉起现有 worker 子进程承接媒体链路。helper service 本身不做 ASR / LLM / TTS / 消息 / 计费，也不持久化业务数据。

当 Java 在 `sessions/start.runtime` 下发 `asrRealtimeEnabled=true`、`asrRealtimeUrl` 与 `asrRealtimeChunkDurationMs=200` 时，worker 会从 VAD speech start 起接收 20ms 用户音频帧，聚合成 `16kHz / mono / 200ms` NDJSON chunk 持续上传给 Java；最后一块允许短于 200ms。helper 只负责音频传输和 fallback 编排，ASR provider session、partial/final 归一化、activity 和 `asrResultRef` 仍由 Java 管理；realtime 失败时只回退一次既有 final-only ASR stream。

鉴权口径：

- Java 调 helper 控制面使用 `Authorization: Bearer {helperAuthToken}`。

 docs/runtime-contract.md

@@ -27,6 +27,7 @@
```bash
node tools/validate-turn-stream-fixture.mjs fixtures/turn-stream-happy.ndjson
node tools/validate-turn-stream-fixture.mjs fixtures/turn-stream-mp3-chunks.ndjson
node tools/validate-asr-realtime-fixture.mjs fixtures/asr-realtime-happy.ndjson
```

The fixture validates only the protocol shape and fast-path invariants:

 fixtures/asr-realtime-happy.ndjson

New file
@@ -0,0 +1,5 @@
{"event":"session_start","callId":"call_fixture","traceId":"trace_fixture","turnId":"turn-0001","runtimeSessionNonceHash":"d7c778da6f43","audio":{"format":"pcm_s16le","sampleRate":16000,"channels":1}}
{"event":"audio_chunk","chunkSeq":1,"audioBase64":"AAAAAAAAAAA=","durationMs":200}
{"event":"audio_chunk","chunkSeq":2,"audioBase64":"AAAAAAAAAAA=","durationMs":200}
{"event":"vad_speech_end","speechDurationMs":400,"endReason":"silence"}
{"event":"finish"}

 src/asr_realtime.rs

New file
@@ -0,0 +1,584 @@
use std::{io, time::Duration};

use anyhow::{Context, Result, anyhow};
use base64::{Engine as _, engine::general_purpose};
use futures_util::stream;
use reqwest::Client;
use serde::Deserialize;
use serde_json::json;
use sha2::{Digest, Sha256};
use tokio::{
    sync::mpsc,
    task::JoinHandle,
    time::{Instant, timeout},
};

const SAMPLE_RATE_16K: u32 = 16_000;
const CHANNELS_MONO: u32 = 1;
const MIN_CHUNK_DURATION_MS: u64 = 20;
const MAX_CHUNK_DURATION_MS: u64 = 1_000;
const UPLOAD_QUEUE_CAPACITY: usize = 64;
const MAX_NDJSON_LINE_BYTES: usize = 24 * 1024;
const FINISH_RESPONSE_TIMEOUT: Duration = Duration::from_secs(15);
const CANCEL_RESPONSE_TIMEOUT: Duration = Duration::from_secs(3);

#[derive(Clone)]
pub(crate) struct RealtimeAsrConfig {
    pub(crate) enabled: bool,
    pub(crate) url: Option<String>,
    pub(crate) runtime_token: Option<String>,
    pub(crate) runtime_session_nonce: Option<String>,
    pub(crate) chunk_duration_ms: u64,
}

impl RealtimeAsrConfig {
    pub(crate) fn is_ready(&self) -> bool {
        self.enabled
            && non_blank(&self.url)
            && non_blank(&self.runtime_token)
            && non_blank(&self.runtime_session_nonce)
            && self.chunk_duration_ms >= MIN_CHUNK_DURATION_MS
            && self.chunk_duration_ms <= MAX_CHUNK_DURATION_MS
            && self.chunk_duration_ms % MIN_CHUNK_DURATION_MS == 0
    }
}

pub(crate) struct RealtimeAsrUpload {
    sender: Option<mpsc::Sender<Vec<u8>>>,
    task: JoinHandle<Result<RealtimeAsrOutcome>>,
    chunker: Pcm16kChunker,
}

pub(crate) struct RealtimeAsrOutcome {
    pub(crate) status: String,
    pub(crate) asr_result_ref: Option<String>,
    pub(crate) provider_alias: Option<String>,
    pub(crate) partial_count: u64,
    pub(crate) fallback_reason: Option<String>,
    pub(crate) fallback_stage: Option<String>,
    pub(crate) chunk_count: u64,
    pub(crate) audio_bytes: u64,
    pub(crate) wall_ms: u64,
}

impl RealtimeAsrUpload {
    pub(crate) fn start(
        http: Client,
        config: RealtimeAsrConfig,
        call_id: &str,
        trace_id: &str,
        turn_id: &str,
        initial_samples_48k: &[i16],
    ) -> Result<Self> {
        if !config.is_ready() {
            return Err(anyhow!("realtime asr config is not ready"));
        }
        let nonce = config.runtime_session_nonce.as_deref().unwrap_or_default();
        let session_line = session_start_line(call_id, trace_id, turn_id, nonce)?;
        let (sender, receiver) = mpsc::channel(UPLOAD_QUEUE_CAPACITY);
        let request_call_id = call_id.to_string();
        let request_trace_id = trace_id.to_string();
        let request_turn_id = turn_id.to_string();
        let task = tokio::spawn(run_upload(
            http,
            config.clone(),
            request_call_id,
            request_trace_id,
            request_turn_id,
            receiver,
        ));
        let mut upload = Self {
            sender: Some(sender),
            task,
            chunker: Pcm16kChunker::new(config.chunk_duration_ms),
        };
        if let Err(error) = upload
            .try_send_line(session_line)
            .and_then(|_| upload.push_48k_samples(initial_samples_48k))
        {
            upload.task.abort();
            return Err(error);
        }
        Ok(upload)
    }

    pub(crate) fn push_48k_samples(&mut self, samples: &[i16]) -> Result<()> {
        for chunk in self.chunker.push_48k(samples) {
            let line = audio_chunk_line(chunk.seq, &chunk.samples)?;
            self.try_send_line(line)?;
        }
        Ok(())
    }

    pub(crate) async fn finish(
        mut self,
        speech_duration_ms: u64,
        end_reason: &str,
    ) -> Result<RealtimeAsrOutcome> {
        if let Some(chunk) = self.chunker.flush() {
            let line = audio_chunk_line(chunk.seq, &chunk.samples)?;
            self.try_send_line(line)?;
        }
        self.try_send_line(vad_speech_end_line(speech_duration_ms, end_reason)?)?;
        self.try_send_line(finish_line()?)?;
        self.sender.take();

        let chunk_count = self.chunker.chunk_count;
        let audio_bytes = self.chunker.audio_bytes;
        let mut task = self.task;
        let mut outcome = match timeout(FINISH_RESPONSE_TIMEOUT, &mut task).await {
            Ok(joined) => joined.context("realtime asr upload task failed")??,
            Err(_) => {
                task.abort();
                return Err(anyhow!("realtime asr finish response timeout"));
            }
        };
        outcome.chunk_count = chunk_count;
        outcome.audio_bytes = audio_bytes;
        Ok(outcome)
    }

    pub(crate) async fn cancel(mut self, reason: &str) {
        if let Ok(line) = cancel_line(reason) {
            let _ = self.try_send_line(line);
        }
        self.sender.take();
        if timeout(CANCEL_RESPONSE_TIMEOUT, &mut self.task)
            .await
            .is_err()
        {
            self.task.abort();
        }
    }

    fn try_send_line(&self, line: Vec<u8>) -> Result<()> {
        self.sender
            .as_ref()
            .ok_or_else(|| anyhow!("realtime asr upload is closed"))?
            .try_send(line)
            .map_err(|error| anyhow!("realtime asr upload backpressure: {error}"))
    }
}

async fn run_upload(
    http: Client,
    config: RealtimeAsrConfig,
    call_id: String,
    trace_id: String,
    turn_id: String,
    receiver: mpsc::Receiver<Vec<u8>>,
) -> Result<RealtimeAsrOutcome> {
    let started_at = Instant::now();
    let body_stream = stream::unfold(receiver, |mut receiver| async move {
        receiver
            .recv()
            .await
            .map(|chunk| (Ok::<Vec<u8>, io::Error>(chunk), receiver))
    });
    let response = http
        .post(config.url.as_deref().unwrap_or_default())
        .header("Content-Type", "application/x-ndjson")
        .header(
            "X-CV-Runtime-Token",
            config.runtime_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",
            config.runtime_session_nonce.as_deref().unwrap_or_default(),
        )
        .body(reqwest::Body::wrap_stream(body_stream))
        .send()
        .await
        .context("failed to post realtime asr stream")?;
    let status = response.status();
    if !status.is_success() {
        return Err(anyhow!(
            "realtime asr http failed status={} turn={}",
            status.as_u16(),
            turn_id
        ));
    }
    let body: RuntimeCommonResult<RuntimeRealtimeAsrResp> = response
        .json()
        .await
        .context("failed to decode realtime asr response")?;
    if body.code != 0 {
        return Err(anyhow!(
            "realtime asr common result failed code={}",
            body.code
        ));
    }
    let data = body
        .data
        .ok_or_else(|| anyhow!("realtime asr response data missing"))?;
    Ok(RealtimeAsrOutcome {
        status: data.status.unwrap_or_else(|| "unknown".to_string()),
        asr_result_ref: data.asr_result_ref,
        provider_alias: data.provider_alias,
        partial_count: data.partial_count.unwrap_or_default(),
        fallback_reason: data.fallback_reason,
        fallback_stage: data.fallback_stage,
        chunk_count: 0,
        audio_bytes: 0,
        wall_ms: started_at.elapsed().as_millis() as u64,
    })
}

struct Pcm16kChunker {
    pending: Vec<i16>,
    samples_per_chunk: usize,
    next_seq: u64,
    chunk_count: u64,
    audio_bytes: u64,
}

impl Pcm16kChunker {
    fn new(chunk_duration_ms: u64) -> Self {
        Self {
            pending: Vec::new(),
            samples_per_chunk: SAMPLE_RATE_16K as usize * chunk_duration_ms as usize / 1000,
            next_seq: 0,
            chunk_count: 0,
            audio_bytes: 0,
        }
    }

    fn push_48k(&mut self, samples: &[i16]) -> Vec<PcmChunk> {
        self.pending.extend(samples.iter().step_by(3).copied());
        let mut chunks = Vec::new();
        while self.pending.len() >= self.samples_per_chunk {
            let samples = self.pending.drain(..self.samples_per_chunk).collect();
            chunks.push(self.record_chunk(samples));
        }
        chunks
    }

    fn flush(&mut self) -> Option<PcmChunk> {
        if self.pending.is_empty() {
            return None;
        }
        let samples = std::mem::take(&mut self.pending);
        Some(self.record_chunk(samples))
    }

    fn record_chunk(&mut self, samples: Vec<i16>) -> PcmChunk {
        self.next_seq = self.next_seq.saturating_add(1);
        self.chunk_count = self.chunk_count.saturating_add(1);
        self.audio_bytes = self
            .audio_bytes
            .saturating_add((samples.len() * size_of::<i16>()) as u64);
        PcmChunk {
            seq: self.next_seq,
            samples,
        }
    }
}

struct PcmChunk {
    seq: u64,
    samples: Vec<i16>,
}

fn session_start_line(
    call_id: &str,
    trace_id: &str,
    turn_id: &str,
    runtime_session_nonce: &str,
) -> Result<Vec<u8>> {
    encode_line(json!({
        "event": "session_start",
        "callId": call_id,
        "traceId": trace_id,
        "turnId": turn_id,
        "runtimeSessionNonceHash": sha12(runtime_session_nonce),
        "audio": {
            "format": "pcm_s16le",
            "sampleRate": SAMPLE_RATE_16K,
            "channels": CHANNELS_MONO,
        }
    }))
}

fn audio_chunk_line(chunk_seq: u64, samples: &[i16]) -> Result<Vec<u8>> {
    let mut bytes = Vec::with_capacity(samples.len() * size_of::<i16>());
    for sample in samples {
        bytes.extend_from_slice(&sample.to_le_bytes());
    }
    encode_line(json!({
        "event": "audio_chunk",
        "chunkSeq": chunk_seq,
        "audioBase64": general_purpose::STANDARD.encode(bytes),
        "durationMs": ((samples.len() as u64) * 1000 / u64::from(SAMPLE_RATE_16K)).max(1),
    }))
}

fn vad_speech_end_line(speech_duration_ms: u64, end_reason: &str) -> Result<Vec<u8>> {
    encode_line(json!({
        "event": "vad_speech_end",
        "speechDurationMs": speech_duration_ms,
        "endReason": end_reason,
    }))
}

fn finish_line() -> Result<Vec<u8>> {
    encode_line(json!({"event": "finish"}))
}

fn cancel_line(reason: &str) -> Result<Vec<u8>> {
    encode_line(json!({"event": "cancel", "reason": reason}))
}

fn encode_line(value: serde_json::Value) -> Result<Vec<u8>> {
    let mut line = serde_json::to_vec(&value)?;
    line.push(b'\n');
    if line.len() > MAX_NDJSON_LINE_BYTES {
        return Err(anyhow!("realtime asr ndjson line exceeds 24KiB"));
    }
    Ok(line)
}

fn sha12(value: &str) -> String {
    let digest = Sha256::digest(value.as_bytes());
    format!("{digest:x}")[..12].to_string()
}

fn non_blank(value: &Option<String>) -> bool {
    value.as_ref().is_some_and(|value| !value.trim().is_empty())
}

#[derive(Deserialize)]
struct RuntimeCommonResult<T> {
    code: i64,
    data: Option<T>,
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct RuntimeRealtimeAsrResp {
    status: Option<String>,
    asr_result_ref: Option<String>,
    provider_alias: Option<String>,
    partial_count: Option<u64>,
    fallback_reason: Option<String>,
    fallback_stage: Option<String>,
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::{
        io::{Read, Write},
        net::TcpListener,
        sync::mpsc as std_mpsc,
        thread,
    };

    #[test]
    fn chunker_aggregates_20ms_frames_into_200ms_chunks_and_flushes_tail() {
        let mut chunker = Pcm16kChunker::new(200);
        let first = chunker.push_48k(&vec![7; 9_600]);
        let second = chunker.push_48k(&vec![9; 4_800]);
        let tail = chunker.flush();

        assert_eq!(1, first.len());
        assert_eq!(1, first[0].seq);
        assert_eq!(3_200, first[0].samples.len());
        assert!(second.is_empty());
        assert_eq!(2, tail.as_ref().map(|chunk| chunk.seq).unwrap_or_default());
        assert_eq!(
            1_600,
            tail.as_ref()
                .map(|chunk| chunk.samples.len())
                .unwrap_or_default()
        );
        assert_eq!(2, chunker.chunk_count);
        assert_eq!(9_600, chunker.audio_bytes);
    }

    #[test]
    fn session_start_uses_canonical_nonce_hash_and_audio_contract() {
        let line = session_start_line("call-001", "trace-001", "turn-0001", "nonce-001")
            .expect("session start line");
        let value: serde_json::Value = serde_json::from_slice(&line).expect("valid json");

        assert_eq!("session_start", value["event"]);
        assert_eq!(sha12("nonce-001"), value["runtimeSessionNonceHash"]);
        assert_eq!("pcm_s16le", value["audio"]["format"]);
        assert_eq!(16000, value["audio"]["sampleRate"]);
        assert_eq!(1, value["audio"]["channels"]);
    }

    #[test]
    fn maximum_audio_chunk_stays_within_java_line_limit() {
        let line = audio_chunk_line(1, &vec![0; 8_000]).expect("maximum chunk line");
        assert!(line.len() <= MAX_NDJSON_LINE_BYTES);
    }

    #[tokio::test]
    async fn upload_streams_canonical_ndjson_and_reads_final_response() {
        let response = json!({
            "code": 0,
            "data": {
                "status": "final",
                "asrResultRef": "asr_rt_fixture",
                "providerAlias": "fixture",
                "partialCount": 2
            }
        })
        .to_string();
        let (url, captured, server) = spawn_http_fixture(response);
        let mut upload = RealtimeAsrUpload::start(
            Client::new(),
            fixture_config(url),
            "call-001",
            "trace-001",
            "turn-0001",
            &vec![1; 9_600],
        )
        .expect("start upload");
        upload
            .push_48k_samples(&vec![2; 9_600])
            .expect("push audio");

        let outcome = upload.finish(400, "silence").await.expect("finish upload");
        let request = captured.recv().expect("captured request");
        server.join().expect("fixture server");

        assert_eq!("final", outcome.status);
        assert_eq!(Some("asr_rt_fixture"), outcome.asr_result_ref.as_deref());
        assert_eq!(2, outcome.chunk_count);
        assert_eq!(12_800, outcome.audio_bytes);
        assert!(
            request
                .headers
                .contains("x-cv-runtime-session-nonce: nonce-001")
        );
        let events = request
            .body
            .lines()
            .map(|line| serde_json::from_str::<serde_json::Value>(line).expect("event json"))
            .collect::<Vec<_>>();
        assert_eq!(5, events.len());
        assert_eq!("session_start", events[0]["event"]);
        assert_eq!(1, events[1]["chunkSeq"]);
        assert_eq!(2, events[2]["chunkSeq"]);
        assert_eq!("vad_speech_end", events[3]["event"]);
        assert_eq!("finish", events[4]["event"]);
    }

    #[tokio::test]
    async fn cancel_ends_stream_without_finish_event() {
        let response = json!({"code": 0, "data": {"status": "cancelled"}}).to_string();
        let (url, captured, server) = spawn_http_fixture(response);
        let upload = RealtimeAsrUpload::start(
            Client::new(),
            fixture_config(url),
            "call-002",
            "trace-002",
            "turn-0002",
            &vec![1; 9_600],
        )
        .expect("start upload");

        upload.cancel("call_end").await;
        let request = captured.recv().expect("captured request");
        server.join().expect("fixture server");

        assert!(request.body.contains("\"event\":\"cancel\""));
        assert!(!request.body.contains("\"event\":\"finish\""));
    }

    fn fixture_config(url: String) -> RealtimeAsrConfig {
        RealtimeAsrConfig {
            enabled: true,
            url: Some(url),
            runtime_token: Some("token-001".to_string()),
            runtime_session_nonce: Some("nonce-001".to_string()),
            chunk_duration_ms: 200,
        }
    }

    struct CapturedRequest {
        headers: String,
        body: String,
    }

    fn spawn_http_fixture(
        response_body: String,
    ) -> (
        String,
        std_mpsc::Receiver<CapturedRequest>,
        thread::JoinHandle<()>,
    ) {
        let listener = TcpListener::bind("127.0.0.1:0").expect("bind fixture server");
        let address = listener.local_addr().expect("fixture address");
        let (sender, receiver) = std_mpsc::channel();
        let server = thread::spawn(move || {
            let (mut stream, _) = listener.accept().expect("accept request");
            stream
                .set_read_timeout(Some(Duration::from_secs(3)))
                .expect("read timeout");
            let mut request = Vec::new();
            let mut buffer = [0u8; 4096];
            loop {
                let read = stream.read(&mut buffer).expect("read request");
                if read == 0 {
                    break;
                }
                request.extend_from_slice(&buffer[..read]);
                if request.windows(5).any(|window| window == b"0\r\n\r\n") {
                    break;
                }
            }
            let header_end = request
                .windows(4)
                .position(|window| window == b"\r\n\r\n")
                .expect("request headers");
            let headers = String::from_utf8_lossy(&request[..header_end]).to_ascii_lowercase();
            let body = decode_chunked_body(&request[header_end + 4..]);
            sender
                .send(CapturedRequest {
                    headers,
                    body: String::from_utf8(body).expect("utf8 ndjson"),
                })
                .expect("capture request");
            let response = format!(
                "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                response_body.len(),
                response_body
            );
            stream
                .write_all(response.as_bytes())
                .expect("write response");
        });
        (
            format!("http://{address}/runtime/asr/realtime"),
            receiver,
            server,
        )
    }

    fn decode_chunked_body(mut input: &[u8]) -> Vec<u8> {
        let mut body = Vec::new();
        loop {
            let line_end = input
                .windows(2)
                .position(|window| window == b"\r\n")
                .expect("chunk size line");
            let size = usize::from_str_radix(
                std::str::from_utf8(&input[..line_end]).expect("chunk size utf8"),
                16,
            )
            .expect("chunk size");
            input = &input[line_end + 2..];
            if size == 0 {
                break;
            }
            body.extend_from_slice(&input[..size]);
            input = &input[size + 2..];
        }
        body
    }
}

 src/audio.rs

@@ -1012,10 +1012,7 @@
        assert_eq!(5, frames.len());
        assert_eq!(
            4_800,
            frames
                .iter()
                .map(|frame| frame.samples.len())
                .sum::<usize>()
            frames.iter().map(|frame| frame.data.len()).sum::<usize>()
        );
        assert!(frames.iter().all(|frame| frame.sample_rate == 48_000));
        assert!(frames.iter().all(|frame| frame.num_channels == 1));
@@ -1066,10 +1063,7 @@
        assert!(frames.iter().all(|frame| frame.samples_per_channel == 320));
        assert_eq!(
            1_600,
            frames
                .iter()
                .map(|frame| frame.samples.len())
                .sum::<usize>()
            frames.iter().map(|frame| frame.data.len()).sum::<usize>()
        );
    }
}

 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)
                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,

 src/service.rs

@@ -410,6 +410,18 @@
        if let Some(value) = &runtime.asr_stream_url {
            command.env("CV_RUNTIME_ASR_STREAM_URL", value);
        }
        if let Some(value) = runtime.asr_realtime_enabled {
            command.env("CV_RUNTIME_ASR_REALTIME_ENABLED", value.to_string());
        }
        if let Some(value) = &runtime.asr_realtime_url {
            command.env("CV_RUNTIME_ASR_REALTIME_URL", value);
        }
        if let Some(value) = runtime.asr_realtime_chunk_duration_ms {
            command.env(
                "CV_RUNTIME_ASR_REALTIME_CHUNK_DURATION_MS",
                value.to_string(),
            );
        }
        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);
@@ -976,6 +988,9 @@
    audio_debug_dump_enabled: Option<bool>,
    asr_streaming_enabled: Option<bool>,
    asr_stream_url: Option<String>,
    asr_realtime_enabled: Option<bool>,
    asr_realtime_url: Option<String>,
    asr_realtime_chunk_duration_ms: Option<u64>,
}

#[derive(Default, Deserialize)]

 tools/validate-asr-realtime-fixture.mjs

New file
@@ -0,0 +1,116 @@
#!/usr/bin/env node

import fs from 'node:fs'
import path from 'node:path'

const file = process.argv[2] || 'fixtures/asr-realtime-happy.ndjson'
const absolute = path.resolve(file)
const raw = fs.readFileSync(absolute, 'utf8')
const lines = raw.split(/\r?\n/).map((line) => line.trim()).filter(Boolean)
const allowedEvents = new Set(['session_start', 'audio_chunk', 'vad_speech_end', 'finish', 'cancel'])
const forbiddenActivityEvents = new Set([
  'asr_realtime_session_started',
  'asr_realtime_audio_chunk',
  'asr_realtime_speech_end',
  'asr_realtime_finish_received',
  'asr_realtime_cancelled',
])

let state = 'new'
let lastChunkSeq = 0
let audioChunkCount = 0
let audioBytes = 0

function fail(message) {
  console.error(`asr realtime fixture invalid: ${message}`)
  process.exit(1)
}

for (const [index, line] of lines.entries()) {
  if (Buffer.byteLength(`${line}\n`, 'utf8') > 24 * 1024) {
    fail(`line ${index + 1} exceeds 24KiB`)
  }
  let event
  try {
    event = JSON.parse(line)
  } catch {
    fail(`line ${index + 1} is not JSON`)
  }
  if (forbiddenActivityEvents.has(event.event)) {
    fail(`line ${index + 1} uses activity event ${event.event} as a wire event`)
  }
  if (!allowedEvents.has(event.event)) {
    fail(`line ${index + 1} has unsupported event ${event.event}`)
  }

  if (state === 'new') {
    if (event.event !== 'session_start') {
      fail('first event must be session_start')
    }
    if (!event.callId || !event.traceId || !event.turnId || !event.runtimeSessionNonceHash) {
      fail('session_start missing binding fields')
    }
    if (event.audio?.format !== 'pcm_s16le' || event.audio?.sampleRate !== 16000 || event.audio?.channels !== 1) {
      fail('session_start audio must be pcm_s16le/16000Hz/mono')
    }
    state = 'streaming'
    continue
  }

  if (state === 'terminal') {
    fail(`event ${event.event} arrived after terminal event`)
  }
  if (event.event === 'audio_chunk') {
    if (state !== 'streaming') {
      fail('audio_chunk arrived after vad_speech_end')
    }
    if (!Number.isInteger(event.chunkSeq) || event.chunkSeq !== lastChunkSeq + 1) {
      fail('audio_chunk chunkSeq must start at 1 and be contiguous')
    }
    if (!Number.isInteger(event.durationMs) || event.durationMs < 1 || event.durationMs > 1000) {
      fail('audio_chunk durationMs must be 1..1000')
    }
    const payload = Buffer.from(event.audioBase64 || '', 'base64')
    if (payload.length === 0 || payload.length > 16000 || payload.length % 2 !== 0) {
      fail('audio_chunk payload must be non-empty, <=16000 bytes and 16-bit aligned')
    }
    lastChunkSeq = event.chunkSeq
    audioChunkCount += 1
    audioBytes += payload.length
    continue
  }
  if (event.event === 'vad_speech_end') {
    if (state !== 'streaming' || audioChunkCount === 0) {
      fail('vad_speech_end requires at least one audio_chunk')
    }
    state = 'speech_ended'
    continue
  }
  if (event.event === 'finish') {
    if (state !== 'speech_ended') {
      fail('finish must follow vad_speech_end')
    }
    state = 'terminal'
    continue
  }
  if (event.event === 'cancel') {
    if (!event.reason) {
      fail('cancel missing reason')
    }
    state = 'terminal'
    continue
  }
  fail(`duplicate session_start at line ${index + 1}`)
}

if (state !== 'terminal') {
  fail('fixture must end with finish or cancel')
}

console.log(JSON.stringify({
  ok: true,
  file: path.relative(process.cwd(), absolute),
  eventCount: lines.length,
  audioChunkCount,
  audioBytes,
}))