feat: pass bounded ASR ingress metadata

cai

2026-08-08 efc73c570bb70ba4de23afc785350629ef5016a7

 src/asr_realtime.rs

@@ -1,4 +1,4 @@
use std::{io, time::Duration};
use std::{collections::HashMap, io, time::Duration};

use anyhow::{Context, Result, anyhow};
use base64::{Engine as _, engine::general_purpose};
@@ -28,6 +28,40 @@
    pub(crate) runtime_token: Option<String>,
    pub(crate) runtime_session_nonce: Option<String>,
    pub(crate) chunk_duration_ms: u64,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct AudioIngressMetadata {
    pub(crate) input_source_category: String,
    pub(crate) client_fixture_sequence: String,
}

impl AudioIngressMetadata {
    pub(crate) fn from_participant(
        attributes: &HashMap<String, String>,
    ) -> Result<Option<Self>, &'static str> {
        let source = attributes.get("inputSourceCategory").map(String::as_str);
        let sequence = attributes.get("clientFixtureSequence").map(String::as_str);
        match (source, sequence) {
            (None, None) => Ok(None),
            (Some("controlled_fixture"), Some(sequence)) if valid_sequence(sequence) => {
                Ok(Some(Self {
                    input_source_category: "controlled_fixture".to_string(),
                    client_fixture_sequence: sequence.to_string(),
                }))
            }
            (Some(_), _) => Err("invalid_source_or_sequence"),
            _ => Err("incomplete_metadata"),
        }
    }
}

fn valid_sequence(value: &str) -> bool {
    !value.is_empty()
        && value.len() <= 64
        && value
            .chars()
            .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.'))
}

impl RealtimeAsrConfig {
@@ -66,12 +100,13 @@
        trace_id: &str,
        turn_id: &str,
        initial_samples_48k: &[i16],
        ingress_metadata: Option<&AudioIngressMetadata>,
    ) -> 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 session_line = session_start_line(call_id, trace_id, turn_id, nonce, ingress_metadata)?;
        let (sender, receiver) = mpsc::channel(UPLOAD_QUEUE_CAPACITY);
        let request_call_id = call_id.to_string();
        let request_trace_id = trace_id.to_string();
@@ -295,8 +330,9 @@
    trace_id: &str,
    turn_id: &str,
    runtime_session_nonce: &str,
    ingress_metadata: Option<&AudioIngressMetadata>,
) -> Result<Vec<u8>> {
    encode_line(json!({
    let mut line = json!({
        "event": "session_start",
        "callId": call_id,
        "traceId": trace_id,
@@ -307,7 +343,12 @@
            "sampleRate": SAMPLE_RATE_16K,
            "channels": CHANNELS_MONO,
        }
    }))
    });
    if let Some(metadata) = ingress_metadata {
        line["inputSourceCategory"] = json!(metadata.input_source_category);
        line["clientFixtureSequence"] = json!(metadata.client_fixture_sequence);
    }
    encode_line(line)
}

fn audio_chunk_line(chunk_seq: u64, samples: &[i16]) -> Result<Vec<u8>> {
@@ -408,7 +449,7 @@

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

@@ -417,6 +458,68 @@
        assert_eq!("pcm_s16le", value["audio"]["format"]);
        assert_eq!(16000, value["audio"]["sampleRate"]);
        assert_eq!(1, value["audio"]["channels"]);
    }

    #[test]
    fn participant_attributes_only_metadata_is_bounded_and_frozen() {
        let mut attributes = HashMap::new();
        attributes.insert(
            "inputSourceCategory".to_string(),
            "controlled_fixture".to_string(),
        );
        attributes.insert(
            "clientFixtureSequence".to_string(),
            "fixture-01".to_string(),
        );
        let metadata =
            AudioIngressMetadata::from_participant(&attributes).expect("valid attributes");
        assert_eq!(
            Some(AudioIngressMetadata {
                input_source_category: "controlled_fixture".to_string(),
                client_fixture_sequence: "fixture-01".to_string(),
            }),
            metadata
        );

        attributes.insert(
            "clientFixtureSequence".to_string(),
            "bad sequence".to_string(),
        );
        assert_eq!(
            Err("invalid_source_or_sequence"),
            AudioIngressMetadata::from_participant(&attributes)
        );
        assert_eq!(
            Ok(None),
            AudioIngressMetadata::from_participant(&HashMap::new())
        );
    }

    #[test]
    fn session_start_omits_absent_attributes_and_emits_bound_attributes() {
        let bound = AudioIngressMetadata {
            input_source_category: "controlled_fixture".to_string(),
            client_fixture_sequence: "fixture-01".to_string(),
        };
        let absent = serde_json::from_slice::<serde_json::Value>(
            &session_start_line("call-001", "trace-001", "turn-0001", "nonce-001", None)
                .expect("absent session line"),
        )
        .expect("absent json");
        assert!(absent.get("inputSourceCategory").is_none());
        let with_metadata = serde_json::from_slice::<serde_json::Value>(
            &session_start_line(
                "call-001",
                "trace-001",
                "turn-0001",
                "nonce-001",
                Some(&bound),
            )
            .expect("bound session line"),
        )
        .expect("bound json");
        assert_eq!("controlled_fixture", with_metadata["inputSourceCategory"]);
        assert_eq!("fixture-01", with_metadata["clientFixtureSequence"]);
    }

    #[test]
@@ -487,6 +590,7 @@
            "trace-001",
            "turn-0001",
            &vec![1; 9_600],
            None,
        )
        .expect("start upload");
        upload
@@ -530,6 +634,7 @@
            "trace-002",
            "turn-0002",
            &vec![1; 9_600],
            None,
        )
        .expect("start upload");