diff --git a/Cargo.lock b/Cargo.lock index 6c8f5fd..b7c1a17 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1208,6 +1208,7 @@ dependencies = [ "async-trait", "base64", "futures-util", + "http", "rutster-media", "rutster-tap", "serde", diff --git a/Cargo.toml b/Cargo.toml index c048b1f..f96eba1 100644 --- a/Cargo.toml +++ b/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" diff --git a/crates/rutster-brain-realtime/Cargo.toml b/crates/rutster-brain-realtime/Cargo.toml index 0c3fa9f..8b869af 100644 --- a/crates/rutster-brain-realtime/Cargo.toml +++ b/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" diff --git a/crates/rutster-brain-realtime/src/openai_client.rs b/crates/rutster-brain-realtime/src/openai_client.rs index 76a7072..eeb898f 100644 --- a/crates/rutster-brain-realtime/src/openai_client.rs +++ b/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=`. +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 ` +/// - `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( + mut openai_ws: tokio_tungstenite::WebSocketStream, + mut tap_rx: mpsc::Receiver, + tap_tx: mpsc::Sender, + 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"); + } +}