use std::{collections::HashMap, io, time::Duration};
|
|
use anyhow::{Context, Result, anyhow};
|
use base64::{Engine as _, engine::general_purpose};
|
use futures_util::stream;
|
use reqwest::Client;
|
use serde::Deserialize;
|
use serde_json::json;
|
use sha2::{Digest, Sha256};
|
use tokio::{
|
sync::mpsc,
|
task::JoinHandle,
|
time::{Instant, timeout},
|
};
|
|
const SAMPLE_RATE_16K: u32 = 16_000;
|
const CHANNELS_MONO: u32 = 1;
|
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);
|
const CANCEL_RESPONSE_TIMEOUT: Duration = Duration::from_secs(3);
|
|
#[derive(Clone)]
|
pub(crate) struct RealtimeAsrConfig {
|
pub(crate) enabled: bool,
|
pub(crate) url: Option<String>,
|
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) 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> {
|
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 == REQUIRED_CHUNK_DURATION_MS
|
}
|
}
|
|
pub(crate) struct RealtimeAsrUpload {
|
sender: Option<mpsc::Sender<Vec<u8>>>,
|
task: JoinHandle<Result<RealtimeAsrOutcome>>,
|
chunker: Pcm16kChunker,
|
}
|
|
pub(crate) struct RealtimeAsrOutcome {
|
pub(crate) status: String,
|
pub(crate) asr_result_ref: Option<String>,
|
pub(crate) provider_alias: Option<String>,
|
pub(crate) partial_count: u64,
|
pub(crate) fallback_reason: Option<String>,
|
pub(crate) fallback_stage: Option<String>,
|
pub(crate) chunk_count: u64,
|
pub(crate) audio_bytes: u64,
|
pub(crate) wall_ms: u64,
|
}
|
|
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,
|
call_id: &str,
|
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, 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();
|
let request_turn_id = turn_id.to_string();
|
let task = tokio::spawn(run_upload(
|
http,
|
config.clone(),
|
request_call_id,
|
request_trace_id,
|
request_turn_id,
|
receiver,
|
));
|
let mut upload = Self {
|
sender: Some(sender),
|
task,
|
chunker: Pcm16kChunker::new(config.chunk_duration_ms),
|
};
|
if let Err(error) = upload
|
.try_send_line(session_line)
|
.and_then(|_| upload.push_48k_samples(initial_samples_48k))
|
{
|
upload.task.abort();
|
return Err(error);
|
}
|
Ok(upload)
|
}
|
|
pub(crate) fn push_48k_samples(&mut self, samples: &[i16]) -> Result<()> {
|
for chunk in self.chunker.push_48k(samples) {
|
let line = audio_chunk_line(chunk.seq, &chunk.samples)?;
|
self.try_send_line(line)?;
|
}
|
Ok(())
|
}
|
|
pub(crate) async fn finish(
|
mut self,
|
speech_duration_ms: u64,
|
end_reason: &str,
|
) -> Result<RealtimeAsrOutcome> {
|
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.sender.take();
|
|
let chunk_count = self.chunker.chunk_count;
|
let audio_bytes = self.chunker.audio_bytes;
|
let mut outcome = match timeout(FINISH_RESPONSE_TIMEOUT, &mut self.task).await {
|
Ok(joined) => joined.context("realtime asr upload task failed")??,
|
Err(_) => {
|
self.abort_and_wait().await;
|
return Err(anyhow!("realtime asr finish response timeout"));
|
}
|
};
|
outcome.chunk_count = chunk_count;
|
outcome.audio_bytes = audio_bytes;
|
Ok(outcome)
|
}
|
|
pub(crate) async fn cancel(mut self, reason: &str) {
|
if let Ok(line) = cancel_line(reason) {
|
let _ = self.try_send_line(line);
|
}
|
self.sender.take();
|
if timeout(CANCEL_RESPONSE_TIMEOUT, &mut self.task)
|
.await
|
.is_err()
|
{
|
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<()> {
|
self.sender
|
.as_ref()
|
.ok_or_else(|| anyhow!("realtime asr upload is closed"))?
|
.try_send(line)
|
.map_err(|error| anyhow!("realtime asr upload backpressure: {error}"))
|
}
|
}
|
|
async fn run_upload(
|
http: Client,
|
config: RealtimeAsrConfig,
|
call_id: String,
|
trace_id: String,
|
turn_id: String,
|
receiver: mpsc::Receiver<Vec<u8>>,
|
) -> Result<RealtimeAsrOutcome> {
|
let started_at = Instant::now();
|
let body_stream = stream::unfold(receiver, |mut receiver| async move {
|
receiver
|
.recv()
|
.await
|
.map(|chunk| (Ok::<Vec<u8>, io::Error>(chunk), receiver))
|
});
|
let response = http
|
.post(config.url.as_deref().unwrap_or_default())
|
.header("Content-Type", "application/x-ndjson")
|
.header(
|
"X-CV-Runtime-Token",
|
config.runtime_token.as_deref().unwrap_or_default(),
|
)
|
.header("X-CV-Call-Id", &call_id)
|
.header("X-CV-Trace-Id", &trace_id)
|
.header(
|
"X-CV-Runtime-Session-Nonce",
|
config.runtime_session_nonce.as_deref().unwrap_or_default(),
|
)
|
.body(reqwest::Body::wrap_stream(body_stream))
|
.send()
|
.await
|
.context("failed to post realtime asr stream")?;
|
let status = response.status();
|
if !status.is_success() {
|
return Err(anyhow!(
|
"realtime asr http failed status={} turn={}",
|
status.as_u16(),
|
turn_id
|
));
|
}
|
let body: RuntimeCommonResult<RuntimeRealtimeAsrResp> = response
|
.json()
|
.await
|
.context("failed to decode realtime asr response")?;
|
if body.code != 0 {
|
return Err(anyhow!(
|
"realtime asr common result failed code={}",
|
body.code
|
));
|
}
|
let data = body
|
.data
|
.ok_or_else(|| anyhow!("realtime asr response data missing"))?;
|
Ok(RealtimeAsrOutcome {
|
status: data.status.unwrap_or_else(|| "unknown".to_string()),
|
asr_result_ref: data.asr_result_ref,
|
provider_alias: data.provider_alias,
|
partial_count: data.partial_count.unwrap_or_default(),
|
fallback_reason: data.fallback_reason,
|
fallback_stage: data.fallback_stage,
|
chunk_count: 0,
|
audio_bytes: 0,
|
wall_ms: started_at.elapsed().as_millis() as u64,
|
})
|
}
|
|
struct Pcm16kChunker {
|
pending: Vec<i16>,
|
samples_per_chunk: usize,
|
next_seq: u64,
|
chunk_count: u64,
|
audio_bytes: u64,
|
}
|
|
impl Pcm16kChunker {
|
fn new(chunk_duration_ms: u64) -> Self {
|
Self {
|
pending: Vec::new(),
|
samples_per_chunk: SAMPLE_RATE_16K as usize * chunk_duration_ms as usize / 1000,
|
next_seq: 0,
|
chunk_count: 0,
|
audio_bytes: 0,
|
}
|
}
|
|
fn push_48k(&mut self, samples: &[i16]) -> Vec<PcmChunk> {
|
self.pending.extend(samples.iter().step_by(3).copied());
|
let mut chunks = Vec::new();
|
while self.pending.len() >= self.samples_per_chunk {
|
let samples = self.pending.drain(..self.samples_per_chunk).collect();
|
chunks.push(self.record_chunk(samples));
|
}
|
chunks
|
}
|
|
fn flush(&mut self) -> Option<PcmChunk> {
|
if self.pending.is_empty() {
|
return None;
|
}
|
let samples = std::mem::take(&mut self.pending);
|
Some(self.record_chunk(samples))
|
}
|
|
fn record_chunk(&mut self, samples: Vec<i16>) -> PcmChunk {
|
self.next_seq = self.next_seq.saturating_add(1);
|
self.chunk_count = self.chunk_count.saturating_add(1);
|
self.audio_bytes = self
|
.audio_bytes
|
.saturating_add((samples.len() * size_of::<i16>()) as u64);
|
PcmChunk {
|
seq: self.next_seq,
|
samples,
|
}
|
}
|
}
|
|
struct PcmChunk {
|
seq: u64,
|
samples: Vec<i16>,
|
}
|
|
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>> {
|
let mut line = json!({
|
"event": "session_start",
|
"callId": call_id,
|
"traceId": trace_id,
|
"turnId": turn_id,
|
"runtimeSessionNonceHash": sha12(runtime_session_nonce),
|
"audio": {
|
"format": "pcm_s16le",
|
"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>> {
|
let mut bytes = Vec::with_capacity(samples.len() * size_of::<i16>());
|
for sample in samples {
|
bytes.extend_from_slice(&sample.to_le_bytes());
|
}
|
encode_line(json!({
|
"event": "audio_chunk",
|
"chunkSeq": chunk_seq,
|
"audioBase64": general_purpose::STANDARD.encode(bytes),
|
"durationMs": ((samples.len() as u64) * 1000 / u64::from(SAMPLE_RATE_16K)).max(1),
|
}))
|
}
|
|
fn vad_speech_end_line(speech_duration_ms: u64, end_reason: &str) -> Result<Vec<u8>> {
|
encode_line(json!({
|
"event": "vad_speech_end",
|
"speechDurationMs": speech_duration_ms,
|
"endReason": end_reason,
|
}))
|
}
|
|
fn finish_line() -> Result<Vec<u8>> {
|
encode_line(json!({"event": "finish"}))
|
}
|
|
fn cancel_line(reason: &str) -> Result<Vec<u8>> {
|
encode_line(json!({"event": "cancel", "reason": reason}))
|
}
|
|
fn encode_line(value: serde_json::Value) -> Result<Vec<u8>> {
|
let mut line = serde_json::to_vec(&value)?;
|
line.push(b'\n');
|
if line.len() > MAX_NDJSON_LINE_BYTES {
|
return Err(anyhow!("realtime asr ndjson line exceeds 24KiB"));
|
}
|
Ok(line)
|
}
|
|
fn sha12(value: &str) -> String {
|
let digest = Sha256::digest(value.as_bytes());
|
format!("{digest:x}")[..12].to_string()
|
}
|
|
fn non_blank(value: &Option<String>) -> bool {
|
value.as_ref().is_some_and(|value| !value.trim().is_empty())
|
}
|
|
#[derive(Deserialize)]
|
struct RuntimeCommonResult<T> {
|
code: i64,
|
data: Option<T>,
|
}
|
|
#[derive(Deserialize)]
|
#[serde(rename_all = "camelCase")]
|
struct RuntimeRealtimeAsrResp {
|
status: Option<String>,
|
asr_result_ref: Option<String>,
|
provider_alias: Option<String>,
|
partial_count: Option<u64>,
|
fallback_reason: Option<String>,
|
fallback_stage: Option<String>,
|
}
|
|
#[cfg(test)]
|
mod tests {
|
use super::*;
|
use std::{
|
io::{Read, Write},
|
net::TcpListener,
|
sync::mpsc as std_mpsc,
|
thread,
|
};
|
|
#[test]
|
fn chunker_aggregates_20ms_frames_into_200ms_chunks_and_flushes_tail() {
|
let mut chunker = Pcm16kChunker::new(200);
|
let first = chunker.push_48k(&vec![7; 9_600]);
|
let second = chunker.push_48k(&vec![9; 4_800]);
|
let tail = chunker.flush();
|
|
assert_eq!(1, first.len());
|
assert_eq!(1, first[0].seq);
|
assert_eq!(3_200, first[0].samples.len());
|
assert!(second.is_empty());
|
assert_eq!(2, tail.as_ref().map(|chunk| chunk.seq).unwrap_or_default());
|
assert_eq!(
|
1_600,
|
tail.as_ref()
|
.map(|chunk| chunk.samples.len())
|
.unwrap_or_default()
|
);
|
assert_eq!(2, chunker.chunk_count);
|
assert_eq!(9_600, chunker.audio_bytes);
|
}
|
|
#[test]
|
fn session_start_uses_canonical_nonce_hash_and_audio_contract() {
|
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");
|
|
assert_eq!("session_start", value["event"]);
|
assert_eq!(sha12("nonce-001"), value["runtimeSessionNonceHash"]);
|
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!(
|
"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]
|
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]
|
async fn upload_streams_canonical_ndjson_and_reads_final_response() {
|
let response = json!({
|
"code": 0,
|
"data": {
|
"status": "final",
|
"asrResultRef": "asr_rt_fixture",
|
"providerAlias": "fixture",
|
"partialCount": 2
|
}
|
})
|
.to_string();
|
let (url, captured, server) = spawn_http_fixture(response);
|
let mut upload = RealtimeAsrUpload::start(
|
Client::new(),
|
fixture_config(url),
|
"call-001",
|
"trace-001",
|
"turn-0001",
|
&vec![1; 9_600],
|
None,
|
)
|
.expect("start upload");
|
upload
|
.push_48k_samples(&vec![2; 9_600])
|
.expect("push audio");
|
|
let outcome = upload.finish(400, "silence").await.expect("finish upload");
|
let request = captured.recv().expect("captured request");
|
server.join().expect("fixture server");
|
|
assert_eq!("final", outcome.status);
|
assert_eq!(Some("asr_rt_fixture"), outcome.asr_result_ref.as_deref());
|
assert_eq!(2, outcome.chunk_count);
|
assert_eq!(12_800, outcome.audio_bytes);
|
assert!(
|
request
|
.headers
|
.contains("x-cv-runtime-session-nonce: nonce-001")
|
);
|
let events = request
|
.body
|
.lines()
|
.map(|line| serde_json::from_str::<serde_json::Value>(line).expect("event json"))
|
.collect::<Vec<_>>();
|
assert_eq!(5, events.len());
|
assert_eq!("session_start", events[0]["event"]);
|
assert_eq!(1, events[1]["chunkSeq"]);
|
assert_eq!(2, events[2]["chunkSeq"]);
|
assert_eq!("vad_speech_end", events[3]["event"]);
|
assert_eq!("finish", events[4]["event"]);
|
}
|
|
#[tokio::test]
|
async fn cancel_ends_stream_without_finish_event() {
|
let response = json!({"code": 0, "data": {"status": "cancelled"}}).to_string();
|
let (url, captured, server) = spawn_http_fixture(response);
|
let upload = RealtimeAsrUpload::start(
|
Client::new(),
|
fixture_config(url),
|
"call-002",
|
"trace-002",
|
"turn-0002",
|
&vec![1; 9_600],
|
None,
|
)
|
.expect("start upload");
|
|
upload.cancel("call_end").await;
|
let request = captured.recv().expect("captured request");
|
server.join().expect("fixture server");
|
|
assert!(request.body.contains("\"event\":\"cancel\""));
|
assert!(!request.body.contains("\"event\":\"finish\""));
|
}
|
|
fn fixture_config(url: String) -> RealtimeAsrConfig {
|
RealtimeAsrConfig {
|
enabled: true,
|
url: Some(url),
|
runtime_token: Some("token-001".to_string()),
|
runtime_session_nonce: Some("nonce-001".to_string()),
|
chunk_duration_ms: 200,
|
}
|
}
|
|
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,
|
}
|
|
fn spawn_http_fixture(
|
response_body: String,
|
) -> (
|
String,
|
std_mpsc::Receiver<CapturedRequest>,
|
thread::JoinHandle<()>,
|
) {
|
let listener = TcpListener::bind("127.0.0.1:0").expect("bind fixture server");
|
let address = listener.local_addr().expect("fixture address");
|
let (sender, receiver) = std_mpsc::channel();
|
let server = thread::spawn(move || {
|
let (mut stream, _) = listener.accept().expect("accept request");
|
stream
|
.set_read_timeout(Some(Duration::from_secs(3)))
|
.expect("read timeout");
|
let mut request = Vec::new();
|
let mut buffer = [0u8; 4096];
|
loop {
|
let read = stream.read(&mut buffer).expect("read request");
|
if read == 0 {
|
break;
|
}
|
request.extend_from_slice(&buffer[..read]);
|
if request.windows(5).any(|window| window == b"0\r\n\r\n") {
|
break;
|
}
|
}
|
let header_end = request
|
.windows(4)
|
.position(|window| window == b"\r\n\r\n")
|
.expect("request headers");
|
let headers = String::from_utf8_lossy(&request[..header_end]).to_ascii_lowercase();
|
let body = decode_chunked_body(&request[header_end + 4..]);
|
sender
|
.send(CapturedRequest {
|
headers,
|
body: String::from_utf8(body).expect("utf8 ndjson"),
|
})
|
.expect("capture request");
|
let response = format!(
|
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
|
response_body.len(),
|
response_body
|
);
|
stream
|
.write_all(response.as_bytes())
|
.expect("write response");
|
});
|
(
|
format!("http://{address}/runtime/asr/realtime"),
|
receiver,
|
server,
|
)
|
}
|
|
fn decode_chunked_body(mut input: &[u8]) -> Vec<u8> {
|
let mut body = Vec::new();
|
loop {
|
let line_end = input
|
.windows(2)
|
.position(|window| window == b"\r\n")
|
.expect("chunk size line");
|
let size = usize::from_str_radix(
|
std::str::from_utf8(&input[..line_end]).expect("chunk size utf8"),
|
16,
|
)
|
.expect("chunk size");
|
input = &input[line_end + 2..];
|
if size == 0 {
|
break;
|
}
|
body.extend_from_slice(&input[..size]);
|
input = &input[size + 2..];
|
}
|
body
|
}
|
}
|
