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 | 45 ++++++++
src/main.rs | 210 +++++++++++++++++++++++++++++++++++++++--
2 files changed, 240 insertions(+), 15 deletions(-)
diff --git a/src/asr_realtime.rs b/src/asr_realtime.rs
index 3903ca0..8007cdd 100644
--- a/src/asr_realtime.rs
+++ b/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);
diff --git a/src/main.rs b/src/main.rs
index 0e179f5..b18a534 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -70,7 +70,7 @@
const CONTROLLED_FIXTURE_ACK_RESULTS: [&str; 4] =
["observed", "rejected", "timeout", "publish_failed"];
-const CONTROLLED_FIXTURE_REJECT_REASONS: [&str; 10] = [
+const CONTROLLED_FIXTURE_REJECT_REASONS: [&str; 16] = [
"missing_attributes",
"wrong_source",
"missing_sequence",
@@ -80,6 +80,12 @@
"duplicate_or_old_sequence",
"no_current_participant",
"ack_publish_failed",
+ "missing_generation",
+ "invalid_generation",
+ "wrong_generation",
+ "invalid_language",
+ "incomplete_metadata",
+ "invalid_source_or_sequence",
"unknown",
];
@@ -264,7 +270,7 @@
trace_id_hash: &str,
generation: u64,
sequence: &str,
- acknowledged_probe_sequences: &mut HashSet<String>,
+ acknowledged_probe_sequences: &mut HashSet<(u64, String)>,
) -> ControlledFixtureAckPublishOutcome
where
F: Future<Output = Result<(), E>>,
@@ -282,7 +288,7 @@
Some(ack_result),
reject_reason,
);
- acknowledged_probe_sequences.insert(sequence.to_string());
+ acknowledged_probe_sequences.insert((generation, sequence.to_string()));
ControlledFixtureAckPublishOutcome {
observed,
published: true,
@@ -338,7 +344,7 @@
let probe: ControlledFixtureAttributeProbe = serde_json::from_slice(payload).ok()?;
if probe.message_type != CONTROLLED_FIXTURE_PROBE_TOPIC
|| probe.protocol_version != CONTROLLED_FIXTURE_PROTOCOL_VERSION
- || probe.generation != CONTROLLED_FIXTURE_GENERATION
+ || !valid_input_generation(probe.generation)
|| probe.call_id_hash != sha256_hex(call_id)
|| probe.call_trace_id_hash != sha256_hex(trace_id)
|| probe.client_fixture_sequence.trim().is_empty()
@@ -565,6 +571,58 @@
return Err("wrong_sequence");
}
Ok(())
+}
+
+fn controlled_fixture_probe_attribute_decision(
+ actual_participant: &str,
+ expected_participant: Option<&str>,
+ attributes: &std::collections::HashMap<String, String>,
+ probe: &PendingControlledFixtureProbe,
+) -> Result<(), &'static str> {
+ if !is_bound_user_participant(actual_participant, expected_participant) {
+ return Err("wrong_participant");
+ }
+ let metadata = match AudioIngressMetadata::from_participant(attributes) {
+ Ok(Some(metadata)) => metadata,
+ Ok(None) => return Err("missing_attributes"),
+ Err(reason) => return Err(reason),
+ };
+ if metadata.client_fixture_sequence != probe.sequence {
+ return Err("wrong_sequence");
+ }
+ if metadata.input_generation != probe.generation {
+ return Err("wrong_generation");
+ }
+ Ok(())
+}
+
+async fn observe_controlled_fixture_probe_attributes<F>(
+ expires_at: Instant,
+ actual_participant: &str,
+ expected_participant: Option<&str>,
+ probe: &PendingControlledFixtureProbe,
+ mut read_attributes: F,
+) -> Result<(), &'static str>
+where
+ F: FnMut() -> std::collections::HashMap<String, String>,
+{
+ loop {
+ if Instant::now() > expires_at {
+ return Err("timeout");
+ }
+ match controlled_fixture_probe_attribute_decision(
+ actual_participant,
+ expected_participant,
+ &read_attributes(),
+ probe,
+ ) {
+ Ok(()) => return Ok(()),
+ Err("missing_attributes") | Err("missing_generation") => {
+ sleep(CONTROLLED_FIXTURE_PROBE_RECHECK_DELAY).await;
+ }
+ Err(reason) => return Err(reason),
+ }
+ }
}
async fn observe_controlled_fixture_attributes<F>(
@@ -1301,6 +1359,10 @@
expected.is_none_or(|value| identity == value)
}
+fn valid_input_generation(generation: u64) -> bool {
+ generation > 0
+}
+
async fn observe_user_audio_events(
mut events: UnboundedReceiver<RoomEvent>,
call_id: String,
@@ -1450,7 +1512,9 @@
if let Ok(probe) =
serde_json::from_slice::<ControlledFixtureAttributeProbe>(&payload)
{
- if acknowledged_probe_sequences.contains(&probe.client_fixture_sequence) {
+ if acknowledged_probe_sequences
+ .contains(&(probe.generation, probe.client_fixture_sequence.clone()))
+ {
record_controlled_fixture_probe_event(
&call_id,
&trace_id,
@@ -1543,7 +1607,7 @@
async fn process_controlled_fixture_probe(
pending_probe: &mut Option<PendingControlledFixtureProbe>,
- acknowledged_probe_sequences: &mut HashSet<String>,
+ acknowledged_probe_sequences: &mut HashSet<(u64, String)>,
participant: Option<&RemoteParticipant>,
sink: &BotAudioOutputSink,
expected_participant: Option<&str>,
@@ -1554,7 +1618,7 @@
let Some(probe) = pending_probe.take() else {
return None;
};
- if acknowledged_probe_sequences.contains(&probe.sequence) {
+ if acknowledged_probe_sequences.contains(&(probe.generation, probe.sequence.clone())) {
record_controlled_fixture_probe_event(
runtime_call_id,
runtime_trace_id,
@@ -1615,11 +1679,14 @@
);
return Some(false);
}
- let decision = controlled_fixture_probe_binding_decision(
- probe.sender.as_str(),
- Some(participant.identity().as_str()),
+ let decision = observe_controlled_fixture_probe_attributes(
+ probe.expires_at,
+ participant.identity().as_str(),
expected_participant,
- );
+ &probe,
+ || participant.attributes(),
+ )
+ .await;
if let Err(reason) = decision {
record_controlled_fixture_attribute_decision(
Err(reason),
@@ -3842,6 +3909,7 @@
};
let mut realtime_asr_upload: Option<RealtimeAsrUpload> = None;
let mut last_fixture_sequence: Option<String> = None;
+ let mut last_fixture_generation: Option<u64> = None;
while let Some(drained) = frame_rx.recv().await {
let frame = drained.frame;
@@ -3892,6 +3960,7 @@
|| participant.attributes(),
&mut realtime_asr_upload,
&mut last_fixture_sequence,
+ &mut last_fixture_generation,
turn_bridge_config.asr_realtime_enabled,
);
@@ -4028,6 +4097,7 @@
read_attributes: F,
upload_slot: &mut Option<RealtimeAsrUpload>,
last_fixture_sequence: &mut Option<String>,
+ last_fixture_generation: &mut Option<u64>,
realtime_enabled: bool,
) -> (bool, bool, Option<FinishedSpeechTurn>)
where
@@ -4054,6 +4124,7 @@
read_attributes,
upload_slot,
last_fixture_sequence,
+ last_fixture_generation,
realtime_enabled,
);
}
@@ -4076,6 +4147,7 @@
read_attributes: F,
upload_slot: &mut Option<RealtimeAsrUpload>,
last_fixture_sequence: &mut Option<String>,
+ last_fixture_generation: &mut Option<u64>,
realtime_enabled: bool,
) -> (bool, bool, Option<FinishedSpeechTurn>)
where
@@ -4102,6 +4174,7 @@
read_attributes,
upload_slot,
last_fixture_sequence,
+ last_fixture_generation,
realtime_enabled,
)
}
@@ -4115,6 +4188,7 @@
read_attributes: impl FnOnce() -> std::collections::HashMap<String, String>,
upload_slot: &mut Option<RealtimeAsrUpload>,
last_fixture_sequence: &mut Option<String>,
+ last_fixture_generation: &mut Option<u64>,
realtime_enabled: bool,
) {
let turn_id = format!("turn-{:04}", vad.turn_index);
@@ -4132,9 +4206,11 @@
}
};
if let Some(metadata) = metadata.as_ref() {
- if !fixture_sequence_is_new(
+ if !fixture_binding_is_new(
last_fixture_sequence.as_deref(),
+ *last_fixture_generation,
&metadata.client_fixture_sequence,
+ metadata.input_generation,
) {
warn!(call_id = %call_id, trace_id = %trace_id, turn_id = %turn_id,
audioIngressOriginStatus = "sequence_replayed_or_regressed",
@@ -4154,6 +4230,7 @@
Ok(upload) => {
if let Some(metadata) = metadata {
*last_fixture_sequence = Some(metadata.client_fixture_sequence);
+ *last_fixture_generation = Some(metadata.input_generation);
}
info!(call_id = %call_id, trace_id = %trace_id, turn_id = %turn_id,
origin_status, "runtime helper asr_realtime_session_started");
@@ -4181,6 +4258,23 @@
match (previous_number, current_number) {
(Some(previous), Some(current)) => current > previous,
_ => previous != current,
+ }
+}
+
+fn fixture_binding_is_new(
+ previous_sequence: Option<&str>,
+ previous_generation: Option<u64>,
+ current_sequence: &str,
+ current_generation: u64,
+) -> bool {
+ if !valid_input_generation(current_generation) {
+ return false;
+ }
+ match previous_generation {
+ Some(previous) if current_generation < previous => false,
+ Some(previous) if current_generation > previous => true,
+ Some(_) => fixture_sequence_is_new(previous_sequence, current_sequence),
+ None => true,
}
}
@@ -5040,6 +5134,7 @@
"clientFixtureSequence".to_string(),
"fixture-01".to_string(),
),
+ ("inputGeneration".to_string(), "1".to_string()),
("language".to_string(), "ja-JP".to_string()),
]);
let mut starts = Vec::new();
@@ -5174,10 +5269,12 @@
"clientFixtureSequence".to_string(),
"fixture-01".to_string(),
),
+ ("inputGeneration".to_string(), "1".to_string()),
("language".to_string(), "ja-JP".to_string()),
]);
let mut upload = None;
let mut last_fixture_sequence = None;
+ let mut last_fixture_generation = None;
let config = RealtimeAsrConfig {
enabled: true,
url: Some(format!("http://{address}/runtime/asr/realtime")),
@@ -5201,6 +5298,7 @@
|| attrs.clone(),
&mut upload,
&mut last_fixture_sequence,
+ &mut last_fixture_generation,
true,
);
assert!(!was && is && turn.is_none());
@@ -5211,6 +5309,7 @@
"clientFixtureSequence".to_string(),
"fixture-02".to_string(),
);
+ attrs.insert("inputGeneration".to_string(), "2".to_string());
attrs.insert("language".to_string(), "zh-CN".to_string());
let (was, is, turn) = observe_bound_participant_frame(
"participant-user",
@@ -5224,10 +5323,11 @@
2_000,
&frame,
Client::new(),
- config,
+ config.clone(),
|| attrs.clone(),
&mut upload,
&mut last_fixture_sequence,
+ &mut last_fixture_generation,
true,
);
assert!(!was && is && turn.is_none());
@@ -5242,6 +5342,14 @@
.expect("second session request");
assert!(first_request.contains("\"clientFixtureSequence\":\"fixture-01\""));
assert!(second_request.contains("\"clientFixtureSequence\":\"fixture-02\""));
+ assert!(first_request.contains("\"callId\":\"call-001\""));
+ assert!(first_request.contains("\"traceId\":\"trace-001\""));
+ assert!(second_request.contains("\"callId\":\"call-001\""));
+ assert!(second_request.contains("\"traceId\":\"trace-001\""));
+ assert!(first_request.contains("\"inputSourceCategory\":\"controlled_fixture\""));
+ assert!(second_request.contains("\"inputSourceCategory\":\"controlled_fixture\""));
+ assert!(first_request.contains("\"inputGeneration\":1"));
+ assert!(second_request.contains("\"inputGeneration\":2"));
assert!(first_request.contains("\"language\":\"ja-JP\""));
assert!(second_request.contains("\"language\":\"zh-CN\""));
assert!(
@@ -5260,6 +5368,32 @@
attrs.insert(
"clientFixtureSequence".to_string(),
"fixture-03".to_string(),
+ );
+ attrs.insert("inputGeneration".to_string(), "1".to_string());
+ let (_, is_old_generation, old_generation_turn) = observe_bound_participant_frame(
+ "participant-user",
+ Some("participant-user"),
+ &mut vad,
+ "call-001",
+ "trace-001",
+ "participant-user",
+ "track-001",
+ 3,
+ 3_000,
+ &frame,
+ Client::new(),
+ config.clone(),
+ || attrs.clone(),
+ &mut upload,
+ &mut last_fixture_sequence,
+ &mut last_fixture_generation,
+ true,
+ );
+ assert!(is_old_generation && old_generation_turn.is_none() && upload.is_none());
+ vad.reset_current_turn();
+ attrs.insert(
+ "clientFixtureSequence".to_string(),
+ "fixture-04".to_string(),
);
let (_, is_wrong, wrong_turn) = observe_bound_participant_frame(
"participant-other",
@@ -5283,6 +5417,7 @@
|| attrs.clone(),
&mut upload,
&mut last_fixture_sequence,
+ &mut last_fixture_generation,
true,
);
assert!(!is_wrong && wrong_turn.is_none() && upload.is_none());
@@ -5313,6 +5448,7 @@
|| attrs.clone(),
&mut upload,
&mut last_fixture_sequence,
+ &mut last_fixture_generation,
true,
);
assert!(upload.is_none());
@@ -5339,6 +5475,7 @@
|| attrs.clone(),
&mut upload,
&mut last_fixture_sequence,
+ &mut last_fixture_generation,
true,
);
assert!(upload.is_none());
@@ -5366,6 +5503,7 @@
|| attrs.clone(),
&mut upload,
&mut last_fixture_sequence,
+ &mut last_fixture_generation,
true,
);
assert!(missing_sequence_turn.is_none() && upload.is_none());
@@ -5399,6 +5537,7 @@
|| attrs,
&mut invalid_upload,
&mut last_fixture_sequence,
+ &mut last_fixture_generation,
true,
);
assert!(invalid_turn.is_none());
@@ -5841,7 +5980,7 @@
published: true,
}
);
- assert!(acknowledged.contains("fixture-02"));
+ assert!(acknowledged.contains(&(CONTROLLED_FIXTURE_GENERATION, "fixture-02".to_string())));
assert_eq!(successful_observer_starts, 1);
}
@@ -5878,6 +6017,49 @@
}
#[tokio::test]
+ async fn production_probe_waits_for_generation_attribute_before_binding() {
+ let received_at = Instant::now();
+ let probe = PendingControlledFixtureProbe {
+ sender: ParticipantIdentity("user-1".to_string()),
+ call_id_hash: "call-hash".to_string(),
+ call_trace_id_hash: "trace-hash".to_string(),
+ generation: 2,
+ sequence: "fixture-02".to_string(),
+ received_at,
+ expires_at: received_at + CONTROLLED_FIXTURE_PROBE_TTL,
+ };
+ let expected = HashMap::from([
+ (
+ "inputSourceCategory".to_string(),
+ "controlled_fixture".to_string(),
+ ),
+ (
+ "clientFixtureSequence".to_string(),
+ "fixture-02".to_string(),
+ ),
+ ("inputGeneration".to_string(), "2".to_string()),
+ ]);
+ let mut reads = 0;
+ let result = observe_controlled_fixture_probe_attributes(
+ probe.expires_at,
+ "user-1",
+ Some("user-1"),
+ &probe,
+ || {
+ reads += 1;
+ if reads == 1 {
+ HashMap::new()
+ } else {
+ expected.clone()
+ }
+ },
+ )
+ .await;
+ assert_eq!(result, Ok(()));
+ assert_eq!(reads, 2);
+ }
+
+ #[tokio::test]
async fn production_attribute_observation_rejects_wrong_sequence_without_audio_effect() {
let attributes = HashMap::from([
(
--
Gitblit v1.9.3