test: exercise VAD session metadata boundary

cai

2026-08-08 ae894957b3ba4903e4e21920dfb1aad28d30de07

 src/main.rs

@@ -15,7 +15,9 @@
};

use anyhow::{Context, Result, anyhow};
use asr_realtime::{RealtimeAsrConfig, RealtimeAsrOutcome, RealtimeAsrUpload};
use asr_realtime::{
    AudioIngressMetadata, RealtimeAsrConfig, RealtimeAsrOutcome, RealtimeAsrUpload,
};
use audio::{AudioDiagnostics, load_pre_recorded_frames};
use base64::{Engine as _, engine::general_purpose};
use futures_util::StreamExt;
@@ -28,7 +30,7 @@
    options::TrackPublishOptions,
    prelude::{
        DataPacket, LocalAudioTrack, LocalTrack, ParticipantIdentity, RemoteAudioTrack,
        RemoteTrack, Room, RoomEvent, RoomOptions,
        RemoteParticipant, RemoteTrack, Room, RoomEvent, RoomOptions,
    },
};
use reqwest::Client;
@@ -707,6 +709,7 @@
    let enabled = config.user_audio_observer_enabled;
    let simple_vad_enabled = config.simple_vad_enabled;
    let simple_vad_config = config.simple_vad_config.clone();
    let user_participant_identity = config.user_participant_identity.clone();
    let turn_bridge_config = TurnBridgeConfig::from_config(config);

    tokio::spawn(async move {
@@ -728,6 +731,7 @@
            turn_bridge_config,
            http,
            sink,
            user_participant_identity,
        )
        .await;
    })
@@ -743,6 +747,7 @@
    turn_bridge_config: TurnBridgeConfig,
    http: Client,
    sink: Arc<BotAudioOutputSink>,
    user_participant_identity: Option<String>,
) {
    info!(
        call_id = %call_id,
@@ -766,6 +771,15 @@
                publication: _,
                participant,
            } => {
                if user_participant_identity
                    .as_deref()
                    .is_some_and(|expected| participant.identity().to_string() != expected)
                {
                    warn!(call_id = %call_id, trace_id = %trace_id,
                        metadata_status = "wrong_participant",
                        "runtime helper ignored non-user audio participant");
                    continue;
                }
                let participant_alias = redact(&participant.identity().to_string());
                let track_sid_alias = redact(&track.sid().to_string());
                let track_name = track.name();
@@ -791,6 +805,7 @@
                    turn_bridge_config.clone(),
                    http.clone(),
                    sink.clone(),
                    participant,
                );
            }
            RoomEvent::TrackSubscribed {
@@ -985,7 +1000,11 @@
            .await
            {
                Ok(outcome) => {
                    let mut published_device_outputs = HashSet::new();
                    for output in &outcome.device_outputs {
                        if !should_publish_device_output(&mut published_device_outputs, output) {
                            continue;
                        }
                        if let Err(error) = sink
                            .publish_device_output(call_id, trace_id, &turn.turn_id, output)
                            .await
@@ -1593,6 +1612,9 @@
        }
        Some("device_output") => {
            if let Some(output) = event.device_output.as_ref() {
                if !should_publish_device_output(&mut state.published_device_output_ids, output) {
                    return Ok(());
                }
                sink.publish_device_output(call_id, trace_id, &turn.turn_id, output)
                    .await?;
                state.device_output_count = state.device_output_count.saturating_add(1);
@@ -2421,6 +2443,7 @@
    completed: bool,
    audio_chunk_count: u64,
    device_output_count: u64,
    published_device_output_ids: HashSet<String>,
    encoded_audio_buffer: Vec<u8>,
    pcm_stream_decoder: Option<audio::PcmS16leStreamDecoder>,
    pcm_stream_network_chunk_count: u64,
@@ -2438,6 +2461,7 @@
            completed: false,
            audio_chunk_count: 0,
            device_output_count: 0,
            published_device_output_ids: HashSet::new(),
            encoded_audio_buffer: Vec::new(),
            pcm_stream_decoder: None,
            pcm_stream_network_chunk_count: 0,
@@ -2830,6 +2854,30 @@
    params: Option<serde_json::Value>,
}

fn should_publish_device_output(
    published_ids: &mut HashSet<String>,
    output: &RuntimeTurnDeviceOutput,
) -> bool {
    let Some(command_id) = output
        .command_id
        .as_deref()
        .map(str::trim)
        .filter(|value| !value.is_empty())
    else {
        return false;
    };
    if output
        .command_code
        .as_deref()
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .is_none()
    {
        return false;
    }
    published_ids.insert(command_id.to_string())
}

fn require_safe_segment(value: &str) -> Result<()> {
    if value.is_empty()
        || value.contains('/')
@@ -2869,6 +2917,7 @@
    turn_bridge_config: TurnBridgeConfig,
    http: Client,
    sink: Arc<BotAudioOutputSink>,
    participant: RemoteParticipant,
) -> JoinHandle<()> {
    tokio::spawn(async move {
        let mut stream = NativeAudioStream::new(
@@ -2994,13 +3043,14 @@

                    if !was_in_speech && is_in_speech {
                        let turn_id = format!("turn-{:04}", vad.turn_index);
                        match RealtimeAsrUpload::start(
                        match RealtimeAsrUpload::start_with_participant_attributes(
                            http.clone(),
                            turn_bridge_config.realtime_asr_config(),
                            &call_id,
                            &trace_id,
                            &turn_id,
                            &vad.speech_samples,
                            || participant.attributes(),
                        ) {
                            Ok(upload) => {
                                info!(
@@ -3894,9 +3944,85 @@
#[cfg(test)]
mod tests {
    use super::{
        ReplyChunkMarker, ReplyChunkMarkerState, RuntimeTurnStreamEvent, RuntimeTurnStreamState,
        RuntimeTurnStreamTimingPhase, runtime_session_nonce_hash,
        ReplyChunkMarker, ReplyChunkMarkerState, RuntimeTurnDeviceOutput, RuntimeTurnStreamEvent,
        RuntimeTurnStreamState, RuntimeTurnStreamTimingPhase, runtime_session_nonce_hash,
        should_publish_device_output,
    };
    use std::collections::HashSet;

    #[test]
    fn production_vad_session_boundary_reads_updated_attributes() {
        let config = SimpleVadConfig {
            rms_threshold: 0.001,
            peak_threshold: 0.01,
            start_frames: 2,
            end_silence_ms: 100,
            min_speech_ms: 1,
            max_turn_ms: 1_000,
            initial_ignore_ms: 0,
        };
        let mut vad = SimpleVad::new(config);
        let samples = vec![1_000i16; 160];
        let frame = AudioFrame {
            data: samples.as_slice().into(),
            sample_rate: 16_000,
            num_channels: 1,
            samples_per_channel: 160,
        };
        let mut attributes = std::collections::HashMap::from([
            (
                "inputSourceCategory".to_string(),
                "controlled_fixture".to_string(),
            ),
            (
                "clientFixtureSequence".to_string(),
                "fixture-01".to_string(),
            ),
        ]);
        let mut starts = Vec::new();
        for (session_index, sequence) in [(1, "fixture-01"), (2, "fixture-02")] {
            let was_in_speech = vad.in_speech;
            vad.observe_frame(
                "call-001",
                "trace-001",
                "participant",
                "track",
                session_index * 2 - 1,
                1_000 * session_index,
                &frame,
            );
            vad.observe_frame(
                "call-001",
                "trace-001",
                "participant",
                "track",
                session_index * 2,
                1_000 * session_index + 10,
                &frame,
            );
            let is_in_speech = vad.in_speech;
            assert!(!was_in_speech && is_in_speech);
            attributes.insert("clientFixtureSequence".to_string(), sequence.to_string());
            let metadata = AudioIngressMetadata::from_participant(&attributes)
                .expect("valid participant attributes")
                .expect("controlled fixture metadata");
            starts.push(metadata.client_fixture_sequence);
            vad.reset_current_turn();
        }
        assert_eq!(vec!["fixture-01", "fixture-02"], starts);
        attributes.insert("inputSourceCategory".to_string(), "other".to_string());
        assert!(AudioIngressMetadata::from_participant(&attributes).is_err());
        assert!(
            AudioIngressMetadata::from_participant(&std::collections::HashMap::new())
                .expect("missing attributes is absent")
                .is_none()
        );
        attributes.insert(
            "clientFixtureSequence".to_string(),
            "fixture-01".to_string(),
        );
        assert!(AudioIngressMetadata::from_participant(&attributes).is_err());
    }

    #[test]
    fn reply_chunk_marker_state_emits_turn_first_once_and_later_segment_first_once() {
@@ -3993,4 +4119,37 @@
        );
        assert!(state.record_m7().is_none());
    }

    #[test]
    fn device_output_contract_is_reliable_and_deduplicated() {
        let output = RuntimeTurnDeviceOutput {
            command_id: Some("cmd-1".to_string()),
            command_code: Some("custom.app.DeviceLevelChange".to_string()),
            params: Some(serde_json::json!({"level": 1})),
        };
        let mut published = HashSet::new();
        assert!(should_publish_device_output(&mut published, &output));
        assert!(!should_publish_device_output(&mut published, &output));
        assert_eq!(published.len(), 1);
    }

    #[test]
    fn device_output_invalid_or_missing_command_is_fail_closed() {
        for output in [
            RuntimeTurnDeviceOutput {
                command_id: None,
                command_code: Some("custom.app.DeviceLevelChange".to_string()),
                params: None,
            },
            RuntimeTurnDeviceOutput {
                command_id: Some("cmd-1".to_string()),
                command_code: None,
                params: None,
            },
        ] {
            let mut published = HashSet::new();
            assert!(!should_publish_device_output(&mut published, &output));
            assert!(published.is_empty());
        }
    }
}