diff --git a/Cargo.lock b/Cargo.lock index e14c80c..298201e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1287,6 +1287,8 @@ version = "0.0.0" dependencies = [ "axum", "base64", + "futures-util", + "rutster-brain-realtime", "rutster-call-model", "rutster-media", "rutster-tap", @@ -1294,8 +1296,10 @@ dependencies = [ "serde_json", "thiserror 1.0.69", "tokio", + "tokio-tungstenite", "tower", "tracing", + "tracing-subscriber", "url", ] diff --git a/crates/rutster-trunk/Cargo.toml b/crates/rutster-trunk/Cargo.toml index f433896..f766969 100644 --- a/crates/rutster-trunk/Cargo.toml +++ b/crates/rutster-trunk/Cargo.toml @@ -40,3 +40,9 @@ twilio-live = [] # `tower::ServiceExt::oneshot` is the canonical axum router helper for unit tests # (turns a Router into a one-shot Service: `app.oneshot(request).await`). tower = { workspace = true } +# T7/T8 integration tests exercise the trunk reflex stack against the same +# MockRealtimeBrain + tokio-tungstenite surface slice-3/4 used. +futures-util = { workspace = true } +tokio-tungstenite = { workspace = true } +tracing-subscriber = { workspace = true } +rutster-brain-realtime = { path = "../rutster-brain-realtime", features = ["mock"] } diff --git a/crates/rutster-trunk/tests/reflex_on_trunk.rs b/crates/rutster-trunk/tests/reflex_on_trunk.rs new file mode 100644 index 0000000..b73bc0d --- /dev/null +++ b/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` + `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 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" + ); +} diff --git a/crates/rutster-trunk/tests/sim_integ.rs b/crates/rutster-trunk/tests/sim_integ.rs new file mode 100644 index 0000000..4c6833f --- /dev/null +++ b/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>, + 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> { + 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::(64); + let (pump_tap_tx, mut tap_out_rx) = mpsc::channel::(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::(32); + let (tx_audio_out, rx_audio_out) = mpsc::channel::(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::(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::(16); + let (outbound_tx, mut outbound_rx) = mpsc::channel::(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::(8); + let (_tx_function_call_output, mut rx_function_call_output) = + mpsc::channel::(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. +}