From efc73c570bb70ba4de23afc785350629ef5016a7 Mon Sep 17 00:00:00 2001
From: cai <cai@nbcai.cc>
Date: Sat, 08 Aug 2026 18:03:18 +0800
Subject: [PATCH] feat: pass bounded ASR ingress metadata
---
src/asr_realtime.rs | 115 ++++++++++++++++++++++++++++++++++++-
helper-paired-manifest.json | 23 +++++--
src/main.rs | 30 +++++++++
3 files changed, 156 insertions(+), 12 deletions(-)
diff --git a/helper-paired-manifest.json b/helper-paired-manifest.json
index 7e9075c..6df0fb9 100644
--- a/helper-paired-manifest.json
+++ b/helper-paired-manifest.json
@@ -8,11 +8,13 @@
"lmDocHelperCommit": "18008c8c88ba5144b2d728a6f09ce7a46e55e41d",
"lmDocHelperTree": "3462a8af48ca986d4f9f0ea4e5babc551b6f0766",
"manifestCommit": "39dd2f93680a1799845b32e0d24eb9e6ee724954",
- "lmDocRollbackCommit": "0cec0c4ffed92f277778adbbd14eea3e5aa562b6"
+ "lmDocRollbackCommit": "0cec0c4ffed92f277778adbbd14eea3e5aa562b6",
+ "implementationBaseCommit": "afd9833f82e9c1f6d5dfe7f2250f19f1af5cdd7d",
+ "implementationBaseTree": "a9af1be31c66740fefe54223e0ea0559b78aaba0"
},
"pairedJava": {
- "commit": "cdbd85605212297149b434b2cd3154c55f8559d3",
- "tree": "b085cd0a56ce5f77cbcb143a417134248f83a98e"
+ "commit": "46648183280ae5a72c5c237eac26eedd44e06235",
+ "tree": "55b9db04ac30c5f2b4fabaf06f305db065801d87"
},
"downlinkContract": {
"inputEvent": "device_output",
@@ -21,10 +23,19 @@
"dedupeKey": "commandId",
"invalidInput": "fail_closed"
},
+ "asrRealtimeMetadataContract": {
+ "carrier": "livekit_participant_attributes_only",
+ "fields": [
+ "inputSourceCategory",
+ "clientFixtureSequence"
+ ],
+ "absentInput": "omit_fields",
+ "invalidInput": "fail_closed"
+ },
"verification": {
- "status": "JENKINS_30_BUILD_ONLY_SUCCESS",
+ "status": "NEEDS_CI_VALIDATION",
"localCargo": "WEBRTC_SYS_BUILD_TIMEOUT",
- "test": "NOT_RUN",
+ "test": "NOT_RUN_LOCAL_BUILD_BLOCKED",
"deploy": false,
"runtimeChange": false,
"image": {
@@ -36,6 +47,6 @@
"bytes": "UNPROVABLE",
"arch": "UNPROVABLE"
},
- "scope": "src/main.rs only; no dependency, protocol, config, or runtime changes"
+ "scope": "src/main.rs and src/asr_realtime.rs; manifest metadata only; no dependency, protocol, config, or runtime changes"
}
}
diff --git a/src/asr_realtime.rs b/src/asr_realtime.rs
index 16ab4d1..16d2fd3 100644
--- a/src/asr_realtime.rs
+++ b/src/asr_realtime.rs
@@ -1,4 +1,4 @@
-use std::{io, time::Duration};
+use std::{collections::HashMap, io, time::Duration};
use anyhow::{Context, Result, anyhow};
use base64::{Engine as _, engine::general_purpose};
@@ -28,6 +28,40 @@
pub(crate) runtime_token: Option<String>,
pub(crate) runtime_session_nonce: Option<String>,
pub(crate) chunk_duration_ms: u64,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub(crate) struct AudioIngressMetadata {
+ pub(crate) input_source_category: String,
+ pub(crate) client_fixture_sequence: String,
+}
+
+impl AudioIngressMetadata {
+ pub(crate) fn from_participant(
+ attributes: &HashMap<String, String>,
+ ) -> Result<Option<Self>, &'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) => Ok(None),
+ (Some("controlled_fixture"), Some(sequence)) if valid_sequence(sequence) => {
+ Ok(Some(Self {
+ input_source_category: "controlled_fixture".to_string(),
+ client_fixture_sequence: sequence.to_string(),
+ }))
+ }
+ (Some(_), _) => Err("invalid_source_or_sequence"),
+ _ => Err("incomplete_metadata"),
+ }
+ }
+}
+
+fn valid_sequence(value: &str) -> bool {
+ !value.is_empty()
+ && value.len() <= 64
+ && value
+ .chars()
+ .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.'))
}
impl RealtimeAsrConfig {
@@ -66,12 +100,13 @@
trace_id: &str,
turn_id: &str,
initial_samples_48k: &[i16],
+ ingress_metadata: Option<&AudioIngressMetadata>,
) -> Result<Self> {
if !config.is_ready() {
return Err(anyhow!("realtime asr config is not ready"));
}
let nonce = config.runtime_session_nonce.as_deref().unwrap_or_default();
- let session_line = session_start_line(call_id, trace_id, turn_id, nonce)?;
+ let session_line = session_start_line(call_id, trace_id, turn_id, nonce, ingress_metadata)?;
let (sender, receiver) = mpsc::channel(UPLOAD_QUEUE_CAPACITY);
let request_call_id = call_id.to_string();
let request_trace_id = trace_id.to_string();
@@ -295,8 +330,9 @@
trace_id: &str,
turn_id: &str,
runtime_session_nonce: &str,
+ ingress_metadata: Option<&AudioIngressMetadata>,
) -> Result<Vec<u8>> {
- encode_line(json!({
+ let mut line = json!({
"event": "session_start",
"callId": call_id,
"traceId": trace_id,
@@ -307,7 +343,12 @@
"sampleRate": SAMPLE_RATE_16K,
"channels": CHANNELS_MONO,
}
- }))
+ });
+ if let Some(metadata) = ingress_metadata {
+ line["inputSourceCategory"] = json!(metadata.input_source_category);
+ line["clientFixtureSequence"] = json!(metadata.client_fixture_sequence);
+ }
+ encode_line(line)
}
fn audio_chunk_line(chunk_seq: u64, samples: &[i16]) -> Result<Vec<u8>> {
@@ -408,7 +449,7 @@
#[test]
fn session_start_uses_canonical_nonce_hash_and_audio_contract() {
- let line = session_start_line("call-001", "trace-001", "turn-0001", "nonce-001")
+ let line = session_start_line("call-001", "trace-001", "turn-0001", "nonce-001", None)
.expect("session start line");
let value: serde_json::Value = serde_json::from_slice(&line).expect("valid json");
@@ -417,6 +458,68 @@
assert_eq!("pcm_s16le", value["audio"]["format"]);
assert_eq!(16000, value["audio"]["sampleRate"]);
assert_eq!(1, value["audio"]["channels"]);
+ }
+
+ #[test]
+ fn participant_attributes_only_metadata_is_bounded_and_frozen() {
+ let mut attributes = HashMap::new();
+ attributes.insert(
+ "inputSourceCategory".to_string(),
+ "controlled_fixture".to_string(),
+ );
+ attributes.insert(
+ "clientFixtureSequence".to_string(),
+ "fixture-01".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(),
+ }),
+ metadata
+ );
+
+ attributes.insert(
+ "clientFixtureSequence".to_string(),
+ "bad sequence".to_string(),
+ );
+ assert_eq!(
+ Err("invalid_source_or_sequence"),
+ AudioIngressMetadata::from_participant(&attributes)
+ );
+ assert_eq!(
+ Ok(None),
+ AudioIngressMetadata::from_participant(&HashMap::new())
+ );
+ }
+
+ #[test]
+ fn session_start_omits_absent_attributes_and_emits_bound_attributes() {
+ let bound = AudioIngressMetadata {
+ input_source_category: "controlled_fixture".to_string(),
+ client_fixture_sequence: "fixture-01".to_string(),
+ };
+ let absent = serde_json::from_slice::<serde_json::Value>(
+ &session_start_line("call-001", "trace-001", "turn-0001", "nonce-001", None)
+ .expect("absent session line"),
+ )
+ .expect("absent json");
+ assert!(absent.get("inputSourceCategory").is_none());
+ let with_metadata = serde_json::from_slice::<serde_json::Value>(
+ &session_start_line(
+ "call-001",
+ "trace-001",
+ "turn-0001",
+ "nonce-001",
+ Some(&bound),
+ )
+ .expect("bound session line"),
+ )
+ .expect("bound json");
+ assert_eq!("controlled_fixture", with_metadata["inputSourceCategory"]);
+ assert_eq!("fixture-01", with_metadata["clientFixtureSequence"]);
}
#[test]
@@ -487,6 +590,7 @@
"trace-001",
"turn-0001",
&vec![1; 9_600],
+ None,
)
.expect("start upload");
upload
@@ -530,6 +634,7 @@
"trace-002",
"turn-0002",
&vec![1; 9_600],
+ None,
)
.expect("start upload");
diff --git a/src/main.rs b/src/main.rs
index e76d222..c3fa805 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -15,7 +15,9 @@
};
use anyhow::{Context, Result, anyhow};
-use asr_realtime::{RealtimeAsrConfig, RealtimeAsrOutcome, RealtimeAsrUpload};
+use asr_realtime::{
+ AudioIngressMetadata, RealtimeAsrConfig, RealtimeAsrOutcome, RealtimeAsrUpload,
+};
use audio::{AudioDiagnostics, load_pre_recorded_frames};
use base64::{Engine as _, engine::general_purpose};
use futures_util::StreamExt;
@@ -707,6 +709,7 @@
let enabled = config.user_audio_observer_enabled;
let simple_vad_enabled = config.simple_vad_enabled;
let simple_vad_config = config.simple_vad_config.clone();
+ let user_participant_identity = config.user_participant_identity.clone();
let turn_bridge_config = TurnBridgeConfig::from_config(config);
tokio::spawn(async move {
@@ -728,6 +731,7 @@
turn_bridge_config,
http,
sink,
+ user_participant_identity,
)
.await;
})
@@ -743,6 +747,7 @@
turn_bridge_config: TurnBridgeConfig,
http: Client,
sink: Arc<BotAudioOutputSink>,
+ user_participant_identity: Option<String>,
) {
info!(
call_id = %call_id,
@@ -766,6 +771,25 @@
publication: _,
participant,
} => {
+ if user_participant_identity
+ .as_deref()
+ .is_some_and(|expected| participant.identity().to_string() != expected)
+ {
+ warn!(call_id = %call_id, trace_id = %trace_id,
+ metadata_status = "wrong_participant",
+ "runtime helper ignored non-user audio participant");
+ continue;
+ }
+ let ingress_metadata =
+ match AudioIngressMetadata::from_participant(&participant.attributes()) {
+ Ok(value) => value,
+ Err(reason) => {
+ warn!(call_id = %call_id, trace_id = %trace_id,
+ metadata_status = "invalid", reason = reason,
+ "runtime helper ignored invalid audio ingress metadata");
+ None
+ }
+ };
let participant_alias = redact(&participant.identity().to_string());
let track_sid_alias = redact(&track.sid().to_string());
let track_name = track.name();
@@ -777,6 +801,9 @@
track_sid_alias = %track_sid_alias,
track_name = %track_name,
track_source = %track_source,
+ metadata_status = if ingress_metadata.is_some() { "bound" } else { "absent" },
+ metadata_source = ingress_metadata.as_ref().map(|_| "controlled_fixture"),
+ metadata_sequence_present = ingress_metadata.is_some(),
"runtime helper user_track_subscribed"
);
spawn_user_audio_frame_observer(
@@ -3034,6 +3061,7 @@
&trace_id,
&turn_id,
&vad.speech_samples,
+ ingress_metadata.as_ref(),
) {
Ok(upload) => {
info!(
--
Gitblit v1.9.3