From e791d422ed0f8759278018779781fb54d74812ff Mon Sep 17 00:00:00 2001 From: opencode controller Date: Tue, 30 Jun 2026 20:39:49 -0400 Subject: [PATCH] =?UTF-8?q?feat(brain-realtime):=20translator=20=E2=80=94?= =?UTF-8?q?=20tap=20=E2=87=84=20OpenAI=20Realtime=20event=20mapping=20(spe?= =?UTF-8?q?c=20=C2=A74)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pure functions, no I/O, no call state. The OpenAI-side WS client (Task 5) and the tap-side WS server (Task 9) call these per event. S4 turn-ownership decision (spec §4.3, load-bearing per ADR-0008) encoded in build_openai_session_update: turn_detection: null. OpenAI's server-side VAD is disabled; the FOB reflex loop (step 4) owns turn-taking; tap playout stays core-authoritative (slice-2 §4.1). S4 test verifies the load-bearing assertion (v['session']['turn_detection'] == Null). Audio is pass-through — OpenAI's PCM base64 (LE i16 24 kHz mono) is indentical to slice-2's tap PCM wire shape (spec §3.5); decode + re-encode as a PcmFrame so the playout ring stays byte-aligned for slice-2's playout-buffer invariants (samples: 480). --- Cargo.lock | 1 + crates/rutster-brain-realtime/Cargo.toml | 1 + .../rutster-brain-realtime/src/translator.rs | 262 +++++++++++++++++- 3 files changed, 263 insertions(+), 1 deletion(-) diff --git a/Cargo.lock b/Cargo.lock index 1d4b312..6c8f5fd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1208,6 +1208,7 @@ dependencies = [ "async-trait", "base64", "futures-util", + "rutster-media", "rutster-tap", "serde", "serde_json", diff --git a/crates/rutster-brain-realtime/Cargo.toml b/crates/rutster-brain-realtime/Cargo.toml index 895b92b..0c3fa9f 100644 --- a/crates/rutster-brain-realtime/Cargo.toml +++ b/crates/rutster-brain-realtime/Cargo.toml @@ -9,6 +9,7 @@ description = "OpenAI Realtime speech-to-speech brain — translates slice-2's t [dependencies] rutster-tap = { path = "../rutster-tap" } +rutster-media = { path = "../rutster-media" } # for PcmFrame in tests (spec §4.2 audio round-trip) tokio = { workspace = true, features = ["full"] } tokio-tungstenite = { workspace = true, features = ["connect"] } futures-util = { workspace = true } diff --git a/crates/rutster-brain-realtime/src/translator.rs b/crates/rutster-brain-realtime/src/translator.rs index 4413ae4..7c07231 100644 --- a/crates/rutster-brain-realtime/src/translator.rs +++ b/crates/rutster-brain-realtime/src/translator.rs @@ -1 +1,261 @@ -//! Tap ⇄ OpenAI Realtime event translation (spec §4). Filled in by Task 4. +//! # 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 { + 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 { + 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"); + } +}