backend/lm-livekit-helper.git

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

			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")]
			async fn main() -> Result<()> {
			init_tracing();
			@@ -914,33 +933,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,19 +1056,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 +1106,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 +1118,7 @@
			if publish.await.is_ok() {
			acknowledged_probe_sequences.insert(probe.sequence);
			}
			Some(result == "observed")
			}

			async fn handle_finished_turn(
			@@ -4331,7 +4366,7 @@
			mod tests {
			use super::*;
			use std::{
			collections::HashSet,
			collections::{HashMap, HashSet},
			io::{Read, Write},
			net::TcpListener,
			sync::{
			@@ -4342,6 +4377,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() {
			@@ -4998,4 +5089,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"
			);
			}
			}