From f3237616a0bc95cf50fd790f07b72128c624b947 Mon Sep 17 00:00:00 2001
From: cai <cai@nbcai.cc>
Date: Wed, 12 Aug 2026 12:41:44 +0800
Subject: [PATCH] Bind fixture ACK evidence to probe hashes
---
src/asr_realtime.rs | 192 +++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 191 insertions(+), 1 deletions(-)
diff --git a/src/asr_realtime.rs b/src/asr_realtime.rs
index 9358d23..ad99ff2 100644
--- a/src/asr_realtime.rs
+++ b/src/asr_realtime.rs
@@ -37,6 +37,51 @@
}
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);
+ match (source, sequence) {
+ (None, None) => "ordinary_mic_absent",
+ (Some("controlled_fixture"), Some(sequence)) if valid_sequence(sequence) => {
+ "controlled_fixture_bound"
+ }
+ (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> {
@@ -93,6 +138,31 @@
}
impl RealtimeAsrUpload {
+ pub(crate) fn start_with_participant_attributes<F>(
+ http: Client,
+ config: RealtimeAsrConfig,
+ call_id: &str,
+ trace_id: &str,
+ turn_id: &str,
+ initial_samples_48k: &[i16],
+ read_attributes: F,
+ ) -> Result<Self>
+ where
+ F: FnOnce() -> HashMap<String, String>,
+ {
+ let metadata = AudioIngressMetadata::from_participant(&read_attributes())
+ .map_err(|reason| anyhow!(reason))?;
+ Self::start(
+ http,
+ config,
+ call_id,
+ trace_id,
+ turn_id,
+ initial_samples_48k,
+ metadata.as_ref(),
+ )
+ }
+
pub(crate) fn start(
http: Client,
config: RealtimeAsrConfig,
@@ -325,7 +395,7 @@
samples: Vec<i16>,
}
-fn session_start_line(
+pub(crate) fn session_start_line(
call_id: &str,
trace_id: &str,
turn_id: &str,
@@ -348,6 +418,10 @@
line["inputSourceCategory"] = json!(metadata.input_source_category);
line["clientFixtureSequence"] = json!(metadata.client_fixture_sequence);
}
+ line["audioIngressOriginStatus"] = json!(match ingress_metadata {
+ Some(_) => "controlled_fixture_bound",
+ None => "ordinary_mic_absent",
+ });
encode_line(line)
}
@@ -458,6 +532,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]
@@ -480,6 +555,10 @@
}),
metadata
);
+ assert_eq!(
+ "controlled_fixture_bound",
+ AudioIngressMetadata::origin_status(&attributes)
+ );
attributes.insert(
"clientFixtureSequence".to_string(),
@@ -493,6 +572,52 @@
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())
+ );
+ 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]
@@ -507,6 +632,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",
@@ -520,6 +646,10 @@
.expect("bound json");
assert_eq!("controlled_fixture", with_metadata["inputSourceCategory"]);
assert_eq!("fixture-01", with_metadata["clientFixtureSequence"]);
+ 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(),
@@ -540,6 +670,66 @@
);
}
+ #[tokio::test]
+ async fn production_session_boundary_reads_updated_attributes_per_session() {
+ let mut attributes = HashMap::new();
+ attributes.insert(
+ "inputSourceCategory".to_string(),
+ "controlled_fixture".to_string(),
+ );
+ attributes.insert(
+ "clientFixtureSequence".to_string(),
+ "fixture-01".to_string(),
+ );
+ let (url1, captured1, server1) =
+ spawn_http_fixture(json!({"code": 0, "data": {"status": "cancelled"}}).to_string());
+ let upload1 = RealtimeAsrUpload::start_with_participant_attributes(
+ Client::new(),
+ fixture_config(url1),
+ "call-001",
+ "trace-001",
+ "turn-0001",
+ &vec![1; 9_600],
+ || attributes.clone(),
+ )
+ .expect("session one");
+ upload1.cancel("test").await;
+ let request1 = captured1.recv().expect("session one request");
+ server1.join().expect("session one server");
+ assert!(
+ request1
+ .body
+ .contains("\"clientFixtureSequence\":\"fixture-01\"")
+ );
+ assert!(!request1.body.contains("fixture-02"));
+
+ attributes.insert(
+ "clientFixtureSequence".to_string(),
+ "fixture-02".to_string(),
+ );
+ let (url2, captured2, server2) =
+ spawn_http_fixture(json!({"code": 0, "data": {"status": "cancelled"}}).to_string());
+ let upload2 = RealtimeAsrUpload::start_with_participant_attributes(
+ Client::new(),
+ fixture_config(url2),
+ "call-001",
+ "trace-001",
+ "turn-0002",
+ &vec![1; 9_600],
+ || attributes.clone(),
+ )
+ .expect("session two");
+ upload2.cancel("test").await;
+ let request2 = captured2.recv().expect("session two request");
+ server2.join().expect("session two server");
+ assert!(
+ request2
+ .body
+ .contains("\"clientFixtureSequence\":\"fixture-02\"")
+ );
+ assert!(!request2.body.contains("fixture-01"));
+ }
+
#[test]
fn maximum_audio_chunk_stays_within_java_line_limit() {
let line = audio_chunk_line(1, &vec![0; 8_000]).expect("maximum chunk line");
--
Gitblit v1.9.3