1. 1//! Provider-neutral helpers for the OpenAI Realtime wire protocol.
2. 2//!
3. 3//! The model and voice are deliberately supplied by discovery/configuration;
4. 4//! this module contains no catalogue or fallback identifiers. Keeping the
5. 5//! wire messages here makes the streaming transport usable by HTTP, desktop,
6. 6//! and channel surfaces without duplicating protocol details.
7. 7
8. 8use crate::error::LlmError;
9. 9use futures::{SinkExt, StreamExt};
10. 10use serde_json::{Value, json};
11. 11use tokio_tungstenite::tungstenite::Message;
12. 12use tokio_util::sync::CancellationToken;
13. 13
14. 14#[derive(Debug, Clone, PartialEq, Eq)]
15. 15pub struct RealtimeConfig {
16. 16 pub model: String,
17. 17 pub voice: Option<String>,
18. 18 pub input_format: String,
19. 19 pub output_format: String,
20. 20}
21. 21
22. 22impl RealtimeConfig {
23. 23 pub fn validate(&self) -> Result<(), LlmError> {
24. 24 for (name, value) in [
25. 25 ("model", self.model.as_str()),
26. 26 ("input_format", self.input_format.as_str()),
27. 27 ("output_format", self.output_format.as_str()),
28. 28 ] {
29. 29 if value.trim().is_empty() {
30. 30 return Err(LlmError::InvalidRequest(format!(
31. 31 "realtime {name} is required"
32. 32 )));
33. 33 }
34. 34 }
35. 35 if self.voice.as_deref().is_some_and(|v| v.trim().is_empty()) {
36. 36 return Err(LlmError::InvalidRequest(
37. 37 "realtime voice cannot be empty".into(),
38. 38 ));
39. 39 }
40. 40 Ok(())
41. 41 }
42. 42}
43. 43
44. 44/// Build the initial session update. Empty optional voice is omitted so the
45. 45/// provider can apply its configured default.
46. 46pub fn build_session_update(
47. 47 config: &RealtimeConfig,
48. 48 instructions: Option<&str>,
49. 49) -> Result<Value, LlmError> {
50. 50 config.validate()?;
51. 51 let mut session = json!({
52. 52 "modalities": ["text", "audio"],
53. 53 "input_audio_format": config.input_format,
54. 54 "output_audio_format": config.output_format,
55. 55 });
56. 56 if let Some(voice) = config.voice.as_deref().filter(|v| !v.trim().is_empty()) {
57. 57 session["voice"] = json!(voice);
58. 58 }
59. 59 if let Some(text) = instructions.filter(|v| !v.trim().is_empty()) {
60. 60 session["instructions"] = json!(text);
61. 61 }
62. 62 Ok(json!({"type":"session.update", "session": session}))
63. 63}
64. 64
65. 65pub fn build_audio_append(audio: &[u8]) -> Result<Value, LlmError> {
66. 66 if audio.is_empty() {
67. 67 return Err(LlmError::InvalidRequest(
68. 68 "realtime audio cannot be empty".into(),
69. 69 ));
70. 70 }
71. 71 let encoded = base64::engine::general_purpose::STANDARD.encode(audio);
72. 72 Ok(json!({"type":"input_audio_buffer.append", "audio": encoded}))
73. 73}
74. 74
75. 75pub fn build_response_create() -> Value {
76. 76 json!({"type":"response.create", "response": {"modalities":["audio","text"]}})
77. 77}
78. 78
79. 79/// Execute one OpenAI Realtime turn over a websocket. The endpoint is
80. 80/// supplied by configuration so compatible providers can use the same
81. 81/// transport. Audio is returned as the concatenated `response.audio.delta`
82. 82/// payload; all other provider events are ignored by this low-level adapter.
83. 83pub async fn round_trip(
84. 84 api_key: &str,
85. 85 endpoint: &str,
86. 86 config: &RealtimeConfig,
87. 87 audio: &[u8],
88. 88 instructions: Option<&str>,
89. 89 cancel: &CancellationToken,
90. 90) -> Result<Vec<u8>, LlmError> {
91. 91 config.validate()?;
92. 92 if api_key.trim().is_empty() || endpoint.trim().is_empty() {
93. 93 return Err(LlmError::InvalidRequest(
94. 94 "realtime credentials and endpoint are required".into(),
95. 95 ));
96. 96 }
97. 97 if audio.is_empty() {
98. 98 return Err(LlmError::InvalidRequest(
99. 99 "realtime audio cannot be empty".into(),
100. 100 ));
101. 101 }
102. 102 let separator = if endpoint.contains('?') { '&' } else { '?' };
103. 103 let url = format!(
104. 104 "{endpoint}{separator}model={}",
105. 105 percent_encoding::utf8_percent_encode(&config.model, percent_encoding::NON_ALPHANUMERIC)
106. 106 );
107. 107 let request = tokio_tungstenite::tungstenite::http::Request::builder()
108. 108 .uri(url)
109. 109 .header("Authorization", format!("Bearer {api_key}"))
110. 110 .header("OpenAI-Beta", "realtime=v1")
111. 111 .body(())
112. 112 .map_err(|e| LlmError::InvalidRequest(e.to_string()))?;
113. 113 let (mut socket, _) = tokio::select! {
114. 114 _ = cancel.cancelled() => return Err(LlmError::Aborted { partial: None }),
115. 115 result = tokio_tungstenite::connect_async(request) => result.map_err(|e| LlmError::Network(e.to_string()))?,
116. 116 };
117. 117 let send = |value: Value| Message::Text(value.to_string());
118. 118 for value in [
119. 119 build_session_update(config, instructions)?,
120. 120 build_audio_append(audio)?,
121. 121 build_response_create(),
122. 122 ] {
123. 123 tokio::select! {
124. 124 _ = cancel.cancelled() => return Err(LlmError::Aborted { partial: None }),
125. 125 result = socket.send(send(value)) => result.map_err(|e| LlmError::Network(e.to_string()))?,
126. 126 }
127. 127 }
128. 128 const MAX_AUDIO_BYTES: usize = 16 * 1024 * 1024;
129. 129 let mut output = Vec::new();
130. 130 while let Some(message) = tokio::select! {
131. 131 _ = cancel.cancelled() => return Err(LlmError::Aborted {
132. 132 // Realtime audio is returned as bytes, while the shared LLM
133. 133 // abort contract stores textual assistant messages. The caller
134. 134 // still owns the already-emitted audio buffer and can preserve it.
135. 135 partial: None,
136. 136 }),
137. 137 message = socket.next() => message,
138. 138 } {
139. 139 let message = message.map_err(|e| LlmError::Network(e.to_string()))?;
140. 140 let Message::Text(text) = message else {
141. 141 continue;
142. 142 };
143. 143 let event: Value =
144. 144 serde_json::from_str(&text).map_err(|e| LlmError::Parse(e.to_string()))?;
145. 145 match event.get("type").and_then(Value::as_str) {
146. 146 Some("response.audio.delta") => {
147. 147 let Some(delta) = event.get("delta").and_then(Value::as_str) else {
148. 148 continue;
149. 149 };
150. 150 let bytes = base64::engine::general_purpose::STANDARD
151. 151 .decode(delta)
152. 152 .map_err(|e| LlmError::Parse(e.to_string()))?;
153. 153 if output.len().saturating_add(bytes.len()) > MAX_AUDIO_BYTES {
154. 154 return Err(LlmError::InvalidRequest(
155. 155 "provider audio exceeds 16 MiB".into(),
156. 156 ));
157. 157 }
158. 158 output.extend(bytes);
159. 159 }
160. 160 Some("error") => return Err(LlmError::InvalidRequest(event.to_string())),
161. 161 Some("response.done") => break,
162. 162 _ => {}
163. 163 }
164. 164 }
165. 165 if output.is_empty() {
166. 166 return Err(LlmError::Parse(
167. 167 "provider returned empty realtime audio".into(),
168. 168 ));
169. 169 }
170. 170 Ok(output)
171. 171}
172. 172
173. 173use base64::Engine;
174. 174
175. 175#[cfg(test)]
176. 176#[allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)]
177. 177mod tests {
178. 178 use super::*;
179. 179 #[test]
180. 180 fn session_payload_requires_discovered_model_and_omits_voice_default() {
181. 181 let cfg = RealtimeConfig {
182. 182 model: "discovered-model".into(),
183. 183 voice: None,
184. 184 input_format: "pcm16".into(),
185. 185 output_format: "pcm16".into(),
186. 186 };
187. 187 let body = build_session_update(&cfg, Some("be concise")).unwrap();
188. 188 assert_eq!(body["type"], "session.update");
189. 189 assert!(body["session"].get("voice").is_none());
190. 190 assert_eq!(body["session"]["instructions"], "be concise");
191. 191 }
192. 192 #[test]
193. 193 fn audio_append_is_base64_and_rejects_empty() {
194. 194 let body = build_audio_append(&[1, 2, 3]).unwrap();
195. 195 assert_eq!(body["audio"], "AQID");
196. 196 assert!(build_audio_append(&[]).is_err());
197. 197 }
198. 198}
199. 199