test(asr): exercise origin status production boundary

cai

2026-08-11 dc98e7dbb2e909beab577f287da8dccac374c714

 src/main.rs

@@ -809,6 +809,7 @@
                    turn_bridge_config.clone(),
                    http.clone(),
                    sink.clone(),
                    user_participant_identity.clone(),
                    participant,
                );
            }
@@ -2921,6 +2922,7 @@
    turn_bridge_config: TurnBridgeConfig,
    http: Client,
    sink: Arc<BotAudioOutputSink>,
    expected_participant_identity: Option<String>,
    participant: RemoteParticipant,
) -> JoinHandle<()> {
    tokio::spawn(async move {
@@ -3034,7 +3036,10 @@

            if let Some(vad) = simple_vad.as_mut() {
                if vad_enabled_gate.load(Ordering::Acquire) {
                    let (was_in_speech, is_in_speech, turn) = observe_frame_and_start_session(
                    let participant_identity = participant.identity().to_string();
                    let (was_in_speech, is_in_speech, turn) = observe_bound_participant_frame(
                        &participant_identity,
                        expected_participant_identity.as_deref(),
                        vad,
                        &call_id,
                        &trace_id,
@@ -3216,6 +3221,52 @@
    (was_in_speech, is_in_speech, turn)
}

fn observe_bound_participant_frame<F>(
    participant_identity: &str,
    expected_participant: Option<&str>,
    vad: &mut SimpleVad,
    call_id: &str,
    trace_id: &str,
    participant_alias: &str,
    track_sid_alias: &str,
    frame_count: u64,
    elapsed_ms: u64,
    frame: &AudioFrame<'_>,
    http: Client,
    config: RealtimeAsrConfig,
    read_attributes: F,
    upload_slot: &mut Option<RealtimeAsrUpload>,
    last_fixture_sequence: &mut Option<String>,
    realtime_enabled: bool,
) -> (bool, bool, Option<FinishedSpeechTurn>)
where
    F: FnOnce() -> std::collections::HashMap<String, String>,
{
    if !is_bound_user_participant(participant_identity, expected_participant) {
        warn!(
            "audioIngressOriginStatus" = "wrong_participant_or_track",
            "runtime helper rejected audio participant before VAD/session"
        );
        return (vad.in_speech, vad.in_speech, None);
    }
    observe_frame_and_start_session(
        vad,
        call_id,
        trace_id,
        participant_alias,
        track_sid_alias,
        frame_count,
        elapsed_ms,
        frame,
        http,
        config,
        read_attributes,
        upload_slot,
        last_fixture_sequence,
        realtime_enabled,
    )
}

fn start_realtime_session_for_new_speech(
    http: Client,
    config: RealtimeAsrConfig,
@@ -3228,11 +3279,14 @@
    realtime_enabled: bool,
) {
    let turn_id = format!("turn-{:04}", vad.turn_index);
    let metadata = match AudioIngressMetadata::from_participant(&read_attributes()) {
    let attributes = read_attributes();
    let origin_status = AudioIngressMetadata::origin_status(&attributes);
    let metadata = match AudioIngressMetadata::from_participant(&attributes) {
        Ok(metadata) => metadata,
        Err(reason) => {
            warn!(call_id = %call_id, trace_id = %trace_id, turn_id = %turn_id,
                reason, "runtime helper asr_realtime_metadata_rejected");
                reason, audioIngressOriginStatus = %AudioIngressMetadata::rejection_status(reason),
                "runtime helper asr_realtime_metadata_rejected");
            return;
        }
    };
@@ -3242,6 +3296,7 @@
            &metadata.client_fixture_sequence,
        ) {
            warn!(call_id = %call_id, trace_id = %trace_id, turn_id = %turn_id,
                audioIngressOriginStatus = "sequence_replayed_or_regressed",
                "runtime helper asr_realtime_metadata_sequence_rejected");
            return;
        }
@@ -3260,7 +3315,7 @@
                *last_fixture_sequence = Some(metadata.client_fixture_sequence);
            }
            info!(call_id = %call_id, trace_id = %trace_id, turn_id = %turn_id,
                "runtime helper asr_realtime_session_started");
                origin_status, "runtime helper asr_realtime_session_started");
            *upload_slot = Some(upload);
        }
        Err(error) if realtime_enabled => {
@@ -4040,16 +4095,16 @@

#[cfg(test)]
mod tests {
    use super::{
        ReplyChunkMarker, ReplyChunkMarkerState, RuntimeTurnDeviceOutput, RuntimeTurnStreamEvent,
        RuntimeTurnStreamState, RuntimeTurnStreamTimingPhase, runtime_session_nonce_hash,
        should_publish_device_output,
    };
    use super::*;
    use std::{
        collections::HashSet,
        io::{Read, Write},
        net::TcpListener,
        sync::mpsc,
        sync::{
            Arc,
            atomic::{AtomicUsize, Ordering},
            mpsc,
        },
        thread,
        time::Duration,
    };
@@ -4157,6 +4212,8 @@
        let listener = TcpListener::bind("127.0.0.1:0").expect("bind local ASR fixture");
        let address = listener.local_addr().expect("fixture address");
        let (request_tx, request_rx) = mpsc::channel::<String>();
        let captured_count = Arc::new(AtomicUsize::new(0));
        let captured_count_for_server = Arc::clone(&captured_count);
        let server = thread::spawn(move || {
            for _ in 0..2 {
                let (mut stream, _) = listener.accept().expect("accept ASR session");
@@ -4180,6 +4237,7 @@
                request_tx
                    .send(String::from_utf8_lossy(&bytes).into_owned())
                    .expect("capture ASR request");
                captured_count_for_server.fetch_add(1, Ordering::SeqCst);
                stream
                    .write_all(b"HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: 39\r\nconnection: close\r\n\r\n{\"code\":0,\"data\":{\"status\":\"ok\"}}")
                    .expect("write fixture response");
@@ -4220,7 +4278,9 @@
            runtime_session_nonce: Some("test".to_string()),
            chunk_duration_ms: 200,
        };
        let (was, is, turn) = observe_frame_and_start_session(
        let (was, is, turn) = observe_bound_participant_frame(
            "participant-user",
            Some("participant-user"),
            &mut vad,
            "call-001",
            "trace-001",
@@ -4244,7 +4304,9 @@
            "clientFixtureSequence".to_string(),
            "fixture-02".to_string(),
        );
        let (was, is, turn) = observe_frame_and_start_session(
        let (was, is, turn) = observe_bound_participant_frame(
            "participant-user",
            Some("participant-user"),
            &mut vad,
            "call-001",
            "trace-001",
@@ -4270,10 +4332,14 @@
        let second_request = request_rx
            .recv_timeout(Duration::from_secs(2))
            .expect("second session request");
        assert!(first_request.contains("\\\"clientFixtureSequence\\\":\\\"fixture-01\\\""));
        assert!(second_request.contains("\\\"clientFixtureSequence\\\":\\\"fixture-02\\\""));
        server.join().expect("fixture server");

        assert!(first_request.contains("\"clientFixtureSequence\":\"fixture-01\""));
        assert!(second_request.contains("\"clientFixtureSequence\":\"fixture-02\""));
        assert!(
            first_request.contains("\"audioIngressOriginStatus\":\"controlled_fixture_bound\"")
        );
        assert!(
            second_request.contains("\"audioIngressOriginStatus\":\"controlled_fixture_bound\"")
        );
        // The same production boundary rejects a wrong participant before VAD/session creation.
        assert!(!is_bound_user_participant(
            "participant-other",
@@ -4283,9 +4349,11 @@
        vad.reset_current_turn();
        attrs.insert(
            "clientFixtureSequence".to_string(),
            "fixture-01".to_string(),
            "fixture-03".to_string(),
        );
        let (_, _, _) = observe_frame_and_start_session(
        let (_, is_wrong, wrong_turn) = observe_bound_participant_frame(
            "participant-other",
            Some("participant-user"),
            &mut vad,
            "call-001",
            "trace-001",
@@ -4307,13 +4375,15 @@
            &mut last_fixture_sequence,
            true,
        );
        assert!(upload.is_none());
        assert!(!is_wrong && wrong_turn.is_none() && upload.is_none());
        vad.reset_current_turn();
        attrs.insert(
            "clientFixtureSequence".to_string(),
            "fixture-00".to_string(),
            "fixture-01".to_string(),
        );
        let (_, _, _) = observe_frame_and_start_session(
        let (_, _, _) = observe_bound_participant_frame(
            "participant-user",
            Some("participant-user"),
            &mut vad,
            "call-001",
            "trace-001",
@@ -4337,9 +4407,37 @@
        );
        assert!(upload.is_none());
        vad.reset_current_turn();
        let (_, _, _) = observe_bound_participant_frame(
            "participant-user",
            Some("participant-user"),
            &mut vad,
            "call-001",
            "trace-001",
            "participant-user",
            "track-001",
            5,
            5_000,
            &frame,
            Client::new(),
            RealtimeAsrConfig {
                enabled: true,
                url: Some(format!("http://{address}/runtime/asr/realtime")),
                runtime_token: Some("test".to_string()),
                runtime_session_nonce: Some("test".to_string()),
                chunk_duration_ms: 200,
            },
            || attrs.clone(),
            &mut upload,
            &mut last_fixture_sequence,
            true,
        );
        assert!(upload.is_none());
        vad.reset_current_turn();
        attrs.insert("inputSourceCategory".to_string(), "other".to_string());
        let mut invalid_upload = None;
        let (_, _, invalid_turn) = observe_frame_and_start_session(
        let (_, _, invalid_turn) = observe_bound_participant_frame(
            "participant-user",
            Some("participant-user"),
            &mut vad,
            "call-001",
            "trace-001",
@@ -4363,6 +4461,9 @@
        );
        assert!(invalid_turn.is_none());
        assert!(invalid_upload.is_none());
        assert_eq!(2, captured_count.load(Ordering::SeqCst));
        assert_eq!(0, request_rx.try_iter().count());
        server.join().expect("fixture server");
    }

    #[test]