fix(helper): bind ASR ingress to fixture generation

cai

2026-08-25 caeed9acdc676aa10f788e81edf45892b8cdcc8e

 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,122 @@
    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,
    pub(crate) input_generation: u64,
    pub(crate) language: Option<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);
        let generation = attributes.get("inputGeneration").map(String::as_str);
        match (source, sequence) {
            (None, None) => "ordinary_mic_absent",
            (Some("controlled_fixture"), Some(sequence)) if valid_sequence(sequence) => {
                match generation {
                    Some(value) if valid_generation(value) => "controlled_fixture_bound",
                    None => "participant_attributes_absent",
                    Some(_) => "participant_attributes_invalid",
                }
            }
            (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) => {
                let generation = attributes
                    .get("inputGeneration")
                    .ok_or("missing_generation")?
                    .parse::<u64>()
                    .ok()
                    .filter(|value| *value > 0)
                    .ok_or("invalid_generation")?;
                let language = attributes.get("language").map(String::as_str);
                if let Some(language) = language {
                    if !valid_language(language) {
                        return Err("invalid_language");
                    }
                }
                Ok(Some(Self {
                    input_source_category: "controlled_fixture".to_string(),
                    client_fixture_sequence: sequence.to_string(),
                    input_generation: generation,
                    language: language.map(str::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, '-' | '_' | '.'))
}

fn valid_generation(value: &str) -> bool {
    value.parse::<u64>().is_ok_and(|generation| generation > 0)
}

fn valid_language(value: &str) -> bool {
    matches!(value, "ja-JP" | "zh-CN")
}

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 +168,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 +200,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 +247,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 +285,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 +425,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 +443,20 @@
            "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["inputGeneration"] = json!(metadata.input_generation);
        if let Some(language) = metadata.language.as_deref() {
            line["language"] = json!(language);
        }
    }
    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 +557,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 +566,290 @@
        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(),
        );
        attributes.insert("inputGeneration".to_string(), "1".to_string());
        attributes.insert("language".to_string(), "ja-JP".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(),
                input_generation: 1,
                language: Some("ja-JP".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())
        );
        attributes.insert(
            "clientFixtureSequence".to_string(),
            "fixture-01".to_string(),
        );
        attributes.insert("language".to_string(), "ja-jp".to_string());
        assert_eq!(
            Err("invalid_language"),
            AudioIngressMetadata::from_participant(&attributes)
        );
        attributes.insert("language".to_string(), "ja-JP".to_string());
        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(),
            input_generation: 1,
            language: Some("ja-JP".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!(1, with_metadata["inputGeneration"]);
        assert_eq!("ja-JP", with_metadata["language"]);
        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(),
            input_generation: 1,
            language: Some("zh-CN".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_eq!("zh-CN", next_value["language"]);
        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(),
        );
        attributes.insert("inputGeneration".to_string(), "1".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 controlled_fixture_session_requires_current_generation_binding() {
        let attributes = HashMap::from([
            (
                "inputSourceCategory".to_string(),
                "controlled_fixture".to_string(),
            ),
            (
                "clientFixtureSequence".to_string(),
                "fixture-01".to_string(),
            ),
        ]);
        assert_eq!(
            Err("missing_generation"),
            AudioIngressMetadata::from_participant(&attributes)
        );
    }

    #[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 +872,7 @@
            "trace-001",
            "turn-0001",
            &vec![1; 9_600],
            None,
        )
        .expect("start upload");
        upload
@@ -479,6 +916,7 @@
            "trace-002",
            "turn-0002",
            &vec![1; 9_600],
            None,
        )
        .expect("start upload");

@@ -500,6 +938,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,