fix(helper): observe fixture attributes through probe ttl

cai

2026-08-12 9485ced68a81f8880f3b173e62b62b55f0dd80aa

 src/main.rs

@@ -63,7 +63,6 @@
const CONTROLLED_FIXTURE_ACK_TOPIC: &str = "controlled_fixture_attribute_ack";
const CONTROLLED_FIXTURE_PROTOCOL_VERSION: u64 = 1;
const CONTROLLED_FIXTURE_GENERATION: u64 = 1;
const CONTROLLED_FIXTURE_PROBE_RECHECKS: usize = 3;
const CONTROLLED_FIXTURE_PROBE_RECHECK_DELAY: Duration = Duration::from_millis(25);
const CONTROLLED_FIXTURE_PROBE_TTL: Duration = Duration::from_millis(250);

@@ -171,6 +170,59 @@
        return Err("wrong_sequence");
    }
    Ok(())
}

async fn observe_controlled_fixture_attributes<F>(
    expires_at: Instant,
    actual_participant: &str,
    expected_participant: Option<&str>,
    requested_sequence: &str,
    mut read_attributes: F,
) -> Result<(), &'static str>
where
    F: FnMut() -> std::collections::HashMap<String, String>,
{
    loop {
        if Instant::now() > expires_at {
            return Err("timeout");
        }
        let decision = classify_controlled_fixture_attributes(
            actual_participant,
            expected_participant,
            &read_attributes(),
            requested_sequence,
        );
        if decision.is_ok() || !matches!(decision, Err("missing_attributes")) {
            return decision;
        }
        let Some(next_check) = Instant::now().checked_add(CONTROLLED_FIXTURE_PROBE_RECHECK_DELAY)
        else {
            return Err("timeout");
        };
        if next_check > expires_at {
            return Err("timeout");
        }
        sleep(CONTROLLED_FIXTURE_PROBE_RECHECK_DELAY).await;
    }
}

fn controlled_fixture_observer_gate(pending_probe: bool, probe_result: Option<bool>) -> bool {
    !pending_probe || probe_result == Some(true)
}

fn start_observer_after_controlled_fixture_probe<F>(
    pending_probe: bool,
    probe_result: Option<bool>,
    spawn: F,
) -> bool
where
    F: FnOnce(),
{
    if !controlled_fixture_observer_gate(pending_probe, probe_result) {
        return false;
    }
    spawn();
    true
}

#[tokio::main(flavor = "multi_thread")]
@@ -914,33 +966,48 @@
                    track_source = %track_source,
                    "runtime helper user_track_subscribed"
                );
                let pending_sequence = pending_probe.as_ref().map(|probe| probe.sequence.clone());
                let participant_for_probe = participant.clone();
                spawn_user_audio_frame_observer(
                    track,
                    call_id.clone(),
                    trace_id.clone(),
                    participant_alias,
                    track_sid_alias,
                    simple_vad_enabled,
                    simple_vad_config.clone(),
                    vad_enabled_gate.clone(),
                    turn_bridge_config.clone(),
                    http.clone(),
                    sink.clone(),
                    user_participant_identity.clone(),
                    participant,
                );
                current_user_participant = Some(participant_for_probe);
                process_controlled_fixture_probe(
                let probe_result = process_controlled_fixture_probe(
                    &mut pending_probe,
                    &mut acknowledged_probe_sequences,
                    current_user_participant.as_ref(),
                    Some(&participant_for_probe),
                    &sink,
                    &call_id,
                    &trace_id,
                    user_participant_identity.as_deref(),
                )
                .await;
                let observer_started = start_observer_after_controlled_fixture_probe(
                    pending_sequence.is_some(),
                    probe_result,
                    || {
                        spawn_user_audio_frame_observer(
                            track,
                            call_id.clone(),
                            trace_id.clone(),
                            participant_alias,
                            track_sid_alias,
                            simple_vad_enabled,
                            simple_vad_config.clone(),
                            vad_enabled_gate.clone(),
                            turn_bridge_config.clone(),
                            http.clone(),
                            sink.clone(),
                            user_participant_identity.clone(),
                            participant,
                        );
                    },
                );
                if !observer_started {
                    warn!(
                        call_id = %call_id,
                        trace_id = %trace_id,
                        "runtime helper withheld audio observer until controlled fixture ACK"
                    );
                    continue;
                }
                current_user_participant = Some(participant_for_probe);
            }
            RoomEvent::DataReceived {
                payload,
@@ -1022,39 +1089,28 @@
    call_id: &str,
    trace_id: &str,
    expected_participant: Option<&str>,
) {
) -> Option<bool> {
    let Some(probe) = pending_probe.take() else {
        return;
        return None;
    };
    if acknowledged_probe_sequences.contains(&probe.sequence) || Instant::now() > probe.expires_at {
        return;
        return Some(false);
    }
    let Some(participant) = participant else {
        *pending_probe = Some(probe);
        return;
        return None;
    };
    if participant.identity() != probe.sender {
        return;
        return Some(false);
    }
    let mut decision = Err("timeout");
    for attempt in 0..CONTROLLED_FIXTURE_PROBE_RECHECKS {
        if Instant::now() > probe.expires_at {
            break;
        }
        let attributes = participant.attributes();
        decision = classify_controlled_fixture_attributes(
            &participant.identity().to_string(),
            expected_participant,
            &attributes,
            &probe.sequence,
        );
        if decision.is_ok() || !matches!(decision, Err("missing_attributes")) {
            break;
        }
        if attempt + 1 < CONTROLLED_FIXTURE_PROBE_RECHECKS {
            sleep(CONTROLLED_FIXTURE_PROBE_RECHECK_DELAY).await;
        }
    }
    let decision = observe_controlled_fixture_attributes(
        probe.expires_at,
        &participant.identity().to_string(),
        expected_participant,
        &probe.sequence,
        || participant.attributes(),
    )
    .await;
    let (result, input_source_category, reject_reason) = match decision {
        Ok(()) => ("observed", Some("controlled_fixture"), None),
        Err(reason) => ("rejected", None, Some(reason)),
@@ -1072,7 +1128,7 @@
    };
    let payload = match serde_json::to_vec(&ack) {
        Ok(payload) => payload,
        Err(_) => return,
        Err(_) => return Some(false),
    };
    let local_participant = sink.room.local_participant();
    let publish = local_participant.publish_data(DataPacket {
@@ -1084,6 +1140,7 @@
    if publish.await.is_ok() {
        acknowledged_probe_sequences.insert(probe.sequence);
    }
    Some(result == "observed")
}

async fn handle_finished_turn(
@@ -4331,7 +4388,7 @@
mod tests {
    use super::*;
    use std::{
        collections::HashSet,
        collections::{HashMap, HashSet},
        io::{Read, Write},
        net::TcpListener,
        sync::{
@@ -4342,6 +4399,62 @@
        thread,
        time::Duration,
    };

    #[derive(Debug)]
    enum PreAudioOrderEvent {
        DataReceived {
            sender: String,
            sequence: String,
        },
        TrackSubscribed {
            participant: String,
            attributes: HashMap<String, String>,
        },
    }

    fn drive_pre_audio_order_test_seam(events: &[PreAudioOrderEvent]) -> Vec<&'static str> {
        let mut pending_sequence = None;
        let mut effects = Vec::new();
        for event in events {
            match event {
                PreAudioOrderEvent::DataReceived { sender, sequence } if sender == "user-1" => {
                    pending_sequence = Some(sequence.as_str());
                }
                PreAudioOrderEvent::TrackSubscribed {
                    participant,
                    attributes,
                } => {
                    let pending = pending_sequence.is_some();
                    let probe_result = pending_sequence.map(|sequence| {
                        classify_controlled_fixture_attributes(
                            participant,
                            Some("user-1"),
                            attributes,
                            sequence,
                        )
                        .is_ok()
                    });
                    let mut ack_observed = false;
                    let mut observer_started = false;
                    start_observer_after_controlled_fixture_probe(pending, probe_result, || {
                        if pending && probe_result == Some(true) {
                            ack_observed = true;
                        }
                        observer_started = true;
                    });
                    if ack_observed {
                        effects.push("ack_observed");
                    }
                    if observer_started {
                        effects.push("observer_started");
                    }
                    pending_sequence = None;
                }
                PreAudioOrderEvent::DataReceived { .. } => {}
            }
        }
        effects
    }

    #[test]
    fn production_vad_session_boundary_reads_updated_attributes() {
@@ -4950,6 +5063,68 @@
        );
    }

    #[tokio::test]
    async fn production_attribute_observation_accepts_server_visibility_within_probe_ttl() {
        let expected = HashMap::from([
            (
                "inputSourceCategory".to_string(),
                "controlled_fixture".to_string(),
            ),
            (
                "clientFixtureSequence".to_string(),
                "fixture-01".to_string(),
            ),
        ]);
        let mut reads = 0;
        let result = observe_controlled_fixture_attributes(
            Instant::now() + CONTROLLED_FIXTURE_PROBE_TTL,
            "user-1",
            Some("user-1"),
            "fixture-01",
            || {
                reads += 1;
                if reads <= 3 {
                    HashMap::new()
                } else {
                    expected.clone()
                }
            },
        )
        .await;
        assert_eq!(result, Ok(()));
        assert_eq!(reads, 4);
    }

    #[tokio::test]
    async fn production_attribute_observation_rejects_wrong_sequence_without_audio_effect() {
        let attributes = HashMap::from([
            (
                "inputSourceCategory".to_string(),
                "controlled_fixture".to_string(),
            ),
            (
                "clientFixtureSequence".to_string(),
                "fixture-02".to_string(),
            ),
        ]);
        let result = observe_controlled_fixture_attributes(
            Instant::now() + CONTROLLED_FIXTURE_PROBE_TTL,
            "user-1",
            Some("user-1"),
            "fixture-01",
            || attributes.clone(),
        )
        .await;
        let mut observer_starts = 0;
        assert_eq!(result, Err("wrong_sequence"));
        assert!(!start_observer_after_controlled_fixture_probe(
            true,
            Some(result.is_ok()),
            || observer_starts += 1,
        ));
        assert_eq!(observer_starts, 0);
    }

    #[test]
    fn controlled_fixture_ack_payload_is_reliable_and_redacted() {
        let ack = ControlledFixtureAttributeAck {
@@ -4998,4 +5173,76 @@
            "controlled_fixture_attribute_ack"
        );
    }

    #[test]
    fn controlled_fixture_probe_must_be_observed_before_audio_observer() {
        assert!(controlled_fixture_observer_gate(false, None));
        assert!(controlled_fixture_observer_gate(true, Some(true)));
        assert!(!controlled_fixture_observer_gate(true, Some(false)));
        assert!(!controlled_fixture_observer_gate(true, None));
    }

    #[test]
    fn production_event_order_probe_then_track_publishes_ack_before_observer() {
        let attributes = HashMap::from([
            (
                "inputSourceCategory".to_string(),
                "controlled_fixture".to_string(),
            ),
            (
                "clientFixtureSequence".to_string(),
                "fixture-01".to_string(),
            ),
        ]);
        let effects = drive_pre_audio_order_test_seam(&[
            PreAudioOrderEvent::DataReceived {
                sender: "user-1".to_string(),
                sequence: "fixture-01".to_string(),
            },
            PreAudioOrderEvent::TrackSubscribed {
                participant: "user-1".to_string(),
                attributes,
            },
        ]);
        assert_eq!(effects, ["ack_observed", "observer_started"]);
    }

    #[test]
    fn production_event_order_negative_probe_has_no_observer_or_session_effect() {
        let mut invalid = HashMap::new();
        invalid.insert(
            "inputSourceCategory".to_string(),
            "ordinary_mic".to_string(),
        );
        let effects = drive_pre_audio_order_test_seam(&[
            PreAudioOrderEvent::DataReceived {
                sender: "user-1".to_string(),
                sequence: "fixture-01".to_string(),
            },
            PreAudioOrderEvent::TrackSubscribed {
                participant: "user-1".to_string(),
                attributes: invalid,
            },
        ]);
        assert!(effects.is_empty());
    }

    #[test]
    fn production_audio_branch_orders_probe_before_spawn_callsite() {
        let source = include_str!("main.rs");
        let branch = source
            .find("RoomEvent::TrackSubscribed {\n                track: RemoteTrack::Audio")
            .expect("audio TrackSubscribed production branch");
        let branch_source = &source[branch..];
        let probe = branch_source
            .find("let probe_result = process_controlled_fixture_probe")
            .expect("probe must be processed in audio branch");
        let spawn = branch_source
            .find("start_observer_after_controlled_fixture_probe")
            .expect("spawn must use shared order entry");
        assert!(
            probe < spawn,
            "probe must precede shared observer spawn entry"
        );
    }
}