From db6463d9f98f740d4e8b6d3a0a25b65688196321 Mon Sep 17 00:00:00 2001
From: cai <cai@nbcai.cc>
Date: Tue, 11 Aug 2026 18:50:21 +0800
Subject: [PATCH] fix(asr): normalize origin rejection statuses
---
src/asr_realtime.rs | 420 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++--
1 files changed, 400 insertions(+), 20 deletions(-)
diff --git a/src/asr_realtime.rs b/src/asr_realtime.rs
index db28b8c..a7ff7f7 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};
@@ -15,8 +15,7 @@
const SAMPLE_RATE_16K: u32 = 16_000;
const CHANNELS_MONO: u32 = 1;
-const MIN_CHUNK_DURATION_MS: u64 = 20;
-const MAX_CHUNK_DURATION_MS: u64 = 1_000;
+const REQUIRED_CHUNK_DURATION_MS: u64 = 200;
const UPLOAD_QUEUE_CAPACITY: usize = 64;
const MAX_NDJSON_LINE_BYTES: usize = 24 * 1024;
const FINISH_RESPONSE_TIMEOUT: Duration = Duration::from_secs(15);
@@ -31,15 +30,80 @@
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) 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 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 {
pub(crate) fn is_ready(&self) -> bool {
self.enabled
&& non_blank(&self.url)
&& non_blank(&self.runtime_token)
&& non_blank(&self.runtime_session_nonce)
- && self.chunk_duration_ms >= MIN_CHUNK_DURATION_MS
- && self.chunk_duration_ms <= MAX_CHUNK_DURATION_MS
- && self.chunk_duration_ms % MIN_CHUNK_DURATION_MS == 0
+ && self.chunk_duration_ms == REQUIRED_CHUNK_DURATION_MS
}
}
@@ -62,6 +126,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,
@@ -69,12 +158,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();
@@ -115,21 +205,27 @@
speech_duration_ms: u64,
end_reason: &str,
) -> Result<RealtimeAsrOutcome> {
- if let Some(chunk) = self.chunker.flush() {
- let line = audio_chunk_line(chunk.seq, &chunk.samples)?;
- self.try_send_line(line)?;
+ let enqueue_result = (|| -> Result<()> {
+ if let Some(chunk) = self.chunker.flush() {
+ let line = audio_chunk_line(chunk.seq, &chunk.samples)?;
+ self.try_send_line(line)?;
+ }
+ self.try_send_line(vad_speech_end_line(speech_duration_ms, end_reason)?)?;
+ self.try_send_line(finish_line()?)?;
+ Ok(())
+ })();
+ if let Err(error) = enqueue_result {
+ self.abort_and_wait().await;
+ return Err(error);
}
- self.try_send_line(vad_speech_end_line(speech_duration_ms, end_reason)?)?;
- self.try_send_line(finish_line()?)?;
self.sender.take();
let chunk_count = self.chunker.chunk_count;
let audio_bytes = self.chunker.audio_bytes;
- let mut task = self.task;
- let mut outcome = match timeout(FINISH_RESPONSE_TIMEOUT, &mut task).await {
+ let mut outcome = match timeout(FINISH_RESPONSE_TIMEOUT, &mut self.task).await {
Ok(joined) => joined.context("realtime asr upload task failed")??,
Err(_) => {
- task.abort();
+ self.abort_and_wait().await;
return Err(anyhow!("realtime asr finish response timeout"));
}
};
@@ -147,8 +243,14 @@
.await
.is_err()
{
- self.task.abort();
+ self.abort_and_wait().await;
}
+ }
+
+ async fn abort_and_wait(&mut self) {
+ self.sender.take();
+ self.task.abort();
+ let _ = timeout(CANCEL_RESPONSE_TIMEOUT, &mut self.task).await;
}
fn try_send_line(&self, line: Vec<u8>) -> Result<()> {
@@ -281,13 +383,14 @@
samples: Vec<i16>,
}
-fn session_start_line(
+pub(crate) fn session_start_line(
call_id: &str,
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,
@@ -298,7 +401,16 @@
"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);
+ }
+ line["audioIngressOriginStatus"] = json!(match ingress_metadata {
+ Some(_) => "controlled_fixture_bound",
+ None => "ordinary_mic_absent",
+ });
+ encode_line(line)
}
fn audio_chunk_line(chunk_seq: u64, samples: &[i16]) -> Result<Vec<u8>> {
@@ -399,7 +511,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");
@@ -408,12 +520,242 @@
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]
+ 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
+ );
+ assert_eq!(
+ "controlled_fixture_bound",
+ AudioIngressMetadata::origin_status(&attributes)
+ );
+
+ 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())
+ );
+ 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!(
+ "sequence_replayed_or_regressed",
+ AudioIngressMetadata::rejection_status("sequence_replayed")
+ );
+ assert_eq!(
+ "wrong_participant_or_track",
+ AudioIngressMetadata::rejection_status("wrong_participant")
+ );
+ assert_eq!(
+ "origin_unprovable",
+ AudioIngressMetadata::rejection_status("unknown")
+ );
+ assert_eq!(8, AudioIngressMetadata::ORIGIN_STATUSES.len());
+ }
+
+ #[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());
+ assert_eq!("ordinary_mic_absent", absent["audioIngressOriginStatus"]);
+ 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"]);
+ 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(),
+ };
+ let next_line = session_start_line(
+ "call-001",
+ "trace-001",
+ "turn-0002",
+ "nonce-001",
+ Some(&next),
+ )
+ .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_ne!(
+ with_metadata["clientFixtureSequence"],
+ next_value["clientFixtureSequence"]
+ );
+ }
+
+ #[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");
assert!(line.len() <= MAX_NDJSON_LINE_BYTES);
+ }
+
+ #[test]
+ fn realtime_config_rejects_chunk_duration_other_than_frozen_200ms() {
+ let mut config = fixture_config("http://127.0.0.1/realtime".to_string());
+ assert!(config.is_ready());
+
+ config.chunk_duration_ms = 500;
+ assert!(!config.is_ready());
+ config.chunk_duration_ms = 1_000;
+ assert!(!config.is_ready());
+ }
+
+ #[tokio::test]
+ async fn finish_when_tail_chunk_enqueue_fails_then_stops_upload_task() {
+ let (mut upload, task_active) = blocked_upload(true);
+ upload.chunker.push_48k(&vec![1; 480]);
+
+ let result = upload.finish(10, "silence").await;
+
+ assert!(result.is_err());
+ assert!(!task_active.load(std::sync::atomic::Ordering::Acquire));
+ }
+
+ #[tokio::test]
+ async fn finish_when_speech_end_enqueue_fails_then_stops_upload_task() {
+ let (upload, task_active) = blocked_upload(true);
+
+ let result = upload.finish(200, "silence").await;
+
+ assert!(result.is_err());
+ assert!(!task_active.load(std::sync::atomic::Ordering::Acquire));
+ }
+
+ #[tokio::test]
+ async fn finish_when_finish_enqueue_fails_then_stops_upload_task() {
+ let (upload, task_active) = blocked_upload(false);
+
+ let result = upload.finish(200, "silence").await;
+
+ assert!(result.is_err());
+ assert!(!task_active.load(std::sync::atomic::Ordering::Acquire));
}
#[tokio::test]
@@ -436,6 +778,7 @@
"trace-001",
"turn-0001",
&vec![1; 9_600],
+ None,
)
.expect("start upload");
upload
@@ -479,6 +822,7 @@
"trace-002",
"turn-0002",
&vec![1; 9_600],
+ None,
)
.expect("start upload");
@@ -500,6 +844,42 @@
}
}
+ fn blocked_upload(
+ prefill_queue: bool,
+ ) -> (
+ RealtimeAsrUpload,
+ std::sync::Arc<std::sync::atomic::AtomicBool>,
+ ) {
+ let (sender, receiver) = mpsc::channel(1);
+ if prefill_queue {
+ sender.try_send(vec![b'x']).expect("prefill queue");
+ }
+ let task_active = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(true));
+ let task_state = task_active.clone();
+ let active_guard = TestUploadTaskGuard(task_state);
+ let task = tokio::spawn(async move {
+ let _active_guard = active_guard;
+ let _receiver = receiver;
+ std::future::pending::<Result<RealtimeAsrOutcome>>().await
+ });
+ (
+ RealtimeAsrUpload {
+ sender: Some(sender),
+ task,
+ chunker: Pcm16kChunker::new(200),
+ },
+ task_active,
+ )
+ }
+
+ struct TestUploadTaskGuard(std::sync::Arc<std::sync::atomic::AtomicBool>);
+
+ impl Drop for TestUploadTaskGuard {
+ fn drop(&mut self) {
+ self.0.store(false, std::sync::atomic::Ordering::Release);
+ }
+ }
+
struct CapturedRequest {
headers: String,
body: String,
--
Gitblit v1.9.3