fix(helper): bind ASR ingress to fixture generation

cai

2026-08-25 caeed9acdc676aa10f788e81edf45892b8cdcc8e

 src/asr_realtime.rs

@@ -34,6 +34,7 @@
pub(crate) struct AudioIngressMetadata {
    pub(crate) input_source_category: String,
    pub(crate) client_fixture_sequence: String,
    pub(crate) input_generation: u64,
    pub(crate) language: Option<String>,
}

@@ -52,10 +53,15 @@
    pub(crate) fn origin_status(attributes: &HashMap<String, String>) -> &'static str {
        let source = attributes.get("inputSourceCategory").map(String::as_str);
        let sequence = attributes.get("clientFixtureSequence").map(String::as_str);
        let generation = attributes.get("inputGeneration").map(String::as_str);
        match (source, sequence) {
            (None, None) => "ordinary_mic_absent",
            (Some("controlled_fixture"), Some(sequence)) if valid_sequence(sequence) => {
                "controlled_fixture_bound"
                match generation {
                    Some(value) if valid_generation(value) => "controlled_fixture_bound",
                    None => "participant_attributes_absent",
                    Some(_) => "participant_attributes_invalid",
                }
            }
            (Some("controlled_fixture"), None) => "sequence_absent",
            (Some("controlled_fixture"), Some(_)) => "participant_attributes_invalid",
@@ -91,6 +97,13 @@
        match (source, sequence) {
            (None, None) => Ok(None),
            (Some("controlled_fixture"), Some(sequence)) if valid_sequence(sequence) => {
                let generation = attributes
                    .get("inputGeneration")
                    .ok_or("missing_generation")?
                    .parse::<u64>()
                    .ok()
                    .filter(|value| *value > 0)
                    .ok_or("invalid_generation")?;
                let language = attributes.get("language").map(String::as_str);
                if let Some(language) = language {
                    if !valid_language(language) {
@@ -100,6 +113,7 @@
                Ok(Some(Self {
                    input_source_category: "controlled_fixture".to_string(),
                    client_fixture_sequence: sequence.to_string(),
                    input_generation: generation,
                    language: language.map(str::to_string),
                }))
            }
@@ -115,6 +129,10 @@
        && value
            .chars()
            .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.'))
}

fn valid_generation(value: &str) -> bool {
    value.parse::<u64>().is_ok_and(|generation| generation > 0)
}

fn valid_language(value: &str) -> bool {
@@ -429,6 +447,7 @@
    if let Some(metadata) = ingress_metadata {
        line["inputSourceCategory"] = json!(metadata.input_source_category);
        line["clientFixtureSequence"] = json!(metadata.client_fixture_sequence);
        line["inputGeneration"] = json!(metadata.input_generation);
        if let Some(language) = metadata.language.as_deref() {
            line["language"] = json!(language);
        }
@@ -561,6 +580,7 @@
            "clientFixtureSequence".to_string(),
            "fixture-01".to_string(),
        );
        attributes.insert("inputGeneration".to_string(), "1".to_string());
        attributes.insert("language".to_string(), "ja-JP".to_string());
        let metadata =
            AudioIngressMetadata::from_participant(&attributes).expect("valid attributes");
@@ -568,6 +588,7 @@
            Some(AudioIngressMetadata {
                input_source_category: "controlled_fixture".to_string(),
                client_fixture_sequence: "fixture-01".to_string(),
                input_generation: 1,
                language: Some("ja-JP".to_string()),
            }),
            metadata
@@ -652,6 +673,7 @@
        let bound = AudioIngressMetadata {
            input_source_category: "controlled_fixture".to_string(),
            client_fixture_sequence: "fixture-01".to_string(),
            input_generation: 1,
            language: Some("ja-JP".to_string()),
        };
        let absent = serde_json::from_slice::<serde_json::Value>(
@@ -674,6 +696,7 @@
        .expect("bound json");
        assert_eq!("controlled_fixture", with_metadata["inputSourceCategory"]);
        assert_eq!("fixture-01", with_metadata["clientFixtureSequence"]);
        assert_eq!(1, with_metadata["inputGeneration"]);
        assert_eq!("ja-JP", with_metadata["language"]);
        assert_eq!(
            "controlled_fixture_bound",
@@ -682,6 +705,7 @@
        let next = AudioIngressMetadata {
            input_source_category: "controlled_fixture".to_string(),
            client_fixture_sequence: "fixture-02".to_string(),
            input_generation: 1,
            language: Some("zh-CN".to_string()),
        };
        let next_line = session_start_line(
@@ -712,6 +736,7 @@
            "clientFixtureSequence".to_string(),
            "fixture-01".to_string(),
        );
        attributes.insert("inputGeneration".to_string(), "1".to_string());
        let (url1, captured1, server1) =
            spawn_http_fixture(json!({"code": 0, "data": {"status": "cancelled"}}).to_string());
        let upload1 = RealtimeAsrUpload::start_with_participant_attributes(
@@ -762,6 +787,24 @@
    }

    #[test]
    fn controlled_fixture_session_requires_current_generation_binding() {
        let attributes = HashMap::from([
            (
                "inputSourceCategory".to_string(),
                "controlled_fixture".to_string(),
            ),
            (
                "clientFixtureSequence".to_string(),
                "fixture-01".to_string(),
            ),
        ]);
        assert_eq!(
            Err("missing_generation"),
            AudioIngressMetadata::from_participant(&attributes)
        );
    }

    #[test]
    fn maximum_audio_chunk_stays_within_java_line_limit() {
        let line = audio_chunk_line(1, &vec![0; 8_000]).expect("maximum chunk line");
        assert!(line.len() <= MAX_NDJSON_LINE_BYTES);