fix(asr): preserve missing fixture sequence status

cai

2026-08-11 18f3d31052e32ddcb719fdc1ae856f8b86772920

 src/main.rs

@@ -3243,6 +3243,10 @@
    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(
@@ -3275,11 +3279,16 @@
    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::rejected_origin_status(
                    reason, &attributes
                ),
                "runtime helper asr_realtime_metadata_rejected");
            return;
        }
    };
@@ -3289,6 +3298,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;
        }
@@ -3307,7 +3317,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 => {
@@ -4087,11 +4097,7 @@

#[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},
@@ -4328,8 +4334,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\\\""));
        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",
@@ -4423,6 +4435,38 @@
        );
        assert!(upload.is_none());
        vad.reset_current_turn();
        attrs.remove("clientFixtureSequence");
        let (_, _, missing_sequence_turn) = observe_bound_participant_frame(
            "participant-user",
            Some("participant-user"),
            &mut vad,
            "call-001",
            "trace-001",
            "participant-user",
            "track-001",
            6,
            6_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!(missing_sequence_turn.is_none() && upload.is_none());
        assert_eq!(
            "sequence_absent",
            AudioIngressMetadata::rejected_origin_status("incomplete_metadata", &attrs)
        );
        assert_eq!(2, captured_count.load(Ordering::SeqCst));
        vad.reset_current_turn();
        attrs.insert("inputSourceCategory".to_string(), "other".to_string());
        let mut invalid_upload = None;
        let (_, _, invalid_turn) = observe_bound_participant_frame(