diff --git a/crates/rutster-tap/src/tap_client.rs b/crates/rutster-tap/src/tap_client.rs index ed906d4..5c7d39f 100644 --- a/crates/rutster-tap/src/tap_client.rs +++ b/crates/rutster-tap/src/tap_client.rs @@ -103,6 +103,7 @@ pub async fn run_tap_client( tx_audio_out: mpsc::Sender, tx_function_call: mpsc::Sender, rx_function_call_output: &mut mpsc::Receiver, + tx_advisory: mpsc::Sender, metrics: Arc, close: &mut oneshot::Receiver<()>, ) -> Result<(), TapClientError> @@ -260,9 +261,15 @@ where // silently dropped (v1 is text-JSON only — spec §3.4). if let Ok(text) = msg.into_text() { handle_brain_frame( - &text, &mut last_seq_ingress, &tx_audio_out, - &tx_function_call, &metrics, session_start, - ).await; + &text, + &mut last_seq_ingress, + &tx_audio_out, + &tx_function_call, + &tx_advisory, + &metrics, + session_start, + ) + .await; } } } @@ -325,6 +332,7 @@ async fn handle_brain_frame( last_seq_ingress: &mut Option, tx_audio_out: &mpsc::Sender, tx_function_call: &mpsc::Sender, + tx_advisory: &mpsc::Sender, metrics: &Arc, session_start: Instant, ) { @@ -403,12 +411,27 @@ async fn handle_brain_frame( metrics.unknown_frames.fetch_add(1, Ordering::Relaxed); warn!("unexpected function_call_output from brain; dropping"); } - // Slice-3 advisory — same "logged + counted, not forwarded" posture - // as `Unknown`. The FOB reflex loop in step 4 will act on these; - // slice-3 only pre-paves the wire event. - DecodedPayload::SpeechStarted | DecodedPayload::SpeechStopped => { - metrics.unknown_frames.fetch_add(1, Ordering::Relaxed); - debug!("advisory interruption event observed; not acted on in slice-3"); + // slice-4: advisory events forward to the Reflex via the dedicated + // `advisory_tx` channel. The FOB reflex is authoritative: a local + // in-core VAD is the PRIMARY trigger, and the brain's ASR-grade + // advisory is the slower SECONDARY/confirmation trigger (~300 ms + // later). Both sources feed the same advisory mpsc; `Reflex` drains + // them uniformly on the 20 ms tick. + DecodedPayload::SpeechStarted => { + let ev = rutster_media::AdvisoryEvent::SpeechStarted { at: Instant::now() }; + if tx_advisory.try_send(ev).is_err() { + // Channel full → drop + observe (hot-path policy). The + // Reflex counts dropped advisories in its own metrics. + metrics.outbound_dropped.fetch_add(1, Ordering::Relaxed); + warn!("advisory SpeechStarted dropped (advisory_tx full)"); + } + } + DecodedPayload::SpeechStopped => { + let ev = rutster_media::AdvisoryEvent::SpeechStopped { at: Instant::now() }; + if tx_advisory.try_send(ev).is_err() { + metrics.outbound_dropped.fetch_add(1, Ordering::Relaxed); + warn!("advisory SpeechStopped dropped (advisory_tx full)"); + } } DecodedPayload::ToolsUpdate(_) => { metrics.unknown_frames.fetch_add(1, Ordering::Relaxed); @@ -482,6 +505,7 @@ mod tests { async fn handle_brain_frame_forwards_function_call_to_side_channel() { let (tx_fc, mut rx_fc) = mpsc::channel::(8); let (tx_audio_out, _rx_audio_out) = mpsc::channel::(8); + let (tx_advisory, _rx_advisory) = mpsc::channel::(8); let metrics = Arc::new(TapMetrics::new()); // Build a wire function_call frame: id="call-1", name="hangup", args={}. @@ -493,6 +517,7 @@ mod tests { &mut last_seq, &tx_audio_out, &tx_fc, + &tx_advisory, &metrics, Instant::now(), ) @@ -511,18 +536,18 @@ mod tests { assert_eq!(last_seq, Some(1)); } - /// slice-3 spec §5.2 — the *advisory* interrupt events (`speech_started` - /// /`speech_stopped`) and `tools.update` are observed (logged + counted) - /// but do NOT flow through the function_call side-channel (only - /// `function_call` does — that's the only event with a binary-side - /// disposal). This pins that boundary: an advisory event must NOT - /// produce a `FunctionCallEvent` even with the channel plumbed. + /// slice-4: `speech_started`/`speech_stopped` advisories now flow to the + /// dedicated `advisory_tx` side-channel for the Reflex to drain, and + /// STILL do NOT flow through the function_call side-channel (different + /// bus). This pins that boundary. #[tokio::test] - async fn advisory_events_are_logged_not_forwarded_to_function_call_channel() { + async fn advisory_events_forwarded_to_advisory_channel_only() { let (tx_fc, mut rx_fc) = mpsc::channel::(8); let (tx_audio_out, _rx_audio_out) = mpsc::channel::(8); + let (tx_advisory, mut rx_advisory) = mpsc::channel::(8); let metrics = Arc::new(TapMetrics::new()); + // speech_started forwards to advisory_tx. let wire = crate::protocol::encode_speech_started(2, 200).unwrap(); let mut last_seq: Option = None; handle_brain_frame( @@ -530,23 +555,48 @@ mod tests { &mut last_seq, &tx_audio_out, &tx_fc, + &tx_advisory, &metrics, Instant::now(), ) .await; - - // No FunctionCallEvent forwarded — the channel stays empty. Pick a - // tight bounded receive so the test fails fast if a future refactor - // starts forwarding advisory events here. + let advisory = tokio::time::timeout(Duration::from_millis(200), rx_advisory.recv()) + .await + .expect("advisory drained within 200ms") + .expect("channel not closed"); + assert!(matches!( + advisory, + rutster_media::AdvisoryEvent::SpeechStarted { .. } + )); + // function_call channel stays empty. assert!( tokio::time::timeout(Duration::from_millis(50), rx_fc.recv()) .await .is_err(), "no FunctionCallEvent expected for advisory events" ); - // The advisory event IS still observed via metrics (seq gap tracking - // + the unknown-slot counter remains 0 — speech_started is now a - // known payload variant). assert_eq!(last_seq, Some(2)); + + // speech_stopped forwards to advisory_tx. + let wire = crate::protocol::encode_speech_stopped(3, 300).unwrap(); + handle_brain_frame( + &wire, + &mut last_seq, + &tx_audio_out, + &tx_fc, + &tx_advisory, + &metrics, + Instant::now(), + ) + .await; + let advisory = tokio::time::timeout(Duration::from_millis(200), rx_advisory.recv()) + .await + .expect("advisory drained within 200ms") + .expect("channel not closed"); + assert!(matches!( + advisory, + rutster_media::AdvisoryEvent::SpeechStopped { .. } + )); + assert_eq!(last_seq, Some(3)); } } diff --git a/crates/rutster/src/tap_engine.rs b/crates/rutster/src/tap_engine.rs index bb8a580..4ed1765 100644 --- a/crates/rutster/src/tap_engine.rs +++ b/crates/rutster/src/tap_engine.rs @@ -160,6 +160,12 @@ pub fn spawn_tap_engine( mpsc::channel::(TAP_MPSC_CAPACITY); let (tx_function_call_output, rx_function_call_output) = mpsc::channel::(TAP_MPSC_CAPACITY); + // slice-4: advisory channel from TapEngine → Reflex (media thread). + // The media thread will own the channel and clone the sender to both + // the engine (here) and the LocalVadReflex wrapper. For now we create + // the sender internally; Task 5 revises the public signature to accept + // it as a parameter. + let (tx_advisory, _advisory_rx) = mpsc::channel::(16); let metrics = TapMetrics::new(); // slice-3 §6.2: per-channel tool registry. The engine constructs it @@ -193,6 +199,7 @@ pub fn spawn_tap_engine( flush_tx, tx_function_call, rx_function_call_output, + tx_advisory, metrics, ) .await; @@ -246,6 +253,7 @@ async fn run_engine_loop( flush_tx: mpsc::Sender<()>, tx_function_call: mpsc::Sender, mut rx_function_call_output: mpsc::Receiver, + tx_advisory: mpsc::Sender, metrics: Arc, ) { let mut backoff = Backoff::default(); @@ -290,6 +298,7 @@ async fn run_engine_loop( tx_audio_out.clone(), tx_function_call.clone(), &mut rx_function_call_output, + tx_advisory.clone(), metrics.clone(), &mut close, )