feat: bind helper stream timing markers

Ariver

2026-07-12 b0a4e2ea93fafc4d65d474f3ada616a0ad5e19d5

 src/main.rs

@@ -1,7 +1,10 @@
mod asr_realtime;
mod audio;
mod service;

use std::{
    borrow::Cow,
    collections::HashSet,
    env, fs,
    path::{Path, PathBuf},
    sync::{
@@ -12,6 +15,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;
@@ -30,17 +34,29 @@
use reqwest::Client;
use serde::{Deserialize, Serialize};
use serde_json::json;
use tokio::time::{sleep, sleep_until};
use tokio::{sync::mpsc::UnboundedReceiver, task::JoinHandle};
use sha2::{Digest, Sha256};
use tokio::time::{sleep, sleep_until, timeout};
use tokio::{
    sync::{mpsc, mpsc::UnboundedReceiver, watch},
    task::JoinHandle,
};
use tracing::{info, warn};

const USER_AUDIO_SAMPLE_RATE_HZ: u32 = 48_000;
const USER_AUDIO_NUM_CHANNELS: u16 = 1;
const INBOUND_AUDIO_QUEUE_CAPACITY: usize = 100;
const INBOUND_AUDIO_DROP_LOG_INTERVAL: u64 = 100;
const AUDIO_DRAIN_STOP_GRACE: Duration = Duration::from_secs(2);
const DEFAULT_BOT_AUDIO_PROFILE: &str = "pcm-16k";
const LIVEKIT_48K_SAMPLE_RATE_HZ: u32 = 48_000;
const PCM_16K_SAMPLE_RATE_HZ: u32 = 16_000;
const BOT_NUM_CHANNELS: u16 = 1;
const TRACK_NAME: &str = "bot-main-audio";
const STREAM_TIMING_VERSION: u32 = 1;
const STREAM_TIMING_FIRST_SEGMENT: u64 = 1;
const STREAM_TIMING_FIRST_CHUNK: u64 = 1;
const STREAM_TIMING_MAX_ELAPSED_MS: u64 = 5_000;
const STREAM_TIMING_SOURCE: &str = "stream_anchor_monotonic";

#[tokio::main(flavor = "multi_thread")]
async fn main() -> Result<()> {
@@ -272,6 +288,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 +355,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 +371,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 +421,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 +454,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),
@@ -629,6 +670,31 @@
    println!("{payload}");
}

fn emit_anchored_activity(
    call_id: &str,
    trace_id: &str,
    turn_id: &str,
    event_name: &str,
    marker: &RuntimeTurnStreamTimingMarker,
    extension: serde_json::Value,
) {
    let payload = json!({
        "type": "cv_activity",
        "callId": call_id,
        "traceId": trace_id,
        "turnId": turn_id,
        "eventName": event_name,
        "eventWallTimeMs": current_time_millis(),
        "serverDeltaMs": marker.server_delta_ms,
        "serverDeltaSource": STREAM_TIMING_SOURCE,
        "result": "ok",
        "reasonCode": null,
        "retryable": null,
        "extension": marker.extension_with(extension),
    });
    println!("{payload}");
}

fn spawn_user_audio_observer(
    events: UnboundedReceiver<RoomEvent>,
    config: &Config,
@@ -775,6 +841,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 +901,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,
@@ -1382,6 +1451,7 @@
            .await?;
        }
    }
    state.close_timing();
    if !state.completed {
        warn!(
            call_id = %call_id,
@@ -1511,6 +1581,7 @@
                call_id,
                trace_id,
                turn,
                &event,
                audio_chunk,
                state,
                turn_pipeline_started_at,
@@ -1529,6 +1600,7 @@
        }
        Some("turn_completed") => {
            state.completed = true;
            state.close_timing();
            info!(
                call_id = %call_id,
                trace_id = %trace_id,
@@ -1539,6 +1611,7 @@
            );
        }
        Some("turn_failed") => {
            state.close_timing();
            let error = event.error.as_ref();
            let reason_code = error
                .and_then(|value| value.reason_code.as_deref())
@@ -1573,6 +1646,7 @@
        }
        Some("turn_cancelled") => {
            state.completed = true;
            state.close_timing();
            info!(
                call_id = %call_id,
                trace_id = %trace_id,
@@ -1605,6 +1679,19 @@
        }
        Some("activity") => {
            if let Some(activity) = event.activity.as_ref() {
                if activity.event_type.as_deref() == Some("tts_first_audio_chunk_ready") {
                    if let Some(runtime_session_nonce) =
                        bridge_config.runtime_session_nonce.as_deref()
                    {
                        let _ = state.arm_timing_anchor(
                            call_id,
                            trace_id,
                            &turn.turn_id,
                            runtime_session_nonce,
                            &event,
                        );
                    }
                }
                info!(
                    call_id = %call_id,
                    trace_id = %trace_id,
@@ -1663,6 +1750,7 @@
    call_id: &str,
    trace_id: &str,
    turn: &FinishedSpeechTurn,
    event: &RuntimeTurnStreamEvent,
    audio_chunk: &RuntimeTurnStreamAudioChunk,
    state: &mut RuntimeTurnStreamState,
    turn_pipeline_started_at: Instant,
@@ -1682,6 +1770,58 @@
        .unwrap_or("pcm_s16le")
        .trim()
        .to_ascii_lowercase();
    if !matches!(format.as_str(), "pcm_s16le" | "mp3" | "mpeg" | "wav") {
        return Err(anyhow!("unsupported reply_audio_chunk format {format}"));
    }
    match state.reply_chunk_markers.observe(audio_chunk.segment_seq) {
        ReplyChunkMarker::FirstReply => {
            let extension = json!({
                "segmentSeq": audio_chunk.segment_seq,
                "chunkSeq": audio_chunk.chunk_seq,
                "format": format.as_str(),
                "bytes": payload.len(),
            });
            if let Some(marker) =
                state.record_m6(call_id, trace_id, &turn.turn_id, event, audio_chunk)
            {
                emit_anchored_activity(
                    call_id,
                    trace_id,
                    &turn.turn_id,
                    "helper_first_reply_audio_chunk_received",
                    &marker,
                    extension,
                );
            } else {
                emit_activity(
                    call_id,
                    trace_id,
                    Some(&turn.turn_id),
                    "helper_first_reply_audio_chunk_received",
                    "ok",
                    None,
                    None,
                    extension,
                );
            }
        }
        ReplyChunkMarker::SegmentFirst => emit_activity(
            call_id,
            trace_id,
            Some(&turn.turn_id),
            "helper_segment_first_audio_chunk_received",
            "ok",
            None,
            None,
            json!({
                "segmentSeq": audio_chunk.segment_seq,
                "chunkSeq": audio_chunk.chunk_seq,
                "format": format.as_str(),
                "bytes": payload.len(),
            }),
        ),
        ReplyChunkMarker::None => {}
    }
    let frames = if format == "pcm_s16le" {
        let sample_rate = audio_chunk.sample_rate.unwrap_or(sink.sample_rate_hz);
        let channels = audio_chunk.channels.unwrap_or(u32::from(sink.num_channels));
@@ -1773,7 +1913,7 @@
            Err(error) => return Err(error).context("failed to decode final stream audio chunk"),
        }
    } else {
        return Err(anyhow!("unsupported reply_audio_chunk format {format}"));
        unreachable!("supported encoded format checked above")
    };
    if frames.is_empty() {
        return Ok(0);
@@ -1801,21 +1941,33 @@
        sink.write_pcm_frame(frame).await?;
        if !state.first_audio_frame_written {
            state.first_audio_frame_written = true;
            emit_activity(
                call_id,
                trace_id,
                Some(&turn.turn_id),
                "bot_reply_first_audio_frame_written",
                "ok",
                None,
                None,
                json!({
                    "replyPlaybackMode": state.reply_playback_mode.as_str(),
                    "format": format.as_str(),
                    "chunkSeq": audio_chunk.chunk_seq,
                    "replyTotalAfterVadEndMs": turn_pipeline_started_at.elapsed().as_millis() as u64,
                }),
            );
            let extension = json!({
                "replyPlaybackMode": state.reply_playback_mode.as_str(),
                "format": format.as_str(),
                "chunkSeq": audio_chunk.chunk_seq,
                "replyTotalAfterVadEndMs": turn_pipeline_started_at.elapsed().as_millis() as u64,
            });
            if let Some(marker) = state.record_m7() {
                emit_anchored_activity(
                    call_id,
                    trace_id,
                    &turn.turn_id,
                    "bot_reply_first_audio_frame_written",
                    &marker,
                    extension,
                );
            } else {
                emit_activity(
                    call_id,
                    trace_id,
                    Some(&turn.turn_id),
                    "bot_reply_first_audio_frame_written",
                    "ok",
                    None,
                    None,
                    extension,
                );
            }
        }
        sleep_until(pacing_started_at + Duration::from_millis(((index + 1) as u64) * 20)).await;
    }
@@ -2272,6 +2424,8 @@
    encoded_audio_buffer: Vec<u8>,
    pcm_stream_decoder: Option<audio::PcmS16leStreamDecoder>,
    pcm_stream_network_chunk_count: u64,
    reply_chunk_markers: ReplyChunkMarkerState,
    timing: RuntimeTurnStreamTimingState,
}

impl Default for RuntimeTurnStreamState {
@@ -2287,7 +2441,262 @@
            encoded_audio_buffer: Vec::new(),
            pcm_stream_decoder: None,
            pcm_stream_network_chunk_count: 0,
            reply_chunk_markers: ReplyChunkMarkerState::default(),
            timing: RuntimeTurnStreamTimingState::default(),
        }
    }
}

impl RuntimeTurnStreamState {
    fn arm_timing_anchor(
        &mut self,
        call_id: &str,
        trace_id: &str,
        turn_id: &str,
        runtime_session_nonce: &str,
        event: &RuntimeTurnStreamEvent,
    ) -> bool {
        if self.timing.phase != RuntimeTurnStreamTimingPhase::Empty
            || event.call_id.as_deref() != Some(call_id)
            || event.trace_id.as_deref() != Some(trace_id)
            || event.turn_id.as_deref() != Some(turn_id)
        {
            return false;
        }
        let Some(activity) = event.activity.as_ref() else {
            return false;
        };
        if activity.event_type.as_deref() != Some("tts_first_audio_chunk_ready") {
            return false;
        }
        let Some(extension) = activity.extension.as_ref() else {
            return false;
        };
        let Some(anchor_id) = extension.stream_anchor_id.as_deref() else {
            return false;
        };
        let valid_anchor_id = (16..=64).contains(&anchor_id.len()) && anchor_id.is_ascii();
        let expected_nonce_hash = runtime_session_nonce_hash(runtime_session_nonce);
        if extension.stream_timing_version != Some(STREAM_TIMING_VERSION)
            || !valid_anchor_id
            || extension.runtime_session_nonce_hash.as_deref() != Some(expected_nonce_hash.as_str())
            || extension.segment_seq != Some(STREAM_TIMING_FIRST_SEGMENT)
            || extension.stream_timing_validation.as_deref() != Some("bound")
        {
            return false;
        }
        let Some(server_delta_ms) = extension.stream_anchor_server_delta_ms else {
            return false;
        };
        self.timing.anchor = Some(RuntimeTurnStreamTimingAnchor {
            call_id: call_id.to_string(),
            trace_id: trace_id.to_string(),
            turn_id: turn_id.to_string(),
            anchor_id: anchor_id.to_string(),
            server_delta_ms,
            runtime_session_nonce_hash: expected_nonce_hash,
            segment_seq: STREAM_TIMING_FIRST_SEGMENT,
            received_at: Instant::now(),
        });
        self.timing.phase = RuntimeTurnStreamTimingPhase::Armed;
        true
    }

    fn record_m6(
        &mut self,
        call_id: &str,
        trace_id: &str,
        turn_id: &str,
        event: &RuntimeTurnStreamEvent,
        audio_chunk: &RuntimeTurnStreamAudioChunk,
    ) -> Option<RuntimeTurnStreamTimingMarker> {
        if self.timing.phase != RuntimeTurnStreamTimingPhase::Armed {
            return None;
        }
        let anchor = self.timing.anchor.as_ref()?;
        if anchor.call_id != call_id
            || anchor.trace_id != trace_id
            || anchor.turn_id != turn_id
            || event.call_id.as_deref() != Some(call_id)
            || event.trace_id.as_deref() != Some(trace_id)
            || event.turn_id.as_deref() != Some(turn_id)
            || audio_chunk.segment_seq != Some(anchor.segment_seq)
            || audio_chunk.chunk_seq != Some(STREAM_TIMING_FIRST_CHUNK)
            || audio_chunk.stream_timing_version != Some(STREAM_TIMING_VERSION)
            || audio_chunk.stream_anchor_id.as_deref() != Some(anchor.anchor_id.as_str())
        {
            return None;
        }
        let Some(marker) = RuntimeTurnStreamTimingMarker::from_anchor(
            anchor,
            audio_chunk.chunk_seq.unwrap_or(STREAM_TIMING_FIRST_CHUNK),
        ) else {
            self.close_timing();
            return None;
        };
        self.timing.chunk_seq = Some(marker.chunk_seq);
        self.timing.phase = RuntimeTurnStreamTimingPhase::M6Recorded;
        Some(marker)
    }

    fn record_m7(&mut self) -> Option<RuntimeTurnStreamTimingMarker> {
        if self.timing.phase != RuntimeTurnStreamTimingPhase::M6Recorded {
            return None;
        }
        let anchor = self.timing.anchor.as_ref()?;
        let Some(marker) = RuntimeTurnStreamTimingMarker::from_anchor(
            anchor,
            self.timing.chunk_seq.unwrap_or(STREAM_TIMING_FIRST_CHUNK),
        ) else {
            self.close_timing();
            return None;
        };
        self.timing.phase = RuntimeTurnStreamTimingPhase::M7Recorded;
        Some(marker)
    }

    fn close_timing(&mut self) {
        self.timing.close();
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum RuntimeTurnStreamTimingPhase {
    Empty,
    Armed,
    M6Recorded,
    M7Recorded,
    Closed,
}

struct RuntimeTurnStreamTimingState {
    phase: RuntimeTurnStreamTimingPhase,
    anchor: Option<RuntimeTurnStreamTimingAnchor>,
    chunk_seq: Option<u64>,
}

impl Default for RuntimeTurnStreamTimingState {
    fn default() -> Self {
        Self {
            phase: RuntimeTurnStreamTimingPhase::Empty,
            anchor: None,
            chunk_seq: None,
        }
    }
}

impl RuntimeTurnStreamTimingState {
    fn close(&mut self) {
        self.anchor = None;
        self.chunk_seq = None;
        self.phase = RuntimeTurnStreamTimingPhase::Closed;
    }
}

impl Drop for RuntimeTurnStreamTimingState {
    fn drop(&mut self) {
        self.anchor = None;
        self.chunk_seq = None;
    }
}

struct RuntimeTurnStreamTimingAnchor {
    call_id: String,
    trace_id: String,
    turn_id: String,
    anchor_id: String,
    server_delta_ms: u64,
    runtime_session_nonce_hash: String,
    segment_seq: u64,
    received_at: Instant,
}

struct RuntimeTurnStreamTimingMarker {
    version: u32,
    anchor_id: String,
    anchor_server_delta_ms: u64,
    anchor_elapsed_ms: u64,
    server_delta_ms: u64,
    runtime_session_nonce_hash: String,
    segment_seq: u64,
    chunk_seq: u64,
}

impl RuntimeTurnStreamTimingMarker {
    fn from_anchor(anchor: &RuntimeTurnStreamTimingAnchor, chunk_seq: u64) -> Option<Self> {
        let elapsed_ms = u64::try_from(anchor.received_at.elapsed().as_millis()).ok()?;
        if elapsed_ms > STREAM_TIMING_MAX_ELAPSED_MS {
            return None;
        }
        Some(Self {
            version: STREAM_TIMING_VERSION,
            anchor_id: anchor.anchor_id.clone(),
            anchor_server_delta_ms: anchor.server_delta_ms,
            anchor_elapsed_ms: elapsed_ms,
            server_delta_ms: anchor.server_delta_ms.checked_add(elapsed_ms)?,
            runtime_session_nonce_hash: anchor.runtime_session_nonce_hash.clone(),
            segment_seq: anchor.segment_seq,
            chunk_seq,
        })
    }

    fn extension_with(&self, extra: serde_json::Value) -> serde_json::Value {
        let mut extension = match extra {
            serde_json::Value::Object(value) => value,
            _ => serde_json::Map::new(),
        };
        extension.insert("streamTimingVersion".to_string(), json!(self.version));
        extension.insert("streamAnchorId".to_string(), json!(self.anchor_id));
        extension.insert(
            "streamAnchorServerDeltaMs".to_string(),
            json!(self.anchor_server_delta_ms),
        );
        extension.insert("anchorElapsedMs".to_string(), json!(self.anchor_elapsed_ms));
        extension.insert(
            "runtimeSessionNonceHash".to_string(),
            json!(self.runtime_session_nonce_hash),
        );
        extension.insert("segmentSeq".to_string(), json!(self.segment_seq));
        extension.insert("chunkSeq".to_string(), json!(self.chunk_seq));
        extension.insert("streamTimingValidation".to_string(), json!("bound"));
        serde_json::Value::Object(extension)
    }
}

fn runtime_session_nonce_hash(value: &str) -> String {
    let digest = Sha256::digest(value.as_bytes());
    digest[..6]
        .iter()
        .map(|byte| format!("{byte:02x}"))
        .collect()
}

#[derive(Debug, PartialEq, Eq)]
enum ReplyChunkMarker {
    FirstReply,
    SegmentFirst,
    None,
}

#[derive(Default)]
struct ReplyChunkMarkerState {
    first_reply_seen: bool,
    seen_segments: HashSet<u64>,
}

impl ReplyChunkMarkerState {
    fn observe(&mut self, segment_seq: Option<u64>) -> ReplyChunkMarker {
        let first_for_segment = segment_seq
            .map(|value| self.seen_segments.insert(value))
            .unwrap_or(false);
        if !self.first_reply_seen {
            self.first_reply_seen = true;
            return ReplyChunkMarker::FirstReply;
        }
        if first_for_segment {
            return ReplyChunkMarker::SegmentFirst;
        }
        ReplyChunkMarker::None
    }
}

@@ -2307,6 +2716,12 @@
struct RuntimeTurnStreamEvent {
    #[serde(rename = "type", alias = "event")]
    event_type: Option<String>,
    #[serde(rename = "callId")]
    call_id: Option<String>,
    #[serde(rename = "traceId")]
    trace_id: Option<String>,
    #[serde(rename = "turnId")]
    turn_id: Option<String>,
    #[serde(rename = "seq")]
    seq: Option<u64>,
    #[serde(rename = "replyPlaybackMode")]
@@ -2342,6 +2757,12 @@
struct RuntimeTurnStreamAudioChunk {
    #[serde(rename = "chunkSeq", alias = "seq")]
    chunk_seq: Option<u64>,
    #[serde(rename = "segmentSeq")]
    segment_seq: Option<u64>,
    #[serde(rename = "streamTimingVersion")]
    stream_timing_version: Option<u32>,
    #[serde(rename = "streamAnchorId")]
    stream_anchor_id: Option<String>,
    format: Option<String>,
    #[serde(rename = "sampleRate")]
    sample_rate: Option<u32>,
@@ -2358,6 +2779,23 @@
    stage: Option<String>,
    #[serde(rename = "reasonCode")]
    reason_code: Option<String>,
    extension: Option<RuntimeTurnStreamTimingExtension>,
}

#[derive(Clone, Deserialize)]
struct RuntimeTurnStreamTimingExtension {
    #[serde(rename = "streamTimingVersion")]
    stream_timing_version: Option<u32>,
    #[serde(rename = "streamAnchorId")]
    stream_anchor_id: Option<String>,
    #[serde(rename = "streamAnchorServerDeltaMs")]
    stream_anchor_server_delta_ms: Option<u64>,
    #[serde(rename = "runtimeSessionNonceHash")]
    runtime_session_nonce_hash: Option<String>,
    #[serde(rename = "segmentSeq")]
    segment_seq: Option<u64>,
    #[serde(rename = "streamTimingValidation")]
    stream_timing_validation: Option<String>,
}

#[derive(Deserialize)]
@@ -2439,6 +2877,68 @@
            i32::from(USER_AUDIO_NUM_CHANNELS),
        );
        let started_at = Instant::now();
        let (frame_tx, mut frame_rx) =
            mpsc::channel::<DrainedUserAudioFrame>(INBOUND_AUDIO_QUEUE_CAPACITY);
        let (drain_shutdown_tx, mut drain_shutdown_rx) = watch::channel(false);
        let drain_call_id = call_id.clone();
        let drain_trace_id = trace_id.clone();
        let drain_task = tokio::spawn(async move {
            let mut received_frame_count: u64 = 0;
            let mut dropped_frame_count: u64 = 0;
            loop {
                tokio::select! {
                    changed = drain_shutdown_rx.changed() => {
                        match changed {
                            Ok(()) if *drain_shutdown_rx.borrow() => break,
                            Ok(()) => {}
                            Err(_) => break,
                        }
                    }
                    maybe_frame = stream.next() => {
                        let Some(frame) = maybe_frame else {
                            break;
                        };
                        received_frame_count = received_frame_count.saturating_add(1);
                        let drained = DrainedUserAudioFrame {
                            frame_index: received_frame_count,
                            captured_elapsed_ms: started_at.elapsed().as_millis() as u64,
                            frame: AudioFrame {
                                data: Cow::Owned(frame.data.as_ref().to_vec()),
                                sample_rate: frame.sample_rate,
                                num_channels: frame.num_channels,
                                samples_per_channel: frame.samples_per_channel,
                            },
                        };
                        match frame_tx.try_send(drained) {
                            Ok(()) => {}
                            Err(mpsc::error::TrySendError::Full(_)) => {
                                dropped_frame_count = dropped_frame_count.saturating_add(1);
                                if dropped_frame_count % INBOUND_AUDIO_DROP_LOG_INTERVAL == 1 {
                                    warn!(
                                        call_id = %drain_call_id,
                                        trace_id = %drain_trace_id,
                                        dropped_frame_count,
                                        received_frame_count,
                                        queue_capacity = INBOUND_AUDIO_QUEUE_CAPACITY,
                                        "runtime helper inbound_audio_queue_full_dropping_newest"
                                    );
                                }
                            }
                            Err(mpsc::error::TrySendError::Closed(_)) => break,
                        }
                    }
                }
            }
            stream.close();
            info!(
                call_id = %drain_call_id,
                trace_id = %drain_trace_id,
                received_frame_count,
                dropped_frame_count,
                queue_capacity = INBOUND_AUDIO_QUEUE_CAPACITY,
                "runtime helper user_audio_drain_ended"
            );
        });
        let mut frame_count: u64 = 0;
        let mut sample_count: u64 = 0;
        let mut simple_vad = if simple_vad_enabled {
@@ -2446,11 +2946,13 @@
        } else {
            None
        };
        let mut realtime_asr_upload: Option<RealtimeAsrUpload> = None;

        while let Some(frame) = stream.next().await {
            frame_count += 1;
        while let Some(drained) = frame_rx.recv().await {
            let frame = drained.frame;
            frame_count = drained.frame_index;
            sample_count += u64::from(frame.samples_per_channel) * u64::from(frame.num_channels);
            let elapsed_ms = started_at.elapsed().as_millis() as u64;
            let elapsed_ms = drained.captured_elapsed_ms;
            if frame_count == 1 {
                info!(
                    call_id = %call_id,
@@ -2478,7 +2980,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 +2989,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 +3055,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 +3091,41 @@
                &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;
        }
        let _ = drain_shutdown_tx.send(true);
        let mut drain_task = drain_task;
        if timeout(AUDIO_DRAIN_STOP_GRACE, &mut drain_task)
            .await
            .is_err()
        {
            drain_task.abort();
            let _ = drain_task.await;
            warn!(
                call_id = %call_id,
                trace_id = %trace_id,
                stop_grace_ms = AUDIO_DRAIN_STOP_GRACE.as_millis() as u64,
                "runtime helper user_audio_drain_stop_timeout"
            );
        }

        info!(
@@ -2534,6 +3139,64 @@
            "runtime helper user_audio_stream_ended"
        );
    })
}

struct DrainedUserAudioFrame {
    frame_index: u64,
    captured_elapsed_ms: u64,
    frame: AudioFrame<'static>,
}

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)]
@@ -3227,3 +3890,107 @@
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::{
        ReplyChunkMarker, ReplyChunkMarkerState, RuntimeTurnStreamEvent, RuntimeTurnStreamState,
        RuntimeTurnStreamTimingPhase, runtime_session_nonce_hash,
    };

    #[test]
    fn reply_chunk_marker_state_emits_turn_first_once_and_later_segment_first_once() {
        let mut state = ReplyChunkMarkerState::default();

        assert_eq!(ReplyChunkMarker::FirstReply, state.observe(Some(1)));
        assert_eq!(ReplyChunkMarker::None, state.observe(Some(1)));
        assert_eq!(ReplyChunkMarker::SegmentFirst, state.observe(Some(2)));
        assert_eq!(ReplyChunkMarker::None, state.observe(Some(2)));
        assert_eq!(ReplyChunkMarker::SegmentFirst, state.observe(Some(3)));
    }

    #[test]
    fn reply_chunk_marker_state_without_segment_only_emits_turn_first() {
        let mut state = ReplyChunkMarkerState::default();

        assert_eq!(ReplyChunkMarker::FirstReply, state.observe(None));
        assert_eq!(ReplyChunkMarker::None, state.observe(None));
    }

    #[test]
    fn runtime_turn_stream_audio_chunk_reads_segment_seq() {
        let event: RuntimeTurnStreamEvent = serde_json::from_str(
            r#"{"type":"reply_audio_chunk","audioChunk":{"chunkSeq":4,"segmentSeq":2,"format":"pcm_s16le","payloadBase64":"AA==","last":false}}"#,
        )
        .expect("turn stream event");

        assert_eq!(
            Some(2),
            event.audio_chunk.and_then(|chunk| chunk.segment_seq)
        );
    }

    #[test]
    fn runtime_turn_stream_reads_frozen_m5_and_audio_chunk_timing_contract() {
        let activity: RuntimeTurnStreamEvent = serde_json::from_str(
            r#"{"type":"activity","callId":"call-1","traceId":"trace-1","turnId":"turn-1","activity":{"eventType":"tts_first_audio_chunk_ready","extension":{"streamTimingVersion":1,"streamAnchorId":"0123456789abcdef","streamAnchorServerDeltaMs":1200,"runtimeSessionNonceHash":"abcdef012345","segmentSeq":1,"streamTimingValidation":"bound"}}}"#,
        )
        .expect("m5 activity event");
        let audio: RuntimeTurnStreamEvent = serde_json::from_str(
            r#"{"type":"reply_audio_chunk","callId":"call-1","traceId":"trace-1","turnId":"turn-1","audioChunk":{"chunkSeq":1,"segmentSeq":1,"streamTimingVersion":1,"streamAnchorId":"0123456789abcdef","format":"pcm_s16le","payloadBase64":"AA==","last":false}}"#,
        )
        .expect("timed audio chunk event");

        assert_eq!(Some("call-1"), activity.call_id.as_deref());
        assert_eq!(Some("trace-1"), activity.trace_id.as_deref());
        assert_eq!(Some("turn-1"), activity.turn_id.as_deref());
        let extension = activity.activity.unwrap().extension.unwrap();
        assert_eq!(Some(1), extension.stream_timing_version);
        assert_eq!(Some(1200), extension.stream_anchor_server_delta_ms);
        assert_eq!(
            Some("0123456789abcdef"),
            audio.audio_chunk.unwrap().stream_anchor_id.as_deref()
        );
    }

    #[test]
    fn runtime_turn_stream_timing_records_m6_and_m7_once_then_rejects_terminal_late_events() {
        let nonce = "runtime-nonce";
        let nonce_hash = runtime_session_nonce_hash(nonce);
        let activity: RuntimeTurnStreamEvent = serde_json::from_str(&format!(
            r#"{{"type":"activity","callId":"call-1","traceId":"trace-1","turnId":"turn-1","activity":{{"eventType":"tts_first_audio_chunk_ready","extension":{{"streamTimingVersion":1,"streamAnchorId":"0123456789abcdef","streamAnchorServerDeltaMs":1200,"runtimeSessionNonceHash":"{nonce_hash}","segmentSeq":1,"streamTimingValidation":"bound"}}}}}}"#,
        ))
        .expect("m5 activity event");
        let audio: RuntimeTurnStreamEvent = serde_json::from_str(
            r#"{"type":"reply_audio_chunk","callId":"call-1","traceId":"trace-1","turnId":"turn-1","audioChunk":{"chunkSeq":1,"segmentSeq":1,"streamTimingVersion":1,"streamAnchorId":"0123456789abcdef","format":"pcm_s16le","payloadBase64":"AA==","last":false}}"#,
        )
        .expect("timed audio chunk event");
        let mut state = RuntimeTurnStreamState::default();

        assert!(state.arm_timing_anchor("call-1", "trace-1", "turn-1", nonce, &activity));
        let chunk = audio.audio_chunk.as_ref().expect("audio chunk");
        assert!(
            state
                .record_m6("call-1", "trace-1", "turn-1", &audio, chunk)
                .is_some()
        );
        assert!(
            state
                .record_m6("call-1", "trace-1", "turn-1", &audio, chunk)
                .is_none()
        );
        assert!(state.record_m7().is_some());
        assert!(state.record_m7().is_none());

        state.close_timing();
        assert_eq!(RuntimeTurnStreamTimingPhase::Closed, state.timing.phase);
        assert!(state.timing.anchor.is_none());
        assert!(!state.arm_timing_anchor("call-1", "trace-1", "turn-1", nonce, &activity));
        assert!(
            state
                .record_m6("call-1", "trace-1", "turn-1", &audio, chunk)
                .is_none()
        );
        assert!(state.record_m7().is_none());
    }
}