feat(asr): record audio ingress origin status

cai

2026-08-11 dbb73f5406a884a108d2498bb40501e9a876c52d

 src/asr_realtime.rs

@@ -37,6 +37,20 @@
}

impl AudioIngressMetadata {
    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 from_participant(
        attributes: &HashMap<String, String>,
    ) -> Result<Option<Self>, &'static str> {
@@ -93,6 +107,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,
@@ -325,7 +364,7 @@
    samples: Vec<i16>,
}

fn session_start_line(
pub(crate) fn session_start_line(
    call_id: &str,
    trace_id: &str,
    turn_id: &str,
@@ -348,6 +387,10 @@
        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)
}

@@ -458,6 +501,7 @@
        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]
@@ -480,6 +524,10 @@
            }),
            metadata
        );
        assert_eq!(
            "controlled_fixture_bound",
            AudioIngressMetadata::origin_status(&attributes)
        );

        attributes.insert(
            "clientFixtureSequence".to_string(),
@@ -492,6 +540,23 @@
        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())
        );
    }

@@ -507,6 +572,7 @@
        )
        .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",
@@ -520,6 +586,10 @@
        .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(),
@@ -540,6 +610,66 @@
        );
    }

    #[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");