From dc98e7dbb2e909beab577f287da8dccac374c714 Mon Sep 17 00:00:00 2001
From: cai <cai@nbcai.cc>
Date: Tue, 11 Aug 2026 19:00:19 +0800
Subject: [PATCH] test(asr): exercise origin status production boundary
---
src/main.rs | 145 +++++++++++++++++++++++++++++++++++++++++-------
1 files changed, 123 insertions(+), 22 deletions(-)
diff --git a/src/main.rs b/src/main.rs
index 29c1d2a..b880777 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -809,6 +809,7 @@
turn_bridge_config.clone(),
http.clone(),
sink.clone(),
+ user_participant_identity.clone(),
participant,
);
}
@@ -2921,6 +2922,7 @@
turn_bridge_config: TurnBridgeConfig,
http: Client,
sink: Arc<BotAudioOutputSink>,
+ expected_participant_identity: Option<String>,
participant: RemoteParticipant,
) -> JoinHandle<()> {
tokio::spawn(async move {
@@ -3034,7 +3036,10 @@
if let Some(vad) = simple_vad.as_mut() {
if vad_enabled_gate.load(Ordering::Acquire) {
- let (was_in_speech, is_in_speech, turn) = observe_frame_and_start_session(
+ let participant_identity = participant.identity().to_string();
+ let (was_in_speech, is_in_speech, turn) = observe_bound_participant_frame(
+ &participant_identity,
+ expected_participant_identity.as_deref(),
vad,
&call_id,
&trace_id,
@@ -3216,6 +3221,52 @@
(was_in_speech, is_in_speech, turn)
}
+fn observe_bound_participant_frame<F>(
+ participant_identity: &str,
+ expected_participant: Option<&str>,
+ vad: &mut SimpleVad,
+ call_id: &str,
+ trace_id: &str,
+ participant_alias: &str,
+ track_sid_alias: &str,
+ frame_count: u64,
+ elapsed_ms: u64,
+ frame: &AudioFrame<'_>,
+ http: Client,
+ config: RealtimeAsrConfig,
+ read_attributes: F,
+ upload_slot: &mut Option<RealtimeAsrUpload>,
+ last_fixture_sequence: &mut Option<String>,
+ realtime_enabled: bool,
+) -> (bool, bool, Option<FinishedSpeechTurn>)
+where
+ F: FnOnce() -> std::collections::HashMap<String, String>,
+{
+ if !is_bound_user_participant(participant_identity, expected_participant) {
+ warn!(
+ "audioIngressOriginStatus" = "wrong_participant_or_track",
+ "runtime helper rejected audio participant before VAD/session"
+ );
+ return (vad.in_speech, vad.in_speech, None);
+ }
+ observe_frame_and_start_session(
+ vad,
+ call_id,
+ trace_id,
+ participant_alias,
+ track_sid_alias,
+ frame_count,
+ elapsed_ms,
+ frame,
+ http,
+ config,
+ read_attributes,
+ upload_slot,
+ last_fixture_sequence,
+ realtime_enabled,
+ )
+}
+
fn start_realtime_session_for_new_speech(
http: Client,
config: RealtimeAsrConfig,
@@ -3228,11 +3279,14 @@
realtime_enabled: bool,
) {
let turn_id = format!("turn-{:04}", vad.turn_index);
- let metadata = match AudioIngressMetadata::from_participant(&read_attributes()) {
+ let attributes = read_attributes();
+ let origin_status = AudioIngressMetadata::origin_status(&attributes);
+ let metadata = match AudioIngressMetadata::from_participant(&attributes) {
Ok(metadata) => metadata,
Err(reason) => {
warn!(call_id = %call_id, trace_id = %trace_id, turn_id = %turn_id,
- reason, "runtime helper asr_realtime_metadata_rejected");
+ reason, audioIngressOriginStatus = %AudioIngressMetadata::rejection_status(reason),
+ "runtime helper asr_realtime_metadata_rejected");
return;
}
};
@@ -3242,6 +3296,7 @@
&metadata.client_fixture_sequence,
) {
warn!(call_id = %call_id, trace_id = %trace_id, turn_id = %turn_id,
+ audioIngressOriginStatus = "sequence_replayed_or_regressed",
"runtime helper asr_realtime_metadata_sequence_rejected");
return;
}
@@ -3260,7 +3315,7 @@
*last_fixture_sequence = Some(metadata.client_fixture_sequence);
}
info!(call_id = %call_id, trace_id = %trace_id, turn_id = %turn_id,
- "runtime helper asr_realtime_session_started");
+ origin_status, "runtime helper asr_realtime_session_started");
*upload_slot = Some(upload);
}
Err(error) if realtime_enabled => {
@@ -4040,16 +4095,16 @@
#[cfg(test)]
mod tests {
- use super::{
- ReplyChunkMarker, ReplyChunkMarkerState, RuntimeTurnDeviceOutput, RuntimeTurnStreamEvent,
- RuntimeTurnStreamState, RuntimeTurnStreamTimingPhase, runtime_session_nonce_hash,
- should_publish_device_output,
- };
+ use super::*;
use std::{
collections::HashSet,
io::{Read, Write},
net::TcpListener,
- sync::mpsc,
+ sync::{
+ Arc,
+ atomic::{AtomicUsize, Ordering},
+ mpsc,
+ },
thread,
time::Duration,
};
@@ -4157,6 +4212,8 @@
let listener = TcpListener::bind("127.0.0.1:0").expect("bind local ASR fixture");
let address = listener.local_addr().expect("fixture address");
let (request_tx, request_rx) = mpsc::channel::<String>();
+ let captured_count = Arc::new(AtomicUsize::new(0));
+ let captured_count_for_server = Arc::clone(&captured_count);
let server = thread::spawn(move || {
for _ in 0..2 {
let (mut stream, _) = listener.accept().expect("accept ASR session");
@@ -4180,6 +4237,7 @@
request_tx
.send(String::from_utf8_lossy(&bytes).into_owned())
.expect("capture ASR request");
+ captured_count_for_server.fetch_add(1, Ordering::SeqCst);
stream
.write_all(b"HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: 39\r\nconnection: close\r\n\r\n{\"code\":0,\"data\":{\"status\":\"ok\"}}")
.expect("write fixture response");
@@ -4220,7 +4278,9 @@
runtime_session_nonce: Some("test".to_string()),
chunk_duration_ms: 200,
};
- let (was, is, turn) = observe_frame_and_start_session(
+ let (was, is, turn) = observe_bound_participant_frame(
+ "participant-user",
+ Some("participant-user"),
&mut vad,
"call-001",
"trace-001",
@@ -4244,7 +4304,9 @@
"clientFixtureSequence".to_string(),
"fixture-02".to_string(),
);
- let (was, is, turn) = observe_frame_and_start_session(
+ let (was, is, turn) = observe_bound_participant_frame(
+ "participant-user",
+ Some("participant-user"),
&mut vad,
"call-001",
"trace-001",
@@ -4270,10 +4332,14 @@
let second_request = request_rx
.recv_timeout(Duration::from_secs(2))
.expect("second session request");
- assert!(first_request.contains("\\\"clientFixtureSequence\\\":\\\"fixture-01\\\""));
- assert!(second_request.contains("\\\"clientFixtureSequence\\\":\\\"fixture-02\\\""));
- server.join().expect("fixture server");
-
+ assert!(first_request.contains("\"clientFixtureSequence\":\"fixture-01\""));
+ assert!(second_request.contains("\"clientFixtureSequence\":\"fixture-02\""));
+ assert!(
+ first_request.contains("\"audioIngressOriginStatus\":\"controlled_fixture_bound\"")
+ );
+ assert!(
+ second_request.contains("\"audioIngressOriginStatus\":\"controlled_fixture_bound\"")
+ );
// The same production boundary rejects a wrong participant before VAD/session creation.
assert!(!is_bound_user_participant(
"participant-other",
@@ -4283,9 +4349,11 @@
vad.reset_current_turn();
attrs.insert(
"clientFixtureSequence".to_string(),
- "fixture-01".to_string(),
+ "fixture-03".to_string(),
);
- let (_, _, _) = observe_frame_and_start_session(
+ let (_, is_wrong, wrong_turn) = observe_bound_participant_frame(
+ "participant-other",
+ Some("participant-user"),
&mut vad,
"call-001",
"trace-001",
@@ -4307,13 +4375,15 @@
&mut last_fixture_sequence,
true,
);
- assert!(upload.is_none());
+ assert!(!is_wrong && wrong_turn.is_none() && upload.is_none());
vad.reset_current_turn();
attrs.insert(
"clientFixtureSequence".to_string(),
- "fixture-00".to_string(),
+ "fixture-01".to_string(),
);
- let (_, _, _) = observe_frame_and_start_session(
+ let (_, _, _) = observe_bound_participant_frame(
+ "participant-user",
+ Some("participant-user"),
&mut vad,
"call-001",
"trace-001",
@@ -4337,9 +4407,37 @@
);
assert!(upload.is_none());
vad.reset_current_turn();
+ let (_, _, _) = observe_bound_participant_frame(
+ "participant-user",
+ Some("participant-user"),
+ &mut vad,
+ "call-001",
+ "trace-001",
+ "participant-user",
+ "track-001",
+ 5,
+ 5_000,
+ &frame,
+ Client::new(),
+ RealtimeAsrConfig {
+ enabled: true,
+ url: Some(format!("http://{address}/runtime/asr/realtime")),
+ runtime_token: Some("test".to_string()),
+ runtime_session_nonce: Some("test".to_string()),
+ chunk_duration_ms: 200,
+ },
+ || attrs.clone(),
+ &mut upload,
+ &mut last_fixture_sequence,
+ true,
+ );
+ assert!(upload.is_none());
+ vad.reset_current_turn();
attrs.insert("inputSourceCategory".to_string(), "other".to_string());
let mut invalid_upload = None;
- let (_, _, invalid_turn) = observe_frame_and_start_session(
+ let (_, _, invalid_turn) = observe_bound_participant_frame(
+ "participant-user",
+ Some("participant-user"),
&mut vad,
"call-001",
"trace-001",
@@ -4363,6 +4461,9 @@
);
assert!(invalid_turn.is_none());
assert!(invalid_upload.is_none());
+ assert_eq!(2, captured_count.load(Ordering::SeqCst));
+ assert_eq!(0, request_rx.try_iter().count());
+ server.join().expect("fixture server");
}
#[test]
--
Gitblit v1.9.3