Cargo.lock

@@ -1208,6 +1208,7 @@ dependencies = [
  "async-trait",
  "base64",
  "futures-util",
+ "http",
  "rutster-media",
  "rutster-tap",
  "serde",

Cargo.toml

@@ -54,3 +54,8 @@ url = "2"
 base64 = "0.22"
 # async-trait 0.1: async fns in trait objects (Tool trait, slice-3 spec §6.1).
 async-trait = "0.1"
+# http 1: Request/Response builders for the OpenAI WS handshake (slice-3
+# spec §5.3). Task 9's main.rs builds the http::Request<...> the tungstenite
+# client sends before WS upgrade; `openai_client::openai_headers` produces
+# the header pairs the caller stuffs into it.
+http = "1"

crates/rutster-brain-realtime/Cargo.toml

@@ -21,6 +21,9 @@ url = { workspace = true }
 async-trait = { workspace = true }
 base64 = { workspace = true }
 thiserror = { workspace = true }
+# http: Request/Response builder for the WS handshake (Task 9's main.rs).
+# This task's unit tests assert URL + headers shape only.
+http = { workspace = true }
 
 [dev-dependencies]
 tempfile = "3"

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

@@ -1 +1,222 @@
-//! wss://api.openai.com/v1/realtime client (spec §4). Filled in by Task 5.
+//! # 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");
+    }
+}