alee/rutster
0
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

feat(tap): forward speech_started/stopped to advisory_tx (slice-4 §3.1)

Browse Source 
The brain's advisory events now flow to the FOB reflex via the dedicated
advisory_tx side-channel (3rd mpsc alongside tx_pcm_in/tx_audio_out).
handle_brain_frame + run_tap_client thread the sender through; the engine
loop constructs the channel internally until Task 5 moves ownership to the
caller.

Signed-off-by: Aaron D. Lee <himself@adlee.work>
This commit is contained in:
A.D.Lee
2026-07-03 09:26:41 -04:00
parent 58919b8bf7
commit 1cb7260f66
2 changed files with 82 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
crates/rutster-tap/src/tap_client.rs
View File

@@ -103,6 +103,7 @@ pub async fn run_tap_client<T>(
tx_audio_out: mpsc::Sender<PcmFrame>,
tx_function_call: mpsc::Sender<FunctionCallEvent>,
rx_function_call_output: &mut mpsc::Receiver<FunctionCallOutputEvent>,
tx_advisory: mpsc::Sender<rutster_media::AdvisoryEvent>,
metrics: Arc<TapMetrics>,
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<u64>,
tx_audio_out: &mpsc::Sender<PcmFrame>,
tx_function_call: &mpsc::Sender<FunctionCallEvent>,
tx_advisory: &mpsc::Sender<rutster_media::AdvisoryEvent>,
metrics: &Arc<TapMetrics>,
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::<FunctionCallEvent>(8);
let (tx_audio_out, _rx_audio_out) = mpsc::channel::<PcmFrame>(8);
let (tx_advisory, _rx_advisory) = mpsc::channel::<rutster_media::AdvisoryEvent>(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::<FunctionCallEvent>(8);
let (tx_audio_out, _rx_audio_out) = mpsc::channel::<PcmFrame>(8);
let (tx_advisory, mut rx_advisory) = mpsc::channel::<rutster_media::AdvisoryEvent>(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<u64> = 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));
}
}

9
crates/rutster/src/tap_engine.rs
View File

@@ -160,6 +160,12 @@ pub fn spawn_tap_engine(
mpsc::channel::<FunctionCallEvent>(TAP_MPSC_CAPACITY);
let (tx_function_call_output, rx_function_call_output) =
mpsc::channel::<FunctionCallOutputEvent>(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::<rutster_media::AdvisoryEvent>(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<FunctionCallEvent>,
mut rx_function_call_output: mpsc::Receiver<FunctionCallOutputEvent>,
tx_advisory: mpsc::Sender<rutster_media::AdvisoryEvent>,
metrics: Arc<TapMetrics>,
) {
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,
)