From caeed9acdc676aa10f788e81edf45892b8cdcc8e Mon Sep 17 00:00:00 2001
From: cai <cai@nbcai.cc>
Date: Tue, 25 Aug 2026 12:48:13 +0800
Subject: [PATCH] fix(helper): bind ASR ingress to fixture generation

---
 src/asr_realtime.rs |  100 ++++++++++++++++++++++++++++++++++++++++++++++++-
 1 files changed, 97 insertions(+), 3 deletions(-)

diff --git a/src/asr_realtime.rs b/src/asr_realtime.rs
index a7ff7f7..8007cdd 100644
--- a/src/asr_realtime.rs
+++ b/src/asr_realtime.rs
@@ -34,6 +34,8 @@
 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>,
 }
 
 impl AudioIngressMetadata {
@@ -51,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",
@@ -70,6 +77,18 @@
         }
     }
 
+    pub(crate) fn rejected_origin_status(
+        reason: &str,
+        attributes: &HashMap<String, String>,
+    ) -> &'static str {
+        let status = Self::origin_status(attributes);
+        if status == "sequence_absent" {
+            status
+        } else {
+            Self::rejection_status(reason)
+        }
+    }
+
     pub(crate) fn from_participant(
         attributes: &HashMap<String, String>,
     ) -> Result<Option<Self>, &'static str> {
@@ -78,9 +97,24 @@
         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) {
+                        return Err("invalid_language");
+                    }
+                }
                 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),
                 }))
             }
             (Some(_), _) => Err("invalid_source_or_sequence"),
@@ -95,6 +129,14 @@
         && 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 {
+    matches!(value, "ja-JP" | "zh-CN")
 }
 
 impl RealtimeAsrConfig {
@@ -405,6 +447,10 @@
     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);
+        }
     }
     line["audioIngressOriginStatus"] = json!(match ingress_metadata {
         Some(_) => "controlled_fixture_bound",
@@ -534,12 +580,16 @@
             "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");
         assert_eq!(
             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
         );
@@ -577,6 +627,16 @@
             "ordinary_mic_absent",
             AudioIngressMetadata::origin_status(&HashMap::new())
         );
+        attributes.insert(
+            "clientFixtureSequence".to_string(),
+            "fixture-01".to_string(),
+        );
+        attributes.insert("language".to_string(), "ja-jp".to_string());
+        assert_eq!(
+            Err("invalid_language"),
+            AudioIngressMetadata::from_participant(&attributes)
+        );
+        attributes.insert("language".to_string(), "ja-JP".to_string());
         assert_eq!(
             "participant_attributes_absent",
             AudioIngressMetadata::rejection_status("incomplete_metadata")
@@ -586,16 +646,24 @@
             AudioIngressMetadata::rejection_status("invalid_source_or_sequence")
         );
         assert_eq!(
-            "sequence_replayed_or_regressed",
+            "origin_unprovable",
             AudioIngressMetadata::rejection_status("sequence_replayed")
         );
         assert_eq!(
-            "wrong_participant_or_track",
+            "origin_unprovable",
             AudioIngressMetadata::rejection_status("wrong_participant")
         );
         assert_eq!(
             "origin_unprovable",
             AudioIngressMetadata::rejection_status("unknown")
+        );
+        let missing_sequence = HashMap::from([(
+            "inputSourceCategory".to_string(),
+            "controlled_fixture".to_string(),
+        )]);
+        assert_eq!(
+            "sequence_absent",
+            AudioIngressMetadata::rejected_origin_status("incomplete_metadata", &missing_sequence)
         );
         assert_eq!(8, AudioIngressMetadata::ORIGIN_STATUSES.len());
     }
@@ -605,6 +673,8 @@
         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>(
             &session_start_line("call-001", "trace-001", "turn-0001", "nonce-001", None)
@@ -626,6 +696,8 @@
         .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",
             with_metadata["audioIngressOriginStatus"]
@@ -633,6 +705,8 @@
         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(
             "call-001",
@@ -644,6 +718,7 @@
         .expect("next bound session line");
         let next_value: serde_json::Value = serde_json::from_slice(&next_line).expect("next json");
         assert_eq!("fixture-02", next_value["clientFixtureSequence"]);
+        assert_eq!("zh-CN", next_value["language"]);
         assert_ne!(
             with_metadata["clientFixtureSequence"],
             next_value["clientFixtureSequence"]
@@ -661,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(
@@ -711,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);

--
Gitblit v1.9.3