fix(helper): gate audio observer on fixture ack

cai

2026-08-11 949be8e5f1a75286b8b517dacda76ed27646e77b

 src/main.rs

@@ -173,6 +173,10 @@
    Ok(())
}

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

#[tokio::main(flavor = "multi_thread")]
async fn main() -> Result<()> {
    init_tracing();
@@ -914,7 +918,26 @@
                    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();
                let probe_result = process_controlled_fixture_probe(
                    &mut pending_probe,
                    &mut acknowledged_probe_sequences,
                    Some(&participant_for_probe),
                    &sink,
                    &call_id,
                    &trace_id,
                    user_participant_identity.as_deref(),
                )
                .await;
                if !controlled_fixture_observer_gate(pending_sequence.is_some(), probe_result) {
                    warn!(
                        call_id = %call_id,
                        trace_id = %trace_id,
                        "runtime helper withheld audio observer until controlled fixture ACK"
                    );
                    continue;
                }
                spawn_user_audio_frame_observer(
                    track,
                    call_id.clone(),
@@ -931,16 +954,6 @@
                    participant,
                );
                current_user_participant = Some(participant_for_probe);
                process_controlled_fixture_probe(
                    &mut pending_probe,
                    &mut acknowledged_probe_sequences,
                    current_user_participant.as_ref(),
                    &sink,
                    &call_id,
                    &trace_id,
                    user_participant_identity.as_deref(),
                )
                .await;
            }
            RoomEvent::DataReceived {
                payload,
@@ -1022,19 +1035,19 @@
    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 {
@@ -1072,7 +1085,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 +1097,7 @@
    if publish.await.is_ok() {
        acknowledged_probe_sequences.insert(probe.sequence);
    }
    Some(result == "observed")
}

async fn handle_finished_turn(
@@ -4998,4 +5012,12 @@
            "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));
    }
}