Slice 3 — OpenAI Realtime brain: swap echo for the brain (#4)

crates/rutster-brain-realtime/src/api_key.rs

@@ -0,0 +1,199 @@
+//! # API-key loader (spec §5.3)
+//!
+//! Two-source config: `OPENAI_API_KEY` env var default + optional
+//! `OPENAI_API_KEY_FILE` path override. KMS/Vault integration is deferred
+//! to step 6 (spec §1.2); the file-path override makes secret-manager
+//! injection (k8s secrets, Vault agent) trivial when that layer exists.
+//!
+//! # Why file-path override (not env-only)
+//!
+//! A k8s pod mounts secrets at file paths (e.g. `/var/secrets/openai_key`).
+//! Env-var secrets are also fine, but file-path is the standard pattern for
+//! k8s + Vault agent + sidecar secret rotators. The dev loop uses the env
+//! var path (single-source, simplest).
+
+use std::env;
+use std::fs;
+use std::path::PathBuf;
+
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum ApiKeyError {
+    #[error("no API key: set OPENAI_API_KEY or OPENAI_API_KEY_FILE")]
+    NotFound,
+    #[error("OPENAI_API_KEY_FILE path is not valid UTF-8: {0:?}")]
+    PathNotUtf8(PathBuf),
+    #[error("failed to read OPENAI_API_KEY_FILE at {path}: {source}")]
+    FileRead {
+        path: PathBuf,
+        #[source]
+        source: std::io::Error,
+    },
+}
+
+/// Load the OpenAI API key per the env-var + file-path posture (spec §5.3).
+///
+/// Precedence: `OPENAI_API_KEY_FILE` wins over `OPENAI_API_KEY` (a file
+/// path is a more specific override; this matches k8s-secret patterns where
+/// an operator mounts a file to override the default env var).
+///
+/// Trims trailing whitespace from either source (some k8s secret mounts
+/// add trailing newlines).
+pub fn load_api_key() -> Result<String, ApiKeyError> {
+    if let Ok(path_str) = env::var("OPENAI_API_KEY_FILE") {
+        let path = PathBuf::from(&path_str);
+        let raw = fs::read_to_string(&path).map_err(|source| ApiKeyError::FileRead {
+            path: path.clone(),
+            source,
+        })?;
+        return Ok(raw.trim().to_string());
+    }
+    if let Ok(raw) = env::var("OPENAI_API_KEY") {
+        return Ok(raw.trim().to_string());
+    }
+    Err(ApiKeyError::NotFound)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // NOTE: edition 2024 makes env::set_var / env::remove_var `unsafe` (they
+    // can race with concurrent reads of the process environ). The brief was
+    // written against edition 2021 syntax; we wrap each mutation in
+    // `unsafe { ... }` rather than reverting to a pre-2024 edition. The
+    // unsafety is local to the test process and accepted by the stdlib's
+    // own test helpers; no threads here read these env vars concurrently.
+
+    // The process environ is process-global state. Even with the save/restore
+    // pattern below, running these tests in parallel would let one test's
+    // `set_var` race against another's `remove_var` for the same variable,
+    // producing flaky failures. This mutex serializes every test in this
+    // module so the env-var mutations are atomic across tests.
+    static ENV_MUTEX: std::sync::Mutex<()> = std::sync::Mutex::new(());
+
+    /// Acquire the env mutex for the test body. Panics if the lock is held
+    /// (only happens if a test panics while held — fine, fail fast).
+    fn lock_env() -> std::sync::MutexGuard<'static, ()> {
+        ENV_MUTEX.lock().unwrap_or_else(|e| e.into_inner())
+    }
+
+    #[test]
+    fn returns_error_when_neither_env_nor_file_set() {
+        let _guard = lock_env();
+        // Save then clear both env vars to make the test order-independent.
+        let saved_key = env::var_os("OPENAI_API_KEY").map(|v| ("OPENAI_API_KEY".to_string(), v));
+        let saved_file =
+            env::var_os("OPENAI_API_KEY_FILE").map(|v| ("OPENAI_API_KEY_FILE".to_string(), v));
+        unsafe {
+            env::remove_var("OPENAI_API_KEY");
+            env::remove_var("OPENAI_API_KEY_FILE");
+        }
+
+        let r = load_api_key();
+        assert!(matches!(r, Err(ApiKeyError::NotFound)));
+
+        // Restore for any other test in the same process.
+        if let Some((k, v)) = saved_key {
+            unsafe {
+                env::set_var(k, v);
+            }
+        }
+        if let Some((k, v)) = saved_file {
+            unsafe {
+                env::set_var(k, v);
+            }
+        }
+    }
+
+    #[test]
+    fn reads_from_env_var() {
+        let _guard = lock_env();
+        let saved = env::var_os("OPENAI_API_KEY").map(|v| ("OPENAI_API_KEY".to_string(), v));
+        let saved_file =
+            env::var_os("OPENAI_API_KEY_FILE").map(|v| ("OPENAI_API_KEY_FILE".to_string(), v));
+        unsafe {
+            env::set_var("OPENAI_API_KEY", "sk-test-12345");
+            env::remove_var("OPENAI_API_KEY_FILE");
+        }
+
+        let r = load_api_key().unwrap();
+        assert_eq!(r, "sk-test-12345");
+
+        if let Some((k, v)) = saved {
+            unsafe {
+                env::set_var(k, v);
+            }
+        } else {
+            unsafe {
+                env::remove_var("OPENAI_API_KEY");
+            }
+        }
+        if let Some((k, v)) = saved_file {
+            unsafe {
+                env::set_var(k, v);
+            }
+        }
+    }
+
+    #[test]
+    fn trim_trailing_newline_from_env() {
+        let _guard = lock_env();
+        let saved = env::var_os("OPENAI_API_KEY").map(|v| ("OPENAI_API_KEY".to_string(), v));
+        unsafe {
+            env::set_var("OPENAI_API_KEY", "sk-test-with-newline\n");
+            env::remove_var("OPENAI_API_KEY_FILE");
+        }
+
+        let r = load_api_key().unwrap();
+        assert_eq!(r, "sk-test-with-newline");
+
+        if let Some((k, v)) = saved {
+            unsafe {
+                env::set_var(k, v);
+            }
+        } else {
+            unsafe {
+                env::remove_var("OPENAI_API_KEY");
+            }
+        }
+    }
+
+    #[test]
+    fn file_path_overrides_env() {
+        let _guard = lock_env();
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("key.txt");
+        std::fs::write(&path, "sk-from-file-98765\n").unwrap();
+
+        unsafe {
+            env::set_var("OPENAI_API_KEY", "sk-from-env-should-be-overridden");
+            env::set_var("OPENAI_API_KEY_FILE", path.to_str().unwrap());
+        }
+
+        let r = load_api_key().unwrap();
+        assert_eq!(r, "sk-from-file-98765");
+
+        unsafe {
+            env::remove_var("OPENAI_API_KEY");
+            env::remove_var("OPENAI_API_KEY_FILE");
+        }
+    }
+
+    #[test]
+    fn file_path_missing_returns_file_read_error() {
+        let _guard = lock_env();
+        unsafe {
+            env::set_var("OPENAI_API_KEY_FILE", "/nonexistent/path/to/key.txt");
+            env::remove_var("OPENAI_API_KEY");
+        }
+
+        let r = load_api_key();
+        assert!(matches!(r, Err(ApiKeyError::FileRead { .. })));
+
+        unsafe {
+            env::remove_var("OPENAI_API_KEY_FILE");
+        }
+    }
+}

crates/rutster-brain-realtime/src/lib.rs

@@ -0,0 +1,37 @@
+//! # rutster-brain-realtime
+//!
+//! **Slice-3 brain:** translates slice-2's tap protocol (the green-zone side of
+//! the seam — core-as-client dials this brain's WS server; ADR-0008 classifies
+//! the brain as green-zone) to OpenAI Realtime's event schema. The brain
+//! process is a WS *server* (core-as-client dials it, unchanged from slice-2)
+//! AND a WS *client* to `wss://api.openai.com/v1/realtime` (OpenAI is server
+//! on *its* leg).
+//!
+//! See `docs/superpowers/specs/2026-06-30-slice-3-realtime-brain-design.md` for
+//! the full design.
+//!
+//! ## Modules
+//!
+//! - [`api_key`] — load API key from env var or file (spec §5.3).
+//! - [`translator`] — pure-function event translation (tap ⇄ OpenAI).
+//! - [`openai_client`] — wss:// client to api.openai.com/v1/realtime.
+//!
+//! ## Dev mode (`--features=mock`)
+//!
+//! When built with `mock`, the binary uses in-process `MockRealtimeBrain`
+//! (defined in `lib.rs`'s test-support module) instead of dialing OpenAI.
+//! No API key required, no real OpenAI calls. Used by the integration test +
+//! the offline dev loop (spec §7.3).
+
+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;

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<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(())
+}

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://<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");
+    }
+}

crates/rutster-brain-realtime/src/openai_client.rs

@@ -0,0 +1,222 @@
+//! # wss://api.openai.com/v1/realtime client (spec §4)
+//!
+//! Owns the OpenAI-side WS connection. Brings the OpenAI Realtime event
+//! stream up, sends `session.update` with `turn_detection: null` (S4,
+//! spec §4.3) on handshake, then runs a pump loop that:
+//! - reads tap-side events (audio_in, function_call_output) from the
+//!   tap WS server's `WebSocketStream` and translates + forwards them
+//!   to OpenAI;
+//! - reads OpenAI events (response.audio.delta, speech_started/stopped,
+//!   function_call_arguments.done, error) and translates + forwards them
+//!   to the tap WS server.
+//!
+//! Reconnects with bounded backoff on OpenAI-side WS failure (spec §4.4 —
+//! the tap side stays connected; the OpenAI side has its own failure
+//! surface). The brain process emits a tap `error` event on OpenAI-side
+//! failure so the core can observe it.
+
+// Imports are split by origin so the reader can trace which types come from
+// the wire-shape (rutster-tap — shared with the core), which come from the
+// pure translators (crate::translator — this crate's pure-function layer),
+// and which come from third-party plumbing.
+use crate::translator::{
+    build_openai_session_update, openai_audio_delta_to_tap_audio_out,
+    openai_function_call_arguments_done_to_tap, openai_speech_event_to_tap,
+    tap_audio_in_to_openai_append, tap_function_call_output_to_openai_create_item,
+};
+use rutster_tap::{DecodedPayload, TapProtoError, decode_envelope, encode_function_call};
+use serde_json::Value;
+use thiserror::Error;
+use tokio::sync::mpsc;
+use tokio_tungstenite::tungstenite::Message;
+use tracing::{info, warn};
+
+#[derive(Debug, Error)]
+pub enum OpenAiClientError {
+    #[error("OpenAI WS error: {0}")]
+    Ws(#[from] tokio_tungstenite::tungstenite::Error),
+    #[error("OpenAI auth failed (401)")]
+    AuthFailed,
+    #[error("translator error: {0}")]
+    Translate(#[from] crate::translator::TranslateError),
+    #[error("tap protocol error: {0}")]
+    TapProto(#[from] TapProtoError),
+}
+
+/// Build the OpenAI Realtime URL for the given model. Spec §4.2 + §5.3:
+/// `wss://api.openai.com/v1/realtime?model=<model>`.
+pub fn openai_realtime_url(model: &str) -> String {
+    format!("wss://api.openai.com/v1/realtime?model={model}")
+}
+
+/// Build the HTTP headers for the OpenAI WS handshake (spec §5.3). Two
+/// required headers:
+/// - `Authorization: Bearer <api_key>`
+/// - `OpenAI-Beta: realtime=v1`
+pub fn openai_headers(api_key: &str) -> Vec<(String, String)> {
+    vec![
+        ("Authorization".to_string(), format!("Bearer {api_key}")),
+        ("OpenAI-Beta".to_string(), "realtime=v1".to_string()),
+    ]
+}
+
+/// Drive the OpenAI Realtime WS connection + the tap-side pump.
+///
+/// `tap_rx`: inbound side — *tap frames* the brain received from the core
+///   (audio_in, function_call_output). We translate each one to its OpenAI
+///   equivalent + send to OpenAI.
+/// `openai_ws`: the OpenAI WS connection (already connected; the caller
+///   does the dial + auth).
+/// `tap_tx`: outbound side — *tap frames* the brain sends back to the core
+///   (audio_out, speech_started/stopped, function_call, error). We
+///   translate OpenAI events into these + send.
+pub async fn run_openai_pump<T>(
+    mut openai_ws: tokio_tungstenite::WebSocketStream<T>,
+    mut tap_rx: mpsc::Receiver<String>,
+    tap_tx: mpsc::Sender<String>,
+    voice: String,
+) -> Result<(), OpenAiClientError>
+where
+    T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
+{
+    // === Handshake: send session.update with turn_detection: null (S4). ===
+    let session_update = build_openai_session_update(&voice);
+    openai_ws
+        .send(Message::Text(session_update.to_string()))
+        .await?;
+    info!(voice = %voice, "sent session.update to OpenAI (turn_detection: null)");
+
+    let mut seq_egress = 0u64;
+
+    use futures_util::{SinkExt, StreamExt};
+    loop {
+        tokio::select! {
+            // Inbound tap frame from the core (audio_in, function_call_output).
+            tap_str = tap_rx.recv() => {
+                let Some(tap_str) = tap_str else {
+                    info!("tap_rx closed; ending OpenAI pump");
+                    return Ok(());
+                };
+                let decoded = decode_envelope(&tap_str)?;
+                match decoded.payload {
+                    DecodedPayload::AudioIn(audio) => {
+                        let append = tap_audio_in_to_openai_append(&audio.pcm);
+                        openai_ws.send(Message::Text(append.to_string())).await?;
+                    }
+                    DecodedPayload::FunctionCallOutput(out) => {
+                        let create_item = tap_function_call_output_to_openai_create_item(
+                            &out.id, &out.status, &out.result,
+                        );
+                        openai_ws.send(Message::Text(create_item.to_string())).await?;
+                    }
+                    // Ignore others; slice-2's hello/audio_in on the tap side
+                    // happens before this pump starts.
+                    _ => warn!(?decoded.payload, "unexpected tap frame to OpenAI pump"),
+                }
+            }
+            // Inbound OpenAI event.
+            msg = openai_ws.next() => {
+                let Some(msg) = msg else {
+                    info!("OpenAI WS stream ended");
+                    return Ok(());
+                };
+                let msg = msg?;
+                let Ok(text) = msg.into_text() else { continue };
+                let openai_event: Value = match serde_json::from_str(&text) {
+                    Ok(v) => v,
+                    Err(e) => {
+                        warn!(error = ?e, "OpenAI sent non-JSON event; ignoring");
+                        continue;
+                    }
+                };
+                let event_type = openai_event.get("type").and_then(|v| v.as_str()).unwrap_or("");
+                match event_type {
+                    "response.audio.delta" => {
+                        let tap_str = openai_audio_delta_to_tap_audio_out(
+                            &openai_event, seq_egress, 0,
+                        )?;
+                        seq_egress += 1;
+                        tap_tx.send(tap_str).await.map_err(|_| {
+                            OpenAiClientError::Ws(
+                                tokio_tungstenite::tungstenite::Error::ConnectionClosed,
+                            )
+                        })?;
+                    }
+                    "input_audio_buffer.speech_started" => {
+                        let tap_str = openai_speech_event_to_tap(
+                            &openai_event, true, seq_egress, 0,
+                        )?;
+                        seq_egress += 1;
+                        tap_tx.send(tap_str).await.map_err(|_| {
+                            OpenAiClientError::Ws(
+                                tokio_tungstenite::tungstenite::Error::ConnectionClosed,
+                            )
+                        })?;
+                    }
+                    "input_audio_buffer.speech_stopped" => {
+                        let tap_str = openai_speech_event_to_tap(
+                            &openai_event, false, seq_egress, 0,
+                        )?;
+                        seq_egress += 1;
+                        tap_tx.send(tap_str).await.map_err(|_| {
+                            OpenAiClientError::Ws(
+                                tokio_tungstenite::tungstenite::Error::ConnectionClosed,
+                            )
+                        })?;
+                    }
+                    "response.function_call_arguments.done" => {
+                        let (call_id, name, args_str) =
+                            openai_function_call_arguments_done_to_tap(&openai_event)?;
+                        let tap_str = encode_function_call(
+                            &call_id, &name, &args_str, seq_egress, 0,
+                        )?;
+                        seq_egress += 1;
+                        tap_tx.send(tap_str).await.map_err(|_| {
+                            OpenAiClientError::Ws(
+                                tokio_tungstenite::tungstenite::Error::ConnectionClosed,
+                            )
+                        })?;
+                    }
+                    "error" => {
+                        // OpenAI emits a typed error event (e.g. 401 auth
+                        // failure surfaces here). Inspect the .error.code.
+                        let code = openai_event
+                            .pointer("/error/code")
+                            .and_then(|v| v.as_str())
+                            .unwrap_or("unknown");
+                        warn!(code = %code, "OpenAI error event");
+                        if code == "invalid_api_key" {
+                            return Err(OpenAiClientError::AuthFailed);
+                        }
+                    }
+                    _ => {
+                        // Unknown OpenAI event — log + drop (don't crash).
+                        warn!(event_type = %event_type, "ignoring OpenAI event type");
+                    }
+                }
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn openai_url_for_known_model() {
+        assert_eq!(
+            openai_realtime_url("gpt-4o-realtime"),
+            "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime"
+        );
+    }
+
+    #[test]
+    fn openai_headers_carry_bearer_auth_and_beta() {
+        let h = openai_headers("sk-test-12345");
+        assert_eq!(h[0].0, "Authorization");
+        assert_eq!(h[0].1, "Bearer sk-test-12345");
+        assert_eq!(h[1].0, "OpenAI-Beta");
+        assert_eq!(h[1].1, "realtime=v1");
+    }
+}

crates/rutster-brain-realtime/src/translator.rs

@@ -0,0 +1,261 @@
+//! # Tap ⇄ OpenAI Realtime event translation (spec §4)
+//!
+//! Pure functions — no async, no I/O, no call state. Each function maps
+//! one event between slice-3's tap protocol (slice-2 v1 + the additive
+//! slice-3 events from Task 2) and OpenAI Realtime's event JSON.
+//!
+//! The translation layer is **stateless** by design: the OpenAI-side WS
+//! client (Task 5) and the tap-side WS server (Task 9) call these
+//! functions per-event. No ownership of OpenAI's `session_id` beyond
+//! the connection's lifetime.
+
+use rutster_tap::{TapProtoError, encode_audio_out, encode_speech_started, encode_speech_stopped};
+use serde_json::{Value, json};
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum TranslateError {
+    #[error("OpenAI event missing required field: {field}")]
+    MissingField { field: &'static str },
+    #[error("tap protocol error: {0}")]
+    TapProto(#[from] TapProtoError),
+}
+
+/// Build OpenAI's `session.update` event (spec §4.2). The S4 turn-ownership
+/// decision (spec §4.3): `turn_detection: null`. OpenAI's server-side VAD
+/// is disabled; the FOB reflex loop (step 4) owns turn-taking.
+pub fn build_openai_session_update(voice: &str) -> Value {
+    json!({
+        "type": "session.update",
+        "session": {
+            "modalities": ["text", "audio"],
+            "voice": voice,
+            "input_audio_format": "pcm16",
+            "output_audio_format": "pcm16",
+            "sample_rate": 24000,
+            "turn_detection": null
+        }
+    })
+}
+
+/// Wrap a base64 PCM payload (already encoded per slice-2 §3 — explicit LE
+/// i16) into OpenAI's `input_audio_buffer.append` event (spec §4.2). Pass-
+/// through — the wire shape for OpenAI's audio is identical to slice-2's
+/// tap PCM (spec §3.5).
+pub fn tap_audio_in_to_openai_append(pcm_b64: &str) -> Value {
+    json!({
+        "type": "input_audio_buffer.append",
+        "audio": pcm_b64
+    })
+}
+
+/// Translate OpenAI's `response.audio.delta` event (carries a `delta` field
+/// with base64 PCM) to a slice-3 tap `audio_out` frame string (spec §4.2).
+/// Pass-through on the audio payload; the envelope carries seq/ts.
+pub fn openai_audio_delta_to_tap_audio_out(
+    openai_event: &Value,
+    seq: u64,
+    ts: u64,
+) -> Result<String, TranslateError> {
+    let pcm_b64 = openai_event
+        .get("delta")
+        .and_then(|v| v.as_str())
+        .ok_or(TranslateError::MissingField { field: "delta" })?;
+    // OpenAI's base64 PCM is LE i16 24 kHz mono — identical wire shape to
+    // slice-2's tap PCM. Decode + re-encode as a PcmFrame (the canonical
+    // 480-sample shape) so slice-2's playout ring stays byte-aligned.
+    let frame = rutster_tap::decode_pcm(pcm_b64, rutster_tap::WIRE_SAMPLES_PER_FRAME)?;
+    Ok(encode_audio_out(&frame, seq, ts)?)
+}
+
+/// Translate OpenAI's `input_audio_buffer.speech_started` (started=true) or
+/// `.speech_stopped` (started=false) event to the matching tap frame
+/// (spec §4.2 + §4.3 — advisory events, the FOB reflex loop in step 4 will
+/// act on them).
+pub fn openai_speech_event_to_tap(
+    _openai_event: &Value,
+    started: bool,
+    seq: u64,
+    ts: u64,
+) -> Result<String, TranslateError> {
+    if started {
+        Ok(encode_speech_started(seq, ts)?)
+    } else {
+        Ok(encode_speech_stopped(seq, ts)?)
+    }
+}
+
+/// Translate OpenAI's `response.function_call_arguments.done` event to the
+/// tuple `(call_id, name, args_json_str)` the brain will emit as a tap
+/// `function_call` frame (spec §4.2 + §4.3). OpenAI sends `arguments` as
+/// a JSON _string_, not a JSON object — we don't reparse it; the brain
+/// process passes the raw string to `encode_function_call` (Task 2's
+/// encode_function_call parses it then into a serde_json::Value).
+pub fn openai_function_call_arguments_done_to_tap(
+    openai_event: &Value,
+) -> Result<(String, String, String), TranslateError> {
+    let call_id = openai_event
+        .get("call_id")
+        .and_then(|v| v.as_str())
+        .ok_or(TranslateError::MissingField { field: "call_id" })?;
+    let name = openai_event
+        .get("name")
+        .and_then(|v| v.as_str())
+        .ok_or(TranslateError::MissingField { field: "name" })?;
+    // OpenAI's `arguments` is a JSON string. If missing, default to "{}".
+    let args_json_str = openai_event
+        .get("arguments")
+        .and_then(|v| v.as_str())
+        .unwrap_or("{}")
+        .to_string();
+    Ok((call_id.to_string(), name.to_string(), args_json_str))
+}
+
+/// Build OpenAI's `conversation.item.create` event for a tool-call result
+/// (spec §4.2). The tap `function_call_output` carries id + status + result;
+/// OpenAI's `conversation.item.create` wraps them as a function_call_output
+/// item with `call_id` (= id) + `output` (= JSON-stringified result; status
+/// is plumbed into the result value as a `_status` field — OpenAI has no
+/// formal status concept, so we encode ours in the result body).
+pub fn tap_function_call_output_to_openai_create_item(
+    id: &str,
+    status: &str,
+    result: &Value,
+) -> Value {
+    let mut output_value = result.clone();
+    if let Some(obj) = output_value.as_object_mut() {
+        obj.insert("_status".to_string(), Value::String(status.to_string()));
+    }
+    json!({
+        "type": "conversation.item.create",
+        "item": {
+            "type": "function_call_output",
+            "call_id": id,
+            "output": output_value.to_string()
+        }
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use rutster_tap::{DecodedPayload, decode_envelope};
+
+    /// S4 turn-ownership test (load-bearing per ADR-0008, spec §4.3):
+    /// the OpenAI session.update must have turn_detection: null.
+    #[test]
+    fn session_update_disables_openai_turn_detection() {
+        let v = build_openai_session_update("alloy");
+        assert_eq!(v["type"], "session.update");
+        assert_eq!(v["session"]["voice"], "alloy");
+        assert_eq!(v["session"]["modalities"][0], "text");
+        assert_eq!(v["session"]["modalities"][1], "audio");
+        assert_eq!(v["session"]["input_audio_format"], "pcm16");
+        assert_eq!(v["session"]["output_audio_format"], "pcm16");
+        assert_eq!(v["session"]["sample_rate"], 24000);
+        // THE load-bearing assertion:
+        assert_eq!(v["session"]["turn_detection"], Value::Null);
+    }
+
+    #[test]
+    fn append_audio_payload_is_passthrough() {
+        // The PCM base64 string for one zero frame (480 samples, every LE i16=0).
+        use base64::Engine;
+        let zeros = [0u8; 960];
+        let pcm_b64 = base64::engine::general_purpose::STANDARD.encode(zeros);
+        let v = tap_audio_in_to_openai_append(&pcm_b64);
+        assert_eq!(v["type"], "input_audio_buffer.append");
+        assert_eq!(v["audio"], pcm_b64);
+    }
+
+    #[test]
+    fn openai_audio_delta_to_tap_round_trips_through_slice2_codec() {
+        // Build a slice-2 PcmFrame, base64-encode it the slice-2 way,
+        // wrap it in an OpenAI response.audio.delta, translate to a tap
+        // audio_out frame, then decode the tap frame with decode_envelope
+        // and assert the PCM round-trips.
+        use rutster_media::PcmFrame;
+        let mut frame = PcmFrame::zeroed();
+        frame.samples[0] = 100;
+        frame.samples[1] = -100;
+        frame.samples[479] = 12345;
+        let pcm_b64 = rutster_tap::encode_pcm(&frame);
+        let openai_event = json!({
+            "type": "response.audio.delta",
+            "delta": pcm_b64
+        });
+        let tap_str = openai_audio_delta_to_tap_audio_out(&openai_event, 5, 1000).unwrap();
+        let decoded = decode_envelope(&tap_str).unwrap();
+        match decoded.payload {
+            DecodedPayload::AudioOut(p) => {
+                let recovered = rutster_tap::decode_pcm(&p.pcm, p.samples).unwrap();
+                assert_eq!(recovered.samples[0], 100);
+                assert_eq!(recovered.samples[1], -100);
+                assert_eq!(recovered.samples[479], 12345);
+                assert_eq!(decoded.seq, 5);
+                assert_eq!(decoded.ts, 1000);
+            }
+            other => panic!("expected AudioOut, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn speech_started_translates_to_tap_speech_started() {
+        let openai_event = json!({ "type": "input_audio_buffer.speech_started" });
+        let s = openai_speech_event_to_tap(&openai_event, true, 3, 300).unwrap();
+        let d = decode_envelope(&s).unwrap();
+        assert!(matches!(d.payload, DecodedPayload::SpeechStarted));
+        assert_eq!(d.seq, 3);
+    }
+
+    #[test]
+    fn speech_stopped_translates_to_tap_speech_stopped() {
+        let openai_event = json!({ "type": "input_audio_buffer.speech_stopped" });
+        let s = openai_speech_event_to_tap(&openai_event, false, 7, 700).unwrap();
+        let d = decode_envelope(&s).unwrap();
+        assert!(matches!(d.payload, DecodedPayload::SpeechStopped));
+        assert_eq!(d.seq, 7);
+    }
+
+    #[test]
+    fn function_call_done_extracts_three_tuple() {
+        let openai_event = json!({
+            "type": "response.function_call_arguments.done",
+            "call_id": "call_abc123",
+            "name": "hangup",
+            "arguments": "{\"reason\": \"caller_requested\"}"
+        });
+        let (id, name, args_str) =
+            openai_function_call_arguments_done_to_tap(&openai_event).unwrap();
+        assert_eq!(id, "call_abc123");
+        assert_eq!(name, "hangup");
+        assert_eq!(args_str, "{\"reason\": \"caller_requested\"}");
+    }
+
+    #[test]
+    fn function_call_done_missing_call_id_errors() {
+        let openai_event = json!({
+            "type": "response.function_call_arguments.done",
+            "name": "hangup"
+        });
+        let r = openai_function_call_arguments_done_to_tap(&openai_event);
+        assert!(matches!(
+            r,
+            Err(TranslateError::MissingField { field: "call_id" })
+        ));
+    }
+
+    #[test]
+    fn function_call_output_to_create_item_plumbs_status_into_result() {
+        let result = json!({"channel_state": "Closing"});
+        let v = tap_function_call_output_to_openai_create_item("call_abc123", "ok", &result);
+        assert_eq!(v["type"], "conversation.item.create");
+        assert_eq!(v["item"]["type"], "function_call_output");
+        assert_eq!(v["item"]["call_id"], "call_abc123");
+        // The output is a JSON string — _status gets plumbed inside.
+        let output_str = v["item"]["output"].as_str().unwrap();
+        let output_val: Value = serde_json::from_str(output_str).unwrap();
+        assert_eq!(output_val["channel_state"], "Closing");
+        assert_eq!(output_val["_status"], "ok");
+    }
+}