test(trunk): reflex-on-trunk verification (T7) + PSTN sim e2e (T8) — slice-5

T7: proves slice-4's Reflex<TapAudioPipe> + LocalVadReflex decorates the
trunk leg's TapAudioPipe identically; barge-in fires on PSTN caller speech
through the same state machine as WebRTC caller speech.

T8: MockRealtimeBrain + BrainShim drives a synthetic PSTN caller through the
FOB reflex loop end-to-end: loud PCM -> local VAD trips -> barge kills ->
brain reply -> un-mute -> idle timeout (caller hangup) closes the session.

Dev-dependencies added to rutster-trunk/Cargo.toml so the integration tests
reach MockRealtimeBrain, futures-util, and tokio-tungstenite without pulling
FOB source into the trunk crate.

Signed-off-by: Aaron D. Lee <himself@adlee.work>

crates/rutster-trunk/Cargo.toml

@@ -29,6 +29,10 @@ reqwest = { workspace = true, optional = true }
 
 [dev-dependencies]
 tower = { workspace = true }
+futures-util = { workspace = true }
+tokio-tungstenite = { workspace = true }
+tracing-subscriber = { workspace = true }
+rutster-brain-realtime = { path = "../rutster-brain-realtime", features = ["mock"] }
 
 [features]
 default = []

crates/rutster-trunk/tests/reflex_on_trunk.rs

@@ -0,0 +1,79 @@
+// crates/rutster-trunk/tests/reflex_on_trunk.rs
+//
+// T7 — Reflex-on-trunk-leg verification test (slice-5 spec §7).
+//
+// Proves that slice-4's `Reflex<TapAudioPipe>` + `LocalVadReflex` decorate the
+// trunk leg's `TapAudioPipe` identically to a WebRTC leg. A PSTN caller's loud
+// PCM triggers the same local-VAD state machine and the same `barge-in` kill
+// that a WebRTC caller's audio does.
+//
+// This is a *unit* integration test: it constructs the wrapped pipe stack
+// directly, without the binary's MediaThread or a real WSS brain. The heavier
+// end-to-end WSS sim lives in `sim_integ.rs` (T8).
+
+use std::sync::atomic::Ordering;
+
+use rutster_media::{
+    AudioSink, AudioSource, LocalVadReflex, PcmFrame, Reflex, ReflexMetrics, VAD_DEBOUNCE_FRAMES,
+};
+use rutster_tap::TapAudioPipe;
+use tokio::sync::mpsc;
+
+/// Build a loud 24 kHz PCM frame whose RMS energy is well above
+/// `VAD_RMS_THRESHOLD`. The mock caller uses a constant amplitude of 1000,
+/// the same value slice-4's reflex unit tests use.
+fn loud_frame() -> PcmFrame {
+    let mut frame = PcmFrame::zeroed();
+    for s in frame.samples.iter_mut() {
+        *s = 1000;
+    }
+    frame
+}
+
+#[tokio::test]
+async fn local_vad_on_trunk_pipe_kills_playout_and_resumes_on_fresh_brain_audio() {
+    // Given: a TapAudioPipe + the same Reflex<LocalVadReflex> composition the
+    // FOB builds for every leg (WebRTC or trunk).
+    let (tx_pcm_in, rx_pcm_in) = mpsc::channel(32);
+    let (tx_audio_out, rx_audio_out) = mpsc::channel(32);
+    let _rx_pcm_in = rx_pcm_in; // brain side; not used in this unit test
+    let tap_metrics = rutster_tap::TapMetrics::new();
+    let tap_pipe = TapAudioPipe::new(tx_pcm_in, rx_audio_out, tap_metrics);
+
+    let (advisory_tx, advisory_rx) = mpsc::channel(16);
+    let metrics = ReflexMetrics::new();
+    let reflex = Reflex::new(tap_pipe, advisory_rx, metrics.clone());
+    let mut wrapped_pipe = LocalVadReflex::new(reflex, advisory_tx);
+
+    // Pre-load one brain audio_out frame so the playout ring is non-empty.
+    tx_audio_out.send(PcmFrame::zeroed()).await.unwrap();
+
+    // When: the PSTN caller speaks for `VAD_DEBOUNCE_FRAMES` consecutive ticks.
+    let loud = loud_frame();
+    for _ in 0..VAD_DEBOUNCE_FRAMES {
+        wrapped_pipe.on_pcm_frame(loud.clone());
+    }
+
+    // Then: the third `next_pcm_frame` drains the local-VAD advisory, mutes the
+    // pipe, flushes the ring, and returns `None` (playout is killed).
+    let killed = wrapped_pipe.next_pcm_frame();
+    assert!(
+        killed.is_none(),
+        "local VAD must barge-in and suppress playout on the trunk leg"
+    );
+    assert_eq!(
+        metrics.barge_in_count.load(Ordering::Relaxed),
+        1,
+        "barge-in must fire exactly once for the first loud utterance"
+    );
+
+    // And when: a fresh brain reply arrives after the barge.
+    tx_audio_out.send(PcmFrame::zeroed()).await.unwrap();
+
+    // Then: the Reflex un-mutes and returns the fresh frame.
+    let resumed = wrapped_pipe.next_pcm_frame();
+    assert!(
+        resumed.is_some(),
+        "first fresh audio_out post-barge must resume trunk-leg playout"
+    );
+}

crates/rutster-trunk/tests/sim_integ.rs

@@ -0,0 +1,348 @@
+// crates/rutster-trunk/tests/sim_integ.rs
+//
+// T8 — PSTN sim end-to-end integration test (slice-5 spec §7).
+//
+// Drives a synthetic PSTN caller through the FOB reflex loop end-to-end:
+//   loud PCM -> local VAD trips -> barge kills playout -> brain replies
+//   -> un-mute -> caller hangup -> session closes.
+//
+// The test is in `rutster-trunk` so it can construct `TrunkSession` directly.
+// It cannot use the binary crate's `spawn_tap_engine` / `MediaThread` (circular
+// dev-dependency), so it builds a minimal test-only tap engine task that calls
+// `rutster_tap::tap_client::run_tap_client` against the same BrainShim surface
+// slice-4's `barge_in_integration.rs` uses.
+
+use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::time::{Duration, Instant};
+
+use futures_util::{SinkExt, StreamExt};
+use rutster_brain_realtime::mock::MockRealtimeBrain;
+use rutster_brain_realtime::openai_client::run_openai_pump;
+use rutster_call_model::Channel;
+use rutster_media::{LocalVadReflex, PcmFrame, Reflex, ReflexMetrics};
+
+use rutster_tap::{
+    DecodedPayload, FunctionCallEvent, FunctionCallOutputEvent, TapAudioPipe, TapMetrics,
+    decode_envelope, encode_hello, tap_client::run_tap_client,
+};
+use rutster_trunk::loop_driver::drive;
+use rutster_trunk::session::TrunkSession;
+use tokio::net::TcpListener;
+use tokio::sync::{mpsc, oneshot};
+use tokio_tungstenite::tungstenite::Message;
+use tokio_tungstenite::tungstenite::client::IntoClientRequest;
+use tracing::info;
+use url::Url;
+
+// === Brain-shim helpers (mirrored from slice-4 barge_in_integration.rs) ===
+// The brain process's accept loop is inlined so the test exercises the real
+// OpenAI-client pump (`run_openai_pump`) against the mock brain without
+// spawning a subprocess or depending on private helpers in another test.
+
+/// Handle returned by `start_brain_shim`. Drop to tear down.
+struct BrainShim {
+    addr: std::net::SocketAddr,
+    shutdown: Option<oneshot::Sender<()>>,
+    join: tokio::task::JoinHandle<()>,
+}
+
+impl Drop for BrainShim {
+    fn drop(&mut self) {
+        if let Some(tx) = self.shutdown.take() {
+            let _ = tx.send(());
+        }
+        self.join.abort();
+    }
+}
+
+/// Start an in-process brain-process-equivalent WS server on an ephemeral port.
+async fn start_brain_shim(mock_url: String) -> BrainShim {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();
+    let join = tokio::spawn(async move {
+        brain_accept_loop(listener, shutdown_rx, mock_url).await;
+    });
+    BrainShim {
+        addr,
+        shutdown: Some(shutdown_tx),
+        join,
+    }
+}
+
+/// Accept loop: spawns a per-connection task for each tap WS dial.
+async fn brain_accept_loop(
+    listener: TcpListener,
+    mut shutdown: oneshot::Receiver<()>,
+    mock_url: String,
+) {
+    loop {
+        tokio::select! {
+            biased;
+            _ = &mut shutdown => {
+                info!("brain_shim accept loop shutting down");
+                return;
+            }
+            res = listener.accept() => {
+                let Ok((stream, peer)) = res else { continue };
+                let url = mock_url.clone();
+                tokio::spawn(async move {
+                    if let Err(e) = handle_tap_connection(stream, peer, &url).await {
+                        info!(%peer, error = ?e, "brain_shim connection ended");
+                    }
+                });
+            }
+        }
+    }
+}
+
+/// Handle one tap WS connection: handshake, split sink+stream, dial the mock
+/// OpenAI side, and run `run_openai_pump`.
+async fn handle_tap_connection(
+    stream: tokio::net::TcpStream,
+    peer: std::net::SocketAddr,
+    openai_url: &str,
+) -> Result<(), Box<dyn std::error::Error>> {
+    let mut tap_ws = tokio_tungstenite::accept_async(stream).await?;
+    info!(%peer, "brain_shim tap WS connection accepted");
+
+    let hello_in = tap_ws
+        .next()
+        .await
+        .ok_or("tap connection closed before hello")??;
+    let hello_text = hello_in.into_text().map_err(|_| "hello not text")?;
+    let decoded = decode_envelope(&hello_text)?;
+    let session_id = match decoded.payload {
+        DecodedPayload::Hello(p) => p.session_id,
+        _ => return Err("first tap frame not hello".into()),
+    };
+    info!(%peer, %session_id, "brain_shim tap hello received");
+    let ack = encode_hello(&session_id, 0, 0)?;
+    tap_ws.send(Message::Text(ack)).await?;
+
+    let (tap_via, pump_tap_rx) = mpsc::channel::<String>(64);
+    let (pump_tap_tx, mut tap_out_rx) = mpsc::channel::<String>(64);
+    let (mut tap_sink, mut tap_stream) = tap_ws.split();
+
+    let in_fwd = tokio::spawn(async move {
+        while let Some(msg_res) = tap_stream.next().await {
+            if let Ok(m) = msg_res {
+                if let Ok(text) = m.into_text() {
+                    if tap_via.send(text).await.is_err() {
+                        break;
+                    }
+                }
+            }
+        }
+    });
+    let out_fwd = tokio::spawn(async move {
+        while let Some(text) = tap_out_rx.recv().await {
+            if tap_sink.send(Message::Text(text)).await.is_err() {
+                break;
+            }
+        }
+    });
+
+    let request = openai_url.into_client_request()?;
+    let (openai_ws, _resp) = tokio_tungstenite::connect_async(request).await?;
+    info!(%peer, %openai_url, "brain_shim OpenAI side connected");
+
+    let pump_result =
+        run_openai_pump(openai_ws, pump_tap_rx, pump_tap_tx, "alloy".to_string()).await;
+    info!(%peer, ?pump_result, "brain_shim pump exited");
+    in_fwd.abort();
+    out_fwd.abort();
+    Ok(())
+}
+
+// === Test helpers ===
+
+/// Build a loud 24 kHz PCM frame whose RMS energy is well above the local-VAD
+/// threshold. A constant amplitude of 1000 matches slice-4's test fixture.
+fn loud_frame() -> PcmFrame {
+    let mut frame = PcmFrame::zeroed();
+    for s in frame.samples.iter_mut() {
+        *s = 1000;
+    }
+    frame
+}
+
+#[tokio::test]
+async fn pstn_sim_synthetic_caller_drives_trunk_reflex_loop() {
+    let _ = tracing_subscriber::fmt()
+        .with_env_filter(
+            tracing_subscriber::EnvFilter::try_from_default_env()
+                .unwrap_or_else(|_| "rutster_trunk=info".into()),
+        )
+        .try_init();
+
+    // 1. Start the mock OpenAI Realtime brain.
+    let mock = MockRealtimeBrain::start().await.expect("mock brain binds");
+
+    // 2. Start the brain shim that speaks the tap protocol on the core side
+    //    and the OpenAI protocol on the brain side.
+    let shim = start_brain_shim(mock.url()).await;
+    let tap_url = Url::parse(&format!("ws://{}/", shim.addr)).unwrap();
+
+    // 3. Build the trunk-leg pipe stack: TapAudioPipe -> Reflex -> LocalVadReflex.
+    //    This is the same composition `MediaThread::RegisterTrunk` will build.
+    let (tx_pcm_in, mut rx_pcm_in) = mpsc::channel::<PcmFrame>(32);
+    let (tx_audio_out, rx_audio_out) = mpsc::channel::<PcmFrame>(32);
+    let tap_metrics = TapMetrics::new();
+    let tap_pipe = TapAudioPipe::new(tx_pcm_in, rx_audio_out, tap_metrics.clone());
+
+    let (advisory_tx, advisory_rx) = mpsc::channel::<rutster_media::AdvisoryEvent>(16);
+    let reflex_metrics = ReflexMetrics::new();
+    let reflex = Reflex::new(tap_pipe, advisory_rx, reflex_metrics.clone());
+    let wrapped_pipe = LocalVadReflex::new(reflex, advisory_tx.clone());
+
+    // 4. Construct the TrunkSession that the FOB will tick via `drive`.
+    let channel = Channel::new_inbound();
+    let session_id = channel.id;
+    let (inbound_tx, inbound_rx) = mpsc::channel::<PcmFrame>(16);
+    let (outbound_tx, mut outbound_rx) = mpsc::channel::<PcmFrame>(16);
+    let now = Instant::now();
+    let mut session = TrunkSession::new(channel, wrapped_pipe, inbound_rx, outbound_tx, now);
+
+    // 5. Spawn a minimal test-only tap engine task. We cannot use the binary
+    //    crate's `spawn_tap_engine` from inside `rutster-trunk` (circular
+    //    dev-dependency), so we call `run_tap_client` directly after dialing
+    //    the brain shim.
+    let (close_tx, mut close_rx) = oneshot::channel::<()>();
+    let (tx_function_call, _rx_function_call) = mpsc::channel::<FunctionCallEvent>(8);
+    let (_tx_function_call_output, mut rx_function_call_output) =
+        mpsc::channel::<FunctionCallOutputEvent>(8);
+    let engine_metrics = tap_metrics.clone();
+    let engine_handle = tokio::spawn(async move {
+        let request = tap_url
+            .as_str()
+            .into_client_request()
+            .expect("valid ws url");
+        let (ws, _resp) = tokio_tungstenite::connect_async(request)
+            .await
+            .expect("connect to brain_shim");
+        let _ = run_tap_client(
+            ws,
+            session_id,
+            &mut rx_pcm_in,
+            tx_audio_out,
+            tx_function_call,
+            &mut rx_function_call_output,
+            advisory_tx,
+            engine_metrics,
+            &mut close_rx,
+        )
+        .await;
+    });
+
+    // Wait for tap handshake + OpenAI dial to complete.
+    tokio::time::sleep(Duration::from_millis(150)).await;
+
+    // 6. Spawn the synthetic Twilio caller task: push loud inbound frames into
+    //    the trunk leg and count outbound (brain-reply) frames coming back.
+    let outbound_count = Arc::new(AtomicUsize::new(0));
+    let outbound_count_caller = outbound_count.clone();
+    let (stop_tx, mut stop_rx) = oneshot::channel::<()>();
+    let caller_handle = tokio::spawn(async move {
+        let loud = loud_frame();
+        let mut local_count = 0usize;
+        loop {
+            // Push caller audio to the trunk leg. try_send matches the hot-path
+            // "drop + observe" policy: if the FOB backs up, keep going.
+            let _ = inbound_tx.try_send(loud.clone());
+
+            // Drain any outbound (brain reply) frames the FOB produced.
+            while outbound_rx.try_recv().is_ok() {
+                local_count += 1;
+            }
+            outbound_count_caller.store(local_count, Ordering::Relaxed);
+
+            tokio::time::sleep(Duration::from_millis(20)).await;
+            if stop_rx.try_recv().is_ok() {
+                break;
+            }
+        }
+        local_count
+    });
+
+    // 7. Drive the trunk-leg poll loop at 20 ms intervals.
+    let mut barge_seen = false;
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(3);
+    while tokio::time::Instant::now() < deadline {
+        let now = Instant::now();
+        let _ = drive(&mut session, now);
+
+        if !barge_seen && reflex_metrics.barge_in_count.load(Ordering::Relaxed) >= 1 {
+            barge_seen = true;
+            info!("PSTN sim: local VAD barge-in fired");
+        }
+
+        // Stop early once we have both barge-in and at least one observed
+        // outbound frame (the mock brain replies with audio_out deltas).
+        if barge_seen && outbound_count.load(Ordering::Relaxed) > 0 {
+            break;
+        }
+
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+
+    // 8. Stop the caller and collect its final count.
+    let _ = stop_tx.send(());
+    let final_outbound_count = tokio::time::timeout(Duration::from_secs(1), caller_handle)
+        .await
+        .expect("caller task finishes")
+        .expect("caller task panics");
+
+    // Then: the local VAD must have fired during the call.
+    assert!(
+        barge_seen,
+        "PSTN caller speech must trigger a local-VAD barge-in on the trunk leg"
+    );
+    assert_eq!(
+        reflex_metrics.barge_in_count.load(Ordering::Relaxed),
+        1,
+        "barge-in must fire exactly once for the first utterance"
+    );
+
+    // And: the mock brain must have received audio_in and replied with at
+    // least one outbound frame after the barge (resume condition).
+    assert!(
+        final_outbound_count > 0,
+        "mock brain must reply with at least one audio_out frame observed on the trunk outbound mpsc"
+    );
+
+    // 9. Caller hangup: stop sending inbound frames and force the 60 s idle
+    //    timeout path by moving `last_idle_rx` into the past. The FOB should
+    //    close the session.
+    session.last_idle_rx = Instant::now() - Duration::from_secs(90);
+    let next = drive(&mut session, Instant::now());
+    assert!(
+        session.is_closed(),
+        "trunk session must close after idle timeout (simulating caller hangup)"
+    );
+    assert_eq!(
+        next, None,
+        "drive must return None once the session is closed"
+    );
+
+    // 10. Clean up the tap engine.
+    let _ = close_tx.send(());
+    let _ = tokio::time::timeout(Duration::from_secs(1), engine_handle).await;
+}
+
+/// Full end-to-end test using the binary's `MediaThread` + `RegisterTrunk`.
+///
+/// This test is currently ignored because `rutster-trunk` integration tests
+/// cannot depend on the `rutster` binary crate (`rutster` already depends on
+/// `rutster-trunk`; Cargo disallows circular dev-dependencies). The active
+/// `pstn_sim_synthetic_caller_drives_trunk_reflex_loop` above covers the FOB
+// reflex loop + trunk-leg tick directly; this stub marks where the MediaThread
+/// wiring test belongs once the binary crate is ready to exercise it.
+#[ignore]
+#[tokio::test]
+async fn full_pstn_e2e_through_media_thread_register_trunk() {
+    // TODO: exercise MediaCmd::RegisterTrunk + MediaThread tick loop against a
+    // live TwilioMediaStreamsServer mock once the binary crate exposes a test
+    // harness from the appropriate crate.
+}