feat(brain-realtime): MockRealtimeBrain + binary (spec §5.3 + §4.2)
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<String> 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.)
This commit is contained in:
@@ -26,3 +26,12 @@
|
|||||||
pub mod api_key;
|
pub mod api_key;
|
||||||
pub mod openai_client;
|
pub mod openai_client;
|
||||||
pub mod translator;
|
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;
|
||||||
|
|||||||
212
crates/rutster-brain-realtime/src/main.rs
Normal file
212
crates/rutster-brain-realtime/src/main.rs
Normal file
@@ -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<dyn std::error::Error>> {
|
||||||
|
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<String>` 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::<String>(64);
|
||||||
|
let (pump_tap_tx, mut tap_out_rx) = mpsc::channel::<String>(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(())
|
||||||
|
}
|
||||||
328
crates/rutster-brain-realtime/src/mock.rs
Normal file
328
crates/rutster-brain-realtime/src/mock.rs
Normal file
@@ -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://<addr>/`) instead of the real `wss://api.openai.com/...`
|
||||||
|
/// when running with `--features=mock`.
|
||||||
|
pub addr: SocketAddr,
|
||||||
|
pub shutdown: Option<oneshot::Sender<()>>,
|
||||||
|
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<Self> {
|
||||||
|
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");
|
||||||
|
}
|
||||||
|
}
|
||||||
88
crates/rutster-brain-realtime/tests/brain_mock_round_trip.rs
Normal file
88
crates/rutster-brain-realtime/tests/brain_mock_round_trip.rs
Normal file
@@ -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::<String>(64);
|
||||||
|
let (pump_tap_tx, mut tap_out_rx) = mpsc::channel::<String>(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
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user