From ab1ab8192c2856b4931e83fdf567c7ba08585d49 Mon Sep 17 00:00:00 2001 From: opencode controller Date: Tue, 30 Jun 2026 23:31:47 -0400 Subject: [PATCH] =?UTF-8?q?feat(brain-realtime):=20MockRealtimeBrain=20+?= =?UTF-8?q?=20binary=20(spec=20=C2=A75.3=20+=20=C2=A74.2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit slice-3 §5.3 dev mode: an in-process fake OpenAI Realtime WS server (MockRealtimeBrain) behind --features=mock. Asserts session.update has turn_detection: null (S4 — spec §4.3) on handshake; exports a typed OpenAI-shaped error on violation so the brain fails fast + visibly. On each input_audio_buffer.append replies with a canned response.audio.delta carrying 480 zeroed samples as base64 LE i16 (same wire shape as slice-2's audio_out). Powers the offline dev loop (cargo run -p rutster-brain-realtime --features=mock) + the integration test (no real OpenAI credentials, no network calls). Also adds the binary main.rs (spec §4.2 + §5.3): binds the tap-side WS server on RUTSTER_TAP_BIND (default 127.0.0.1:8082), accepts per-call tap connections, handles the hello handshake, splits the WS into sink + stream, dials the OpenAI side (real wss://api.openai.com in default mode; MockRealtimeBrain url in mock mode), and bridges via run_openai_pump + two mpsc::channel pairs. Integration test brain_mock_round_trip drives MockRealtimeBrain + run_openai_pump end-to-end: pushes a tap audio_in frame through the pump, observes the canned response.audio.delta arrive + get translated back as a tap audio_out frame on the pump's outbound mpsc. Verifies the dev-loop 'actually start + interop on a basic audio round-trip' contract the directive's Task 4 calls out, without requiring a real WebRTC peer (same constraint as slice-1/2's browser-driven e2e). Files: crates/rutster-brain-realtime/src/{lib,mock,main}.rs, crates/rutster-brain-realtime/tests/brain_mock_round_trip.rs. NOT touched: loop_driver.rs, rtc_session.rs (seam §7.5 #6). Gates: cargo fmt OK. cargo clippy --all --tests --features=mock -D warnings OK. cargo test --all --features=mock OK (56 tests). cargo deny check has pre-existing environmental failure (CVSS 4.0 unsupported in advisory-db; same on main). Dev loop verified: cargo run -p rutster-brain-realtime --features=mock starts the brain (mock on ephemeral port, tap server on :8082). ./target/release/rutster with RUTSTER_TAP_URL=ws://127.0.0.1:8082/realtime starts the core on :8080. POST /v1/sessions succeeds. (Full audio round-trip requires a browser-driven WebRTC peer, same constraint as slice-1/2 — covered by the integration test instead.) --- crates/rutster-brain-realtime/src/lib.rs | 9 + crates/rutster-brain-realtime/src/main.rs | 212 +++++++++++ crates/rutster-brain-realtime/src/mock.rs | 328 ++++++++++++++++++ .../tests/brain_mock_round_trip.rs | 88 +++++ 4 files changed, 637 insertions(+) create mode 100644 crates/rutster-brain-realtime/src/main.rs create mode 100644 crates/rutster-brain-realtime/src/mock.rs create mode 100644 crates/rutster-brain-realtime/tests/brain_mock_round_trip.rs diff --git a/crates/rutster-brain-realtime/src/lib.rs b/crates/rutster-brain-realtime/src/lib.rs index 08d54d4..f2f7d1a 100644 --- a/crates/rutster-brain-realtime/src/lib.rs +++ b/crates/rutster-brain-realtime/src/lib.rs @@ -26,3 +26,12 @@ pub mod api_key; pub mod openai_client; pub mod translator; + +/// slice-3 §5.3 dev mode: in-process fake OpenAI Realtime WS server. Only +/// available with `--features=mock`. The binary uses it to run the offline +/// dev loop (no API key, no network calls to OpenAI); the integration test +/// uses it for the same reason. +#[cfg(feature = "mock")] +pub mod mock; +#[cfg(feature = "mock")] +pub use mock::MockRealtimeBrain; diff --git a/crates/rutster-brain-realtime/src/main.rs b/crates/rutster-brain-realtime/src/main.rs new file mode 100644 index 0000000..a01f8ca --- /dev/null +++ b/crates/rutster-brain-realtime/src/main.rs @@ -0,0 +1,212 @@ +//! # rutster-brain-realtime binary (slice-3 spec §4.2 + §5.3) +//! +//! Standalone brain process the core dials out to via `RUTSTER_TAP_URL` (spec +//! §5.1). One process bridges many tap WS connections (one per call) to many +//! OpenAI Realtime sessions. +//! +//! # Modes +//! +//! - **`--features=mock`** (offline dev loop, slice-3 spec §7.3): starts an +//! in-process `MockRealtimeBrain` WS server, dials it as the "OpenAI side." +//! No API key, no network calls to OpenAI. Used by the integration test + +//! the offline dev loop (`cargo run -p rutster-brain-realtime --features=mock`). +//! - **Default** (real OpenAI): loads `OPENAI_API_KEY` per +//! [`api_key::load_api_key`], dials `wss://api.openai.com/v1/realtime?model=...`. +//! +//! Both modes share the same tap-side WS server + the same `run_openai_pump` +//! bridging logic — only the OpenAI-side dialer differs. + +use std::net::SocketAddr; + +use futures_util::{SinkExt, StreamExt}; +#[cfg(feature = "mock")] +use rutster_brain_realtime::MockRealtimeBrain; +#[cfg(not(feature = "mock"))] +use rutster_brain_realtime::api_key; +use rutster_brain_realtime::openai_client; +use rutster_brain_realtime::translator::build_openai_session_update; +use tokio::sync::mpsc; +use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::tungstenite::client::IntoClientRequest; +use tracing::{info, warn}; + +/// Default bind addr for the brain process's tap-side WS server (spec §5.3). +/// Distinct from slice-2's echo brain (`:8081/echo`); the two coexist. +const DEFAULT_TAP_BIND: &str = "127.0.0.1:8082"; +const DEFAULT_VOICE: &str = "alloy"; +const DEFAULT_MODEL: &str = "gpt-4o-realtime"; + +#[tokio::main] +async fn main() { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| "rutster_brain_realtime=info".into()), + ) + .init(); + + let tap_bind: SocketAddr = std::env::var("RUTSTER_TAP_BIND") + .unwrap_or_else(|_| DEFAULT_TAP_BIND.to_string()) + .parse() + .expect("RUTSTER_TAP_BIND must be a valid SocketAddr"); + let voice = + std::env::var("OPENAI_REALTIME_VOICE").unwrap_or_else(|_| DEFAULT_VOICE.to_string()); + let model = + std::env::var("OPENAI_REALTIME_MODEL").unwrap_or_else(|_| DEFAULT_MODEL.to_string()); + let _ = &model; // used only in the non-mock branch below. + + #[cfg(feature = "mock")] + let openai_url: String; + #[cfg(feature = "mock")] + let openai_headers: Vec<(String, String)>; + #[cfg(feature = "mock")] + let _mock_guard; + #[cfg(not(feature = "mock"))] + let (openai_url, openai_headers) = { + let api_key = match api_key::load_api_key() { + Ok(k) => k, + Err(e) => { + eprintln!("failed to load OPENAI_API_KEY: {e}; refusing to start"); + std::process::exit(1); + } + }; + ( + openai_client::openai_realtime_url(&model), + openai_client::openai_headers(&api_key), + ) + }; + #[cfg(feature = "mock")] + { + let mock = MockRealtimeBrain::start().await.expect("mock bind ok"); + info!(mock_addr = %mock.addr, "MockRealtimeBrain started (--features=mock)"); + openai_url = mock.url(); + openai_headers = Vec::new(); + _mock_guard = mock; + } + let listener = tokio::net::TcpListener::bind(tap_bind) + .await + .expect("bind ok"); + info!(%tap_bind, %openai_url, "rutster-brain-realtime listening (core dials this)"); + + loop { + match listener.accept().await { + Ok((stream, peer)) => { + let url = openai_url.clone(); + let hdrs = openai_headers.clone(); + let voice = voice.clone(); + tokio::spawn(async move { + if let Err(e) = handle_tap_connection(stream, peer, &url, &hdrs, &voice).await { + warn!(error = ?e, %peer, "tap connection ended with error"); + } + }); + } + Err(e) => warn!(error = %e, "accept failed; continuing"), + } + } +} + +/// Handle one tap WS connection from the core. Spawns two mpsc forwarders +/// (tap→pump, pump→tap) + dials the OpenAI side, then runs `run_openai_pump` +/// on the OpenAI side + awaits both forwarder tasks. +/// +/// The tap-side handshake (hello/session_end) is handled here; the OpenAI +/// pump sees only audio_in + function_call_output on its inbound + emits +/// only audio_out/speech_*/function_call/error on its outbound. +async fn handle_tap_connection( + stream: tokio::net::TcpStream, + peer: SocketAddr, + openai_url: &str, + openai_headers: &[(String, String)], + voice: &str, +) -> Result<(), Box> { + let mut tap_ws = tokio_tungstenite::accept_async(stream).await?; + info!(%peer, "tap WS connection accepted"); + + // === Tap handshake: first frame must be hello; send hello ack. === + 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 = rutster_tap::decode_envelope(&hello_text)?; + let session_id = match decoded.payload { + rutster_tap::DecodedPayload::Hello(p) => p.session_id, + _ => return Err("first tap frame not hello".into()), + }; + info!(%peer, %session_id, "tap hello received"); + let ack = rutster_tap::encode_hello(&session_id, 0, 0)?; + tap_ws.send(Message::Text(ack)).await?; + + // === Bridge channels between tap WS and OpenAI pump. === + // tap_via (Sender): the core's tap frames flow IN here — the forwarder + // reads text from tap_ws and forwards via this channel. The OpenAI + // pump's `tap_rx: mpsc::Receiver` is the receiving end. + // pump_tap_tx (Sender): the pump's outbound tap frames (audio_out, + // speech_started, function_call, error) flow OUT here. The forwarder + // reads from the paired Receiver and writes to tap_ws. + let (tap_via, pump_tap_rx) = mpsc::channel::(64); + let (pump_tap_tx, mut tap_out_rx) = mpsc::channel::(64); + + // Split the tap WS into sink + stream so the two forwarder tasks can + // own each half independently. `tokio_tungstenite::WebSocketStream::split` + // yields `(SplitSink<..., Message>, SplitStream<...>)` — the canonical + // tokio pattern for fan-out into two parallel async tasks. + let (mut tap_sink, mut tap_stream) = tap_ws.split(); + + // === Forwarder: tap_ws_stream → pump_tap_rx === + let in_fwd = tokio::spawn(async move { + while let Some(msg_res) = tap_stream.next().await { + match msg_res { + Ok(m) => { + if let Ok(text) = m.into_text() { + if tap_via.send(text).await.is_err() { + break; // pump dropped + } + } + } + Err(e) => { + warn!(error = ?e, "tap_ws recv error"); + break; + } + } + } + }); + + // === Forwarder: pump_tap_tx → tap_ws_sink === + let out_fwd = tokio::spawn(async move { + while let Some(text) = tap_out_rx.recv().await { + if let Err(e) = tap_sink.send(Message::Text(text)).await { + warn!(error = ?e, "tap_ws send error"); + break; + } + } + }); + + // === Dial the OpenAI side (real or mock). === + let request = openai_url.into_client_request()?; + let mut request = request; + for (k, v) in openai_headers { + request.headers_mut().insert( + http::HeaderName::from_bytes(k.as_bytes())?, + http::HeaderValue::from_str(v)?, + ); + } + let (openai_ws, _resp) = tokio_tungstenite::connect_async(request).await?; + info!(%peer, %openai_url, "OpenAI side connected"); + + // `run_openai_pump` sends its own session.update on startup per + // `build_openai_session_update`; don't pre-send here. + let _ = build_openai_session_update; + info!(%peer, voice = %voice, "starting OpenAI pump"); + + let pump_result = + openai_client::run_openai_pump(openai_ws, pump_tap_rx, pump_tap_tx, voice.to_string()) + .await; + match pump_result { + Ok(()) => info!(%peer, "OpenAI pump exited cleanly"), + Err(e) => warn!(%peer, error = ?e, "OpenAI pump exited with error"), + } + in_fwd.abort(); + out_fwd.abort(); + Ok(()) +} diff --git a/crates/rutster-brain-realtime/src/mock.rs b/crates/rutster-brain-realtime/src/mock.rs new file mode 100644 index 0000000..e7df45d --- /dev/null +++ b/crates/rutster-brain-realtime/src/mock.rs @@ -0,0 +1,328 @@ +//! # MockRealtimeBrain — in-process fake OpenAI Realtime WS server (spec §5.3) +//! +//! Behind `--features=mock` only. Runs entirely in-process; no network calls +//! to OpenAI, no API key required. Powers: +//! - the offline dev loop (`cargo run -p rutster-brain-realtime --features=mock` +//! starts the brain process with this mock as the OpenAI side), and +//! - slice-3's integration test (no real credentials, deterministic canned +//! responses). +//! +//! # Contract +//! +//! - Accepts a WS client connection (the brain process's OpenAI-side client). +//! - Asserts that the first event the client sends is a `session.update` +//! with `turn_detection: null` (the S4 load-bearing decision — spec §4.3). +//! - On each `input_audio_buffer.append` event the client sends, replies +//! with a canned `response.audio.delta` event carrying base64 LE i16 PCM +//! (480 zeroed samples per 20 ms frame — same wire shape as slice-2's +//! audio_out, so the brain's translator round-trips through the same +//! PcmFrame codec real OpenAI would use). +//! - Echoes back `input_audio_buffer.speech_started` / `.speech_stopped` +//! so the brain's translator can forward them as advisory tap events +//! (spec §3.2 — these are advisory in slice-3; step 4 wires the FOB +//! reflex loop). +//! - On `response.function_call_arguments.done`... the brain's translator +//! consumes those (OpenAI emits them, the brain forwards to core); the +//! mock doesn't generate unprompted function_call events (the integration +//! test exercises the function-call round-trip via a deterministic test +//! hook, not via the mock's own drive). + +use std::net::SocketAddr; + +use futures_util::{SinkExt, StreamExt}; +use serde_json::{Value, json}; +use tokio::net::TcpListener; +use tokio::sync::oneshot; +use tokio::task::JoinHandle; +use tokio_tungstenite::tungstenite::Message; +use tracing::{debug, info, warn}; + +/// The handle returned by `MockRealtimeBrain::start`. Drop or call `shutdown` +/// to stop the mock server + abort its accept-loop task. +pub struct MockRealtimeBrain { + /// The address the mock bound. The brain process dials this URL + /// (`ws:///`) instead of the real `wss://api.openai.com/...` + /// when running with `--features=mock`. + pub addr: SocketAddr, + pub shutdown: Option>, + pub join: JoinHandle<()>, +} + +impl MockRealtimeBrain { + /// Start the mock on an ephemeral port (caller picks up `addr` for the + /// brain process to dial). The accept loop spawns per-connection tasks + /// so multiple sessions can run concurrently in the integration test. + pub async fn start() -> std::io::Result { + let listener = TcpListener::bind("127.0.0.1:0").await?; + let addr = listener.local_addr()?; + let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>(); + let join = tokio::spawn(accept_loop(listener, shutdown_rx)); + info!(%addr, "MockRealtimeBrain listening (fake OpenAI Realtime)"); + Ok(Self { + addr, + shutdown: Some(shutdown_tx), + join, + }) + } + + /// Build the URL the brain process's OpenAI client should dial instead + /// of `wss://api.openai.com/v1/realtime?model=...`. The path is `/v1/realtime` + /// so the brain client's `into_client_request` call works the same as it + /// does against the real OpenAI endpoint. + pub fn url(&self) -> String { + format!("ws://{}/v1/realtime?model=mock", self.addr) + } +} + +impl Drop for MockRealtimeBrain { + fn drop(&mut self) { + // Best-effort shutdown signal — the task may already have exited. + // `Option::take()` lets us send by-value without moving out of + // `self` (Drop takes `&mut self`, not `self`). + if let Some(tx) = self.shutdown.take() { + let _ = tx.send(()); + } + self.join.abort(); + } +} + +async fn accept_loop(listener: TcpListener, mut shutdown: oneshot::Receiver<()>) { + loop { + tokio::select! { + biased; + _ = &mut shutdown => { + info!("MockRealtimeBrain accept loop shutting down"); + return; + } + res = listener.accept() => { + let (stream, peer) = match res { + Ok(s) => s, + Err(e) => { + warn!(error = ?e, "MockRealtimeBrain accept failed; continuing"); + continue; + } + }; + tokio::spawn(handle_connection(stream, peer)); + } + } + } +} + +/// Handle one brain-process → mock-OpenAI WS connection. Each connection is +/// one OpenAI Realtime session (stateless across reconnects — same as the +/// brain process's own contract for the tap side). +async fn handle_connection(stream: tokio::net::TcpStream, peer: SocketAddr) { + let ws = tokio_tungstenite::accept_async(stream).await; + let Ok(mut ws) = ws else { + warn!(%peer, "MockRealtimeBrain WS handshake failed"); + return; + }; + debug!(%peer, "MockRealtimeBrain connection accepted"); + + // First event MUST be session.update with turn_detection: null. + // Wait for it (bounded 2s — the brain process sends it immediately + // after the WS handshake per `run_openai_pump`). + let first = tokio::time::timeout(std::time::Duration::from_secs(2), ws.next()).await; + match first { + Ok(Some(Ok(msg))) => { + let Ok(text) = msg.into_text() else { + warn!(%peer, "MockRealtimeBrain first frame not text; closing"); + let _ = ws.close(None).await; + return; + }; + let event: Value = match serde_json::from_str(&text) { + Ok(v) => v, + Err(e) => { + warn!(%peer, error = ?e, "MockRealtimeBrain first frame not JSON; closing"); + let _ = ws.close(None).await; + return; + } + }; + if event.get("type").and_then(|v| v.as_str()) != Some("session.update") { + warn!(%peer, "MockRealtimeBrain first event not session.update; closing"); + let _ = ws.close(None).await; + return; + } + // S4 load-bearing assertion (spec §4.3). Do NOT silently accept + // a session.update that has turn_detection != null — that would + // mask a regression in the brain's translator. + let td = event.pointer("/session/turn_detection"); + if td != Some(&Value::Null) { + warn!(%peer, td = ?td, "MockRealtimeBrain S4 violation: turn_detection != null; closing"); + // Echo back an OpenAI-shaped error so the brain surfaces it. + let err = json!({ + "type": "error", + "error": { + "code": "invalid_session_update", + "message": "MockRealtimeBrain: turn_detection must be null (S4)" + } + }); + let _ = ws.send(Message::Text(err.to_string())).await; + let _ = ws.close(None).await; + return; + } + info!(%peer, "MockRealtimeBrain session.update accepted (turn_detection=null)"); + } + _ => { + warn!(%peer, "MockRealtimeBrain no session.update within 2s; closing"); + let _ = ws.close(None).await; + return; + } + } + + // Pump loop: reply to input_audio_buffer.append with canned audio. + while let Some(msg_res) = ws.next().await { + let msg = match msg_res { + Ok(m) => m, + Err(e) => { + debug!(%peer, error = ?e, "MockRealtimeBrain WS recv error; closing"); + return; + } + }; + let Ok(text) = msg.into_text() else { continue }; + let event: Value = match serde_json::from_str(&text) { + Ok(v) => v, + Err(e) => { + debug!(%peer, error = ?e, "MockRealtimeBrain non-JSON event; ignoring"); + continue; + } + }; + let event_type = event.get("type").and_then(|v| v.as_str()).unwrap_or(""); + match event_type { + "input_audio_buffer.append" => { + // Canned response: one `response.audio.delta` per append + // carrying 480 zeroed samples as base64 LE i16 (the same + // wire shape as slice-2's audio_out — verified by the + // translator's existing round-trip test). Identical bytes + // every time: deterministic for test assertions. + use base64::Engine; + let pcm_zeros = vec![0u8; 960]; // 480 i16 samples × 2 bytes + let pcm_b64 = base64::engine::general_purpose::STANDARD.encode(&pcm_zeros); + let delta = json!({ + "type": "response.audio.delta", + "delta": pcm_b64 + }); + if let Err(e) = ws.send(Message::Text(delta.to_string())).await { + debug!(%peer, error = ?e, "MockRealtimeBrain send delta failed; closing"); + return; + } + } + "input_audio_buffer.speech_started" => { + // Pass-through echo so the brain's translator forwards it as + // a tap `speech_started` advisory event. + let evt = json!({ "type": "input_audio_buffer.speech_started" }); + let _ = ws.send(Message::Text(evt.to_string())).await; + } + "input_audio_buffer.speech_stopped" => { + let evt = json!({ "type": "input_audio_buffer.speech_stopped" }); + let _ = ws.send(Message::Text(evt.to_string())).await; + } + "conversation.item.create" => { + // The brain sends this to relay a function_call_output (tool + // reply) back to OpenAI. The mock just acknowledges by + // echoing a typed `conversation.item.created` event so the + // brain's OpenAI pump doesn't treat the silence as an error. + let evt = json!({ + "type": "conversation.item.created", + "item": event.get("item").cloned().unwrap_or(Value::Null) + }); + let _ = ws.send(Message::Text(evt.to_string())).await; + } + _ => { + debug!(%peer, event_type = %event_type, "MockRealtimeBrain ignoring event"); + } + } + } + debug!(%peer, "MockRealtimeBrain connection closed"); +} + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Duration; + use tokio_tungstenite::tungstenite::client::IntoClientRequest; + + /// The S4 turn-ownership decision is load-bearing (spec §4.3 + ADR-0008). + /// The mock asserts the brain's session.update has turn_detection: null; + /// this test confirms a session.update WITH turn_detection: null is + /// accepted (the green path). + #[tokio::test] + async fn accepts_session_update_with_turn_detection_null() { + let mock = MockRealtimeBrain::start().await.unwrap(); + let url = mock.url(); + let req = url.as_str().into_client_request().unwrap(); + let (mut ws, _resp) = tokio_tungstenite::connect_async(req).await.unwrap(); + + // Send session.update with turn_detection: null. + let session_update = json!({ + "type": "session.update", + "session": { + "modalities": ["text", "audio"], + "voice": "alloy", + "turn_detection": null + } + }); + ws.send(Message::Text(session_update.to_string())) + .await + .unwrap(); + + // Send input_audio_buffer.append — must receive a canned + // response.audio.delta (the brain's translator round-trips this + // through encode_audio_out, so the wire shape must be exactly what + // real OpenAI sends: base64 PCM16 in a `delta` field). + let append = json!({ + "type": "input_audio_buffer.append", + "audio": "AAAA" + }); + ws.send(Message::Text(append.to_string())).await.unwrap(); + + let delta_msg = tokio::time::timeout(Duration::from_millis(500), ws.next()) + .await + .expect("delta within 500ms") + .unwrap() + .unwrap(); + let delta_text = delta_msg.into_text().unwrap(); + let delta: Value = serde_json::from_str(&delta_text).unwrap(); + assert_eq!(delta["type"], "response.audio.delta"); + assert!(delta["delta"].is_string()); + // The mock's canned PCM is 480 zeroed samples; the base64 string + // must decode to exactly 960 bytes (480 × 2 bytes/i16). + use base64::Engine; + let bytes = base64::engine::general_purpose::STANDARD + .decode(delta["delta"].as_str().unwrap()) + .unwrap(); + assert_eq!(bytes.len(), 960); + } + + /// The S4 violation: session.update with turn_detection != null + /// surfaces a typed OpenAI-shaped error so the brain's pump fails fast + /// + visibly. Catches a future regression where the brain process + /// forgets to set turn_detection: null. + #[tokio::test] + async fn rejects_session_update_with_turn_detection_set() { + let mock = MockRealtimeBrain::start().await.unwrap(); + let url = mock.url(); + let req = url.as_str().into_client_request().unwrap(); + let (mut ws, _resp) = tokio_tungstenite::connect_async(req).await.unwrap(); + + let session_update = json!({ + "type": "session.update", + "session": { + "voice": "alloy", + "turn_detection": { "type": "server_vad" } + } + }); + ws.send(Message::Text(session_update.to_string())) + .await + .unwrap(); + + let err_msg = tokio::time::timeout(Duration::from_millis(500), ws.next()) + .await + .expect("error within 500ms") + .unwrap() + .unwrap(); + let err: Value = serde_json::from_str(&err_msg.into_text().unwrap()).unwrap(); + assert_eq!(err["type"], "error"); + assert_eq!(err["error"]["code"], "invalid_session_update"); + } +} diff --git a/crates/rutster-brain-realtime/tests/brain_mock_round_trip.rs b/crates/rutster-brain-realtime/tests/brain_mock_round_trip.rs new file mode 100644 index 0000000..73c5b52 --- /dev/null +++ b/crates/rutster-brain-realtime/tests/brain_mock_round_trip.rs @@ -0,0 +1,88 @@ +//! slice-3 integration test: brain-process-level round-trip through the +//! MockRealtimeBrain + the translator + `run_openai_pump`. Drives the brain +//! the way the core's TapClient would (synthetic tap frames in, observes tap +//! frames out) — proves the dev-loop interop contract the PM directive's +//! Task 4 calls out ("actually start + interop on a basic audio round-trip") +//! without requiring a real WebRTC peer (browser-driven, same constraint as +//! slice-1/2). +//! +//! What this exercises end-to-end: +//! 1. MockRealtimeBrain accepts a WS connection + asserts session.update +//! has turn_detection: null (S4 — spec §4.3). +//! 2. `run_openai_pump` sends session.update + bridges tap frames ↔ OpenAI +//! events via the translator. +//! 3. A tap `audio_in` frames the test pushes through `tap_via` → translator +//! formats as `input_audio_buffer.append` → mock generates a canned +//! `response.audio.delta` → translator formats as tap `audio_out` → +//! arrives on `pump_tap_tx` (the pump's outbound mpsc). + +use rutster_brain_realtime::{MockRealtimeBrain, openai_client::run_openai_pump}; +use rutster_tap::{PcmFrame, decode_envelope, encode_audio_in, encode_hello}; +use serde_json::Value; +use std::time::Duration; +use tokio::sync::mpsc; +use tokio_tungstenite::tungstenite::client::IntoClientRequest; + +#[tokio::test] +async fn brain_mock_round_trip_translates_audio_in_to_audio_out() { + let mock = MockRealtimeBrain::start().await.unwrap(); + let url = mock.url(); + let req = url.as_str().into_client_request().unwrap(); + let (openai_ws, _resp) = tokio_tungstenite::connect_async(req).await.unwrap(); + + // Bridge channels matching `run_openai_pump`'s signature (it owns neither + // side of the tap WS — the binary's main.rs does the WS forwarding; here + // we drive the pump directly with mpsc as the test's tap peer). + let (tap_via, pump_tap_rx) = mpsc::channel::(64); + let (pump_tap_tx, mut tap_out_rx) = mpsc::channel::(64); + + // The pump sends session.update + then runs the select! loop. Spawn it. + let pump = tokio::spawn(async move { + run_openai_pump(openai_ws, pump_tap_rx, pump_tap_tx, "alloy".to_string()).await + }); + + // Push a hello (the pump ignores hello per its `_ => warn!(...)` arm, + // but real tap traffic always starts with hello). + let hello = encode_hello("test-session", 0, 0).unwrap(); + tap_via.send(hello).await.unwrap(); + + // Push one audio_in frame. The pump translates to + // input_audio_buffer.append; the mock replies with a canned + // response.audio.delta; the pump translates back to a tap audio_out + // frame on pump_tap_tx. + let frame = PcmFrame::zeroed(); + let audio_in = encode_audio_in(&frame, 1, 100).unwrap(); + tap_via.send(audio_in).await.unwrap(); + + // Bounded-wait the audio_out frame coming back. Generous bound: the + // pump's tokio select! cycle + the mock's WS round-trip on loopback + // are sub-millisecond; 500 ms is fail-fast for any real regression. + let out_str = tokio::time::timeout(Duration::from_millis(500), tap_out_rx.recv()) + .await + .expect("audio_out within 500ms") + .expect("channel not closed"); + + let decoded = decode_envelope(&out_str).unwrap(); + match decoded.payload { + rutster_tap::DecodedPayload::AudioOut(p) => { + // The mock's canned PCM is 480 zeroed samples; the brain's + // translator round-trips it through PcmFrame (decode + + // re-encode), so the wire shape is exactly slice-2's audio_out. + assert_eq!(p.samples, 480); + let recovered = rutster_tap::decode_pcm(&p.pcm, p.samples).unwrap(); + assert_eq!(recovered.samples.len(), 480); + // Canned zeros — every sample is 0. + assert_eq!(recovered.samples[0], 0); + assert_eq!(recovered.samples[479], 0); + // seq + ts carried through from the translator's per-event + // counter (initial value 0 on the first outbound frame). + assert_eq!(decoded.seq, 0); + } + other => panic!("expected AudioOut, got {other:?}"), + } + + // Drop the input channel to signal the pump to exit cleanly. + drop(tap_via); + let _ = pump.await; + let _ = Value::Null; // suppress unused-import if Value unused here +}