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 | 179 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 179 insertions(+), 0 deletions(-)
diff --git a/src/asr_realtime.rs b/src/asr_realtime.rs
index a108c44..8007cdd 100644
--- a/src/asr_realtime.rs
+++ b/src/asr_realtime.rs
@@ -34,9 +34,61 @@
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 {
+ pub(crate) const ORIGIN_STATUSES: [&'static str; 8] = [
+ "controlled_fixture_bound",
+ "ordinary_mic_absent",
+ "participant_attributes_absent",
+ "participant_attributes_invalid",
+ "sequence_absent",
+ "sequence_replayed_or_regressed",
+ "wrong_participant_or_track",
+ "origin_unprovable",
+ ];
+
+ 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) => {
+ 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",
+ (Some(_), _) | (None, Some(_)) => "participant_attributes_invalid",
+ }
+ }
+
+ pub(crate) fn rejection_status(reason: &str) -> &'static str {
+ match reason {
+ "incomplete_metadata" => "participant_attributes_absent",
+ "invalid_source_or_sequence" => "participant_attributes_invalid",
+ _ => "origin_unprovable",
+ }
+ }
+
+ 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> {
@@ -45,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"),
@@ -62,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 {
@@ -372,7 +447,15 @@
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",
+ None => "ordinary_mic_absent",
+ });
encode_line(line)
}
@@ -483,6 +566,7 @@
assert_eq!("pcm_s16le", value["audio"]["format"]);
assert_eq!(16000, value["audio"]["sampleRate"]);
assert_eq!(1, value["audio"]["channels"]);
+ assert_eq!("ordinary_mic_absent", value["audioIngressOriginStatus"]);
}
#[test]
@@ -496,14 +580,22 @@
"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
+ );
+ assert_eq!(
+ "controlled_fixture_bound",
+ AudioIngressMetadata::origin_status(&attributes)
);
attributes.insert(
@@ -518,6 +610,62 @@
Ok(None),
AudioIngressMetadata::from_participant(&HashMap::new())
);
+ attributes.remove("clientFixtureSequence");
+ assert_eq!(
+ "sequence_absent",
+ AudioIngressMetadata::origin_status(&attributes)
+ );
+ attributes.insert(
+ "clientFixtureSequence".to_string(),
+ "bad sequence".to_string(),
+ );
+ assert_eq!(
+ "participant_attributes_invalid",
+ AudioIngressMetadata::origin_status(&attributes)
+ );
+ assert_eq!(
+ "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")
+ );
+ assert_eq!(
+ "participant_attributes_invalid",
+ AudioIngressMetadata::rejection_status("invalid_source_or_sequence")
+ );
+ assert_eq!(
+ "origin_unprovable",
+ AudioIngressMetadata::rejection_status("sequence_replayed")
+ );
+ assert_eq!(
+ "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());
}
#[test]
@@ -525,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)
@@ -532,6 +682,7 @@
)
.expect("absent json");
assert!(absent.get("inputSourceCategory").is_none());
+ assert_eq!("ordinary_mic_absent", absent["audioIngressOriginStatus"]);
let with_metadata = serde_json::from_slice::<serde_json::Value>(
&session_start_line(
"call-001",
@@ -545,9 +696,17 @@
.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"]
+ );
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",
@@ -559,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"]
@@ -576,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(
@@ -626,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