Fix realtime websocket URL params and conversation_id naming

MODULE.bazel.lock

@@ -686,11 +686,11 @@
       "chrono_0.4.43": "{\"dependencies\":[{\"features\":[\"derive\"],\"name\":\"arbitrary\",\"optional\":true,\"req\":\"^1.0.0\"},{\"kind\":\"dev\",\"name\":\"bincode\",\"req\":\"^1.3.0\"},{\"name\":\"defmt\",\"optional\":true,\"req\":\"^1.0.1\"},{\"features\":[\"fallback\"],\"name\":\"iana-time-zone\",\"optional\":true,\"req\":\"^0.1.45\",\"target\":\"cfg(unix)\"},{\"name\":\"js-sys\",\"optional\":true,\"req\":\"^0.3\",\"target\":\"cfg(all(target_arch = \\\"wasm32\\\", not(any(target_os = \\\"emscripten\\\", target_os = \\\"wasi\\\"))))\"},{\"default_features\":false,\"name\":\"num-traits\",\"req\":\"^0.2\"},{\"name\":\"pure-rust-locales\",\"optional\":true,\"req\":\"^0.8.2\"},{\"default_features\":false,\"name\":\"rkyv\",\"optional\":true,\"req\":\"^0.7.43\"},{\"default_features\":false,\"name\":\"serde\",\"optional\":true,\"req\":\"^1.0.99\"},{\"default_features\":false,\"kind\":\"dev\",\"name\":\"serde_derive\",\"req\":\"^1\"},{\"kind\":\"dev\",\"name\":\"serde_json\",\"req\":\"^1\"},{\"kind\":\"dev\",\"name\":\"similar-asserts\",\"req\":\"^1.6.1\"},{\"name\":\"wasm-bindgen\",\"optional\":true,\"req\":\"^0.2\",\"target\":\"cfg(all(target_arch = \\\"wasm32\\\", not(any(target_os = \\\"emscripten\\\", target_os = \\\"wasi\\\"))))\"},{\"kind\":\"dev\",\"name\":\"wasm-bindgen-test\",\"req\":\"^0.3\",\"target\":\"cfg(all(target_arch = \\\"wasm32\\\", not(any(target_os = \\\"emscripten\\\", target_os = \\\"wasi\\\"))))\"},{\"kind\":\"dev\",\"name\":\"windows-bindgen\",\"req\":\"^0.66\"},{\"name\":\"windows-link\",\"optional\":true,\"req\":\"^0.2\",\"target\":\"cfg(windows)\"}],\"features\":{\"__internal_bench\":[],\"alloc\":[],\"clock\":[\"winapi\",\"iana-time-zone\",\"now\"],\"core-error\":[],\"default\":[\"clock\",\"std\",\"oldtime\",\"wasmbind\"],\"defmt\":[\"dep:defmt\",\"pure-rust-locales?/defmt\"],\"libc\":[],\"now\":[\"std\"],\"oldtime\":[],\"rkyv\":[\"dep:rkyv\",\"rkyv/size_32\"],\"rkyv-16\":[\"dep:rkyv\",\"rkyv?/size_16\"],\"rkyv-32\":[\"dep:rkyv\",\"rkyv?/size_32\"],\"rkyv-64\":[\"dep:rkyv\",\"rkyv?/size_64\"],\"rkyv-validation\":[\"rkyv?/validation\"],\"std\":[\"alloc\"],\"unstable-locales\":[\"pure-rust-locales\"],\"wasmbind\":[\"wasm-bindgen\",\"js-sys\"],\"winapi\":[\"windows-link\"]}}",
       "chunked_transfer_1.5.0": "{\"dependencies\":[{\"kind\":\"dev\",\"name\":\"criterion\",\"req\":\"^0.3\"}],\"features\":{}}",
       "cipher_0.4.4": "{\"dependencies\":[{\"name\":\"blobby\",\"optional\":true,\"req\":\"^0.3\"},{\"name\":\"crypto-common\",\"req\":\"^0.1.6\"},{\"name\":\"inout\",\"req\":\"^0.1\"},{\"default_features\":false,\"name\":\"zeroize\",\"optional\":true,\"req\":\"^1.5\"}],\"features\":{\"alloc\":[],\"block-padding\":[\"inout/block-padding\"],\"dev\":[\"blobby\"],\"rand_core\":[\"crypto-common/rand_core\"],\"std\":[\"alloc\",\"crypto-common/std\",\"inout/std\"]}}",
-      "clap_4.5.56": "{\"dependencies\":[{\"kind\":\"dev\",\"name\":\"automod\",\"req\":\"^1.0.14\"},{\"default_features\":false,\"kind\":\"dev\",\"name\":\"clap-cargo\",\"req\":\"^0.15.0\"},{\"default_features\":false,\"name\":\"clap_builder\",\"req\":\"=4.5.56\"},{\"name\":\"clap_derive\",\"optional\":true,\"req\":\"=4.5.55\"},{\"kind\":\"dev\",\"name\":\"jiff\",\"req\":\"^0.2.3\"},{\"kind\":\"dev\",\"name\":\"rustversion\",\"req\":\"^1.0.15\"},{\"kind\":\"dev\",\"name\":\"semver\",\"req\":\"^1.0.26\"},{\"kind\":\"dev\",\"name\":\"shlex\",\"req\":\"^1.3.0\"},{\"features\":[\"term-svg\"],\"kind\":\"dev\",\"name\":\"snapbox\",\"req\":\"^0.6.16\"},{\"kind\":\"dev\",\"name\":\"trybuild\",\"req\":\"^1.0.91\"},{\"default_features\":false,\"features\":[\"color-auto\",\"diff\",\"examples\"],\"kind\":\"dev\",\"name\":\"trycmd\",\"req\":\"^0.15.3\"}],\"features\":{\"cargo\":[\"clap_builder/cargo\"],\"color\":[\"clap_builder/color\"],\"debug\":[\"clap_builder/debug\",\"clap_derive?/debug\"],\"default\":[\"std\",\"color\",\"help\",\"usage\",\"error-context\",\"suggestions\"],\"deprecated\":[\"clap_builder/deprecated\",\"clap_derive?/deprecated\"],\"derive\":[\"dep:clap_derive\"],\"env\":[\"clap_builder/env\"],\"error-context\":[\"clap_builder/error-context\"],\"help\":[\"clap_builder/help\"],\"std\":[\"clap_builder/std\"],\"string\":[\"clap_builder/string\"],\"suggestions\":[\"clap_builder/suggestions\"],\"unicode\":[\"clap_builder/unicode\"],\"unstable-derive-ui-tests\":[],\"unstable-doc\":[\"clap_builder/unstable-doc\",\"derive\"],\"unstable-ext\":[\"clap_builder/unstable-ext\"],\"unstable-markdown\":[\"clap_derive/unstable-markdown\"],\"unstable-styles\":[\"clap_builder/unstable-styles\"],\"unstable-v5\":[\"clap_builder/unstable-v5\",\"clap_derive?/unstable-v5\",\"deprecated\"],\"usage\":[\"clap_builder/usage\"],\"wrap_help\":[\"clap_builder/wrap_help\"]}}",
-      "clap_builder_4.5.56": "{\"dependencies\":[{\"name\":\"anstream\",\"optional\":true,\"req\":\"^0.6.7\"},{\"name\":\"anstyle\",\"req\":\"^1.0.8\"},{\"name\":\"backtrace\",\"optional\":true,\"req\":\"^0.3.73\"},{\"name\":\"clap_lex\",\"req\":\"^0.7.4\"},{\"kind\":\"dev\",\"name\":\"color-print\",\"req\":\"^0.3.6\"},{\"kind\":\"dev\",\"name\":\"snapbox\",\"req\":\"^0.6.16\"},{\"kind\":\"dev\",\"name\":\"static_assertions\",\"req\":\"^1.1.0\"},{\"name\":\"strsim\",\"optional\":true,\"req\":\"^0.11.0\"},{\"name\":\"terminal_size\",\"optional\":true,\"req\":\"^0.4.0\"},{\"kind\":\"dev\",\"name\":\"unic-emoji-char\",\"req\":\"^0.9.0\"},{\"name\":\"unicase\",\"optional\":true,\"req\":\"^2.6.0\"},{\"name\":\"unicode-width\",\"optional\":true,\"req\":\"^0.2.0\"}],\"features\":{\"cargo\":[],\"color\":[\"dep:anstream\"],\"debug\":[\"dep:backtrace\"],\"default\":[\"std\",\"color\",\"help\",\"usage\",\"error-context\",\"suggestions\"],\"deprecated\":[],\"env\":[],\"error-context\":[],\"help\":[],\"std\":[\"anstyle/std\"],\"string\":[],\"suggestions\":[\"dep:strsim\",\"error-context\"],\"unicode\":[\"dep:unicode-width\",\"dep:unicase\"],\"unstable-doc\":[\"cargo\",\"wrap_help\",\"env\",\"unicode\",\"string\",\"unstable-ext\"],\"unstable-ext\":[],\"unstable-styles\":[\"color\"],\"unstable-v5\":[\"deprecated\"],\"usage\":[],\"wrap_help\":[\"help\",\"dep:terminal_size\"]}}",
+      "clap_4.5.58": "{\"dependencies\":[{\"kind\":\"dev\",\"name\":\"automod\",\"req\":\"^1.0.14\"},{\"default_features\":false,\"kind\":\"dev\",\"name\":\"clap-cargo\",\"req\":\"^0.15.0\"},{\"default_features\":false,\"name\":\"clap_builder\",\"req\":\"=4.5.58\"},{\"name\":\"clap_derive\",\"optional\":true,\"req\":\"=4.5.55\"},{\"kind\":\"dev\",\"name\":\"jiff\",\"req\":\"^0.2.3\"},{\"kind\":\"dev\",\"name\":\"rustversion\",\"req\":\"^1.0.15\"},{\"kind\":\"dev\",\"name\":\"semver\",\"req\":\"^1.0.26\"},{\"kind\":\"dev\",\"name\":\"shlex\",\"req\":\"^1.3.0\"},{\"features\":[\"term-svg\"],\"kind\":\"dev\",\"name\":\"snapbox\",\"req\":\"^0.6.16\"},{\"kind\":\"dev\",\"name\":\"trybuild\",\"req\":\"^1.0.91\"},{\"default_features\":false,\"features\":[\"color-auto\",\"diff\",\"examples\"],\"kind\":\"dev\",\"name\":\"trycmd\",\"req\":\"^0.15.3\"}],\"features\":{\"cargo\":[\"clap_builder/cargo\"],\"color\":[\"clap_builder/color\"],\"debug\":[\"clap_builder/debug\",\"clap_derive?/debug\"],\"default\":[\"std\",\"color\",\"help\",\"usage\",\"error-context\",\"suggestions\"],\"deprecated\":[\"clap_builder/deprecated\",\"clap_derive?/deprecated\"],\"derive\":[\"dep:clap_derive\"],\"env\":[\"clap_builder/env\"],\"error-context\":[\"clap_builder/error-context\"],\"help\":[\"clap_builder/help\"],\"std\":[\"clap_builder/std\"],\"string\":[\"clap_builder/string\"],\"suggestions\":[\"clap_builder/suggestions\"],\"unicode\":[\"clap_builder/unicode\"],\"unstable-derive-ui-tests\":[],\"unstable-doc\":[\"clap_builder/unstable-doc\",\"derive\"],\"unstable-ext\":[\"clap_builder/unstable-ext\"],\"unstable-markdown\":[\"clap_derive/unstable-markdown\"],\"unstable-styles\":[\"clap_builder/unstable-styles\"],\"unstable-v5\":[\"clap_builder/unstable-v5\",\"clap_derive?/unstable-v5\",\"deprecated\"],\"usage\":[\"clap_builder/usage\"],\"wrap_help\":[\"clap_builder/wrap_help\"]}}",
+      "clap_builder_4.5.58": "{\"dependencies\":[{\"name\":\"anstream\",\"optional\":true,\"req\":\"^0.6.7\"},{\"name\":\"anstyle\",\"req\":\"^1.0.8\"},{\"name\":\"backtrace\",\"optional\":true,\"req\":\"^0.3.73\"},{\"name\":\"clap_lex\",\"req\":\"^1.0.0\"},{\"kind\":\"dev\",\"name\":\"color-print\",\"req\":\"^0.3.6\"},{\"kind\":\"dev\",\"name\":\"snapbox\",\"req\":\"^0.6.16\"},{\"kind\":\"dev\",\"name\":\"static_assertions\",\"req\":\"^1.1.0\"},{\"name\":\"strsim\",\"optional\":true,\"req\":\"^0.11.0\"},{\"name\":\"terminal_size\",\"optional\":true,\"req\":\"^0.4.0\"},{\"kind\":\"dev\",\"name\":\"unic-emoji-char\",\"req\":\"^0.9.0\"},{\"name\":\"unicase\",\"optional\":true,\"req\":\"^2.6.0\"},{\"name\":\"unicode-width\",\"optional\":true,\"req\":\"^0.2.0\"}],\"features\":{\"cargo\":[],\"color\":[\"dep:anstream\"],\"debug\":[\"dep:backtrace\"],\"default\":[\"std\",\"color\",\"help\",\"usage\",\"error-context\",\"suggestions\"],\"deprecated\":[],\"env\":[],\"error-context\":[],\"help\":[],\"std\":[\"anstyle/std\"],\"string\":[],\"suggestions\":[\"dep:strsim\",\"error-context\"],\"unicode\":[\"dep:unicode-width\",\"dep:unicase\"],\"unstable-doc\":[\"cargo\",\"wrap_help\",\"env\",\"unicode\",\"string\",\"unstable-ext\"],\"unstable-ext\":[],\"unstable-styles\":[\"color\"],\"unstable-v5\":[\"deprecated\"],\"usage\":[],\"wrap_help\":[\"help\",\"dep:terminal_size\"]}}",
       "clap_complete_4.5.65": "{\"dependencies\":[{\"kind\":\"dev\",\"name\":\"automod\",\"req\":\"^1.0.14\"},{\"default_features\":false,\"features\":[\"std\"],\"name\":\"clap\",\"req\":\"^4.5.20\"},{\"default_features\":false,\"features\":[\"std\",\"derive\",\"help\"],\"kind\":\"dev\",\"name\":\"clap\",\"req\":\"^4.5.20\"},{\"name\":\"clap_lex\",\"optional\":true,\"req\":\"^0.7.0\"},{\"name\":\"completest\",\"optional\":true,\"req\":\"^0.4.2\"},{\"name\":\"completest-pty\",\"optional\":true,\"req\":\"^0.5.5\"},{\"name\":\"is_executable\",\"optional\":true,\"req\":\"^1.0.1\"},{\"name\":\"shlex\",\"optional\":true,\"req\":\"^1.3.0\"},{\"features\":[\"diff\",\"dir\",\"examples\"],\"kind\":\"dev\",\"name\":\"snapbox\",\"req\":\"^0.6.0\"},{\"default_features\":false,\"features\":[\"color-auto\",\"diff\",\"examples\"],\"kind\":\"dev\",\"name\":\"trycmd\",\"req\":\"^0.15.1\"}],\"features\":{\"debug\":[\"clap/debug\"],\"default\":[],\"unstable-doc\":[\"unstable-dynamic\"],\"unstable-dynamic\":[\"dep:clap_lex\",\"dep:shlex\",\"dep:is_executable\",\"clap/unstable-ext\"],\"unstable-shell-tests\":[\"dep:completest\",\"dep:completest-pty\"]}}",
       "clap_derive_4.5.55": "{\"dependencies\":[{\"name\":\"anstyle\",\"optional\":true,\"req\":\"^1.0.10\"},{\"name\":\"heck\",\"req\":\"^0.5.0\"},{\"name\":\"proc-macro2\",\"req\":\"^1.0.69\"},{\"default_features\":false,\"name\":\"pulldown-cmark\",\"optional\":true,\"req\":\"^0.13.0\"},{\"name\":\"quote\",\"req\":\"^1.0.9\"},{\"features\":[\"full\"],\"name\":\"syn\",\"req\":\"^2.0.8\"}],\"features\":{\"debug\":[],\"default\":[],\"deprecated\":[],\"raw-deprecated\":[\"deprecated\"],\"unstable-markdown\":[\"dep:pulldown-cmark\",\"dep:anstyle\"],\"unstable-v5\":[\"deprecated\"]}}",
-      "clap_lex_0.7.7": "{\"dependencies\":[{\"kind\":\"dev\",\"name\":\"automod\",\"req\":\"^1.0.14\"}],\"features\":{}}",
+      "clap_lex_1.0.0": "{\"dependencies\":[{\"kind\":\"dev\",\"name\":\"automod\",\"req\":\"^1.0.14\"}],\"features\":{}}",
       "clipboard-win_5.4.1": "{\"dependencies\":[{\"name\":\"error-code\",\"req\":\"^3\",\"target\":\"cfg(windows)\"},{\"name\":\"windows-win\",\"optional\":true,\"req\":\"^3\",\"target\":\"cfg(windows)\"}],\"features\":{\"monitor\":[\"windows-win\"],\"std\":[\"error-code/std\"]}}",
       "cmake_0.1.57": "{\"dependencies\":[{\"name\":\"cc\",\"req\":\"^1.2.46\"}],\"features\":{}}",
       "cmp_any_0.8.1": "{\"dependencies\":[],\"features\":{}}",
@@ -790,9 +790,9 @@
       "enumflags2_0.7.12": "{\"dependencies\":[{\"name\":\"enumflags2_derive\",\"req\":\"=0.7.12\"},{\"default_features\":false,\"name\":\"serde\",\"optional\":true,\"req\":\"^1.0.0\"}],\"features\":{\"std\":[]}}",
       "enumflags2_derive_0.7.12": "{\"dependencies\":[{\"name\":\"proc-macro2\",\"req\":\"^1.0\"},{\"name\":\"quote\",\"req\":\"^1.0\"},{\"default_features\":false,\"features\":[\"parsing\",\"printing\",\"derive\",\"proc-macro\"],\"name\":\"syn\",\"req\":\"^2.0\"}],\"features\":{}}",
       "env-flags_0.1.1": "{\"dependencies\":[],\"features\":{}}",
-      "env_filter_0.1.4": "{\"dependencies\":[{\"features\":[\"std\"],\"name\":\"log\",\"req\":\"^0.4.8\"},{\"default_features\":false,\"features\":[\"std\",\"perf\"],\"name\":\"regex\",\"optional\":true,\"req\":\"^1.0.3\"},{\"kind\":\"dev\",\"name\":\"snapbox\",\"req\":\"^0.6\"}],\"features\":{\"default\":[\"regex\"],\"regex\":[\"dep:regex\"]}}",
+      "env_filter_1.0.0": "{\"dependencies\":[{\"features\":[\"std\"],\"name\":\"log\",\"req\":\"^0.4.8\"},{\"default_features\":false,\"features\":[\"std\",\"perf\"],\"name\":\"regex\",\"optional\":true,\"req\":\"^1.0.3\"},{\"kind\":\"dev\",\"name\":\"snapbox\",\"req\":\"^0.6\"}],\"features\":{\"default\":[\"regex\"],\"regex\":[\"dep:regex\"]}}",
       "env_home_0.1.0": "{\"dependencies\":[],\"features\":{}}",
-      "env_logger_0.11.8": "{\"dependencies\":[{\"default_features\":false,\"features\":[\"wincon\"],\"name\":\"anstream\",\"optional\":true,\"req\":\"^0.6.11\"},{\"name\":\"anstyle\",\"optional\":true,\"req\":\"^1.0.6\"},{\"default_features\":false,\"name\":\"env_filter\",\"req\":\"^0.1.0\"},{\"default_features\":false,\"features\":[\"std\"],\"name\":\"jiff\",\"optional\":true,\"req\":\"^0.2.3\"},{\"features\":[\"std\"],\"name\":\"log\",\"req\":\"^0.4.21\"}],\"features\":{\"auto-color\":[\"color\",\"anstream/auto\"],\"color\":[\"dep:anstream\",\"dep:anstyle\"],\"default\":[\"auto-color\",\"humantime\",\"regex\"],\"humantime\":[\"dep:jiff\"],\"kv\":[\"log/kv\"],\"regex\":[\"env_filter/regex\"],\"unstable-kv\":[\"kv\"]}}",
+      "env_logger_0.11.9": "{\"dependencies\":[{\"default_features\":false,\"features\":[\"wincon\"],\"name\":\"anstream\",\"optional\":true,\"req\":\"^0.6.11\"},{\"name\":\"anstyle\",\"optional\":true,\"req\":\"^1.0.6\"},{\"default_features\":false,\"name\":\"env_filter\",\"req\":\"^1.0.0\"},{\"default_features\":false,\"features\":[\"std\"],\"name\":\"jiff\",\"optional\":true,\"req\":\"^0.2.3\"},{\"features\":[\"std\"],\"name\":\"log\",\"req\":\"^0.4.21\"}],\"features\":{\"auto-color\":[\"color\",\"anstream/auto\"],\"color\":[\"dep:anstream\",\"dep:anstyle\"],\"default\":[\"auto-color\",\"humantime\",\"regex\"],\"humantime\":[\"dep:jiff\"],\"kv\":[\"log/kv\"],\"regex\":[\"env_filter/regex\"],\"unstable-kv\":[\"kv\"]}}",
       "equivalent_1.0.2": "{\"dependencies\":[],\"features\":{}}",
       "erased-serde_0.3.31": "{\"dependencies\":[{\"kind\":\"dev\",\"name\":\"rustversion\",\"req\":\"^1.0.13\"},{\"default_features\":false,\"name\":\"serde\",\"req\":\"^1.0.166\"},{\"kind\":\"dev\",\"name\":\"serde_cbor\",\"req\":\"^0.11.2\"},{\"kind\":\"dev\",\"name\":\"serde_derive\",\"req\":\"^1.0.166\"},{\"kind\":\"dev\",\"name\":\"serde_json\",\"req\":\"^1.0.99\"},{\"features\":[\"diff\"],\"kind\":\"dev\",\"name\":\"trybuild\",\"req\":\"^1.0.83\"}],\"features\":{\"alloc\":[\"serde/alloc\"],\"default\":[\"std\"],\"std\":[\"serde/std\"],\"unstable-debug\":[]}}",
       "errno_0.3.14": "{\"dependencies\":[{\"default_features\":false,\"name\":\"libc\",\"req\":\"^0.2\",\"target\":\"cfg(target_os=\\\"hermit\\\")\"},{\"default_features\":false,\"name\":\"libc\",\"req\":\"^0.2\",\"target\":\"cfg(target_os=\\\"wasi\\\")\"},{\"default_features\":false,\"name\":\"libc\",\"req\":\"^0.2\",\"target\":\"cfg(unix)\"},{\"features\":[\"Win32_Foundation\",\"Win32_System_Diagnostics_Debug\"],\"name\":\"windows-sys\",\"req\":\">=0.52, <0.62\",\"target\":\"cfg(windows)\"}],\"features\":{\"default\":[\"std\"],\"std\":[\"libc/std\"]}}",

codex-rs/codex-api/src/endpoint/mod.rs

@@ -2,6 +2,8 @@ pub mod aggregate;
 pub mod compact;
 pub mod memories;
 pub mod models;
+pub mod realtime_websocket;
 pub mod responses;
 pub mod responses_websocket;
 mod session;
+mod websocket_pump;

codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs

@@ -0,0 +1,735 @@
+use crate::endpoint::realtime_websocket::protocol::ConversationItem;
+use crate::endpoint::realtime_websocket::protocol::ConversationItemContent;
+use crate::endpoint::realtime_websocket::protocol::RealtimeAudioFrame;
+use crate::endpoint::realtime_websocket::protocol::RealtimeEvent;
+use crate::endpoint::realtime_websocket::protocol::RealtimeOutboundMessage;
+use crate::endpoint::realtime_websocket::protocol::RealtimeSessionConfig;
+use crate::endpoint::realtime_websocket::protocol::SessionCreateSession;
+use crate::endpoint::realtime_websocket::protocol::SessionUpdateSession;
+use crate::endpoint::realtime_websocket::protocol::parse_realtime_event;
+use crate::endpoint::websocket_pump::WebsocketMessage;
+use crate::endpoint::websocket_pump::WebsocketPump;
+use crate::error::ApiError;
+use crate::provider::Provider;
+use codex_utils_rustls_provider::ensure_rustls_crypto_provider;
+use http::HeaderMap;
+use std::sync::Arc;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::Ordering;
+use tokio::sync::Mutex;
+use tokio::sync::mpsc;
+use tokio_tungstenite::tungstenite::Error as WsError;
+use tokio_tungstenite::tungstenite::Message;
+use tokio_tungstenite::tungstenite::client::IntoClientRequest;
+use tracing::info;
+use tracing::trace;
+use tungstenite::protocol::WebSocketConfig;
+use url::Url;
+
+pub struct RealtimeWebsocketConnection {
+    writer: RealtimeWebsocketWriter,
+    events: RealtimeWebsocketEvents,
+}
+
+#[derive(Clone)]
+pub struct RealtimeWebsocketWriter {
+    stream: Arc<WebsocketPump>,
+    is_closed: Arc<AtomicBool>,
+}
+
+#[derive(Clone)]
+pub struct RealtimeWebsocketEvents {
+    rx_message: Arc<Mutex<mpsc::UnboundedReceiver<WebsocketMessage>>>,
+    is_closed: Arc<AtomicBool>,
+}
+
+impl RealtimeWebsocketConnection {
+    pub async fn send_audio_frame(&self, frame: RealtimeAudioFrame) -> Result<(), ApiError> {
+        self.writer.send_audio_frame(frame).await
+    }
+
+    pub async fn send_conversation_item_create(&self, text: String) -> Result<(), ApiError> {
+        self.writer.send_conversation_item_create(text).await
+    }
+
+    pub async fn send_session_update(
+        &self,
+        backend_prompt: String,
+        conversation_id: Option<String>,
+    ) -> Result<(), ApiError> {
+        self.writer
+            .send_session_update(backend_prompt, conversation_id)
+            .await
+    }
+
+    pub async fn send_session_create(
+        &self,
+        backend_prompt: String,
+        conversation_id: Option<String>,
+    ) -> Result<(), ApiError> {
+        self.writer
+            .send_session_create(backend_prompt, conversation_id)
+            .await
+    }
+
+    pub async fn close(&self) -> Result<(), ApiError> {
+        self.writer.close().await
+    }
+
+    pub async fn next_event(&self) -> Result<Option<RealtimeEvent>, ApiError> {
+        self.events.next_event().await
+    }
+
+    pub fn writer(&self) -> RealtimeWebsocketWriter {
+        self.writer.clone()
+    }
+
+    pub fn events(&self) -> RealtimeWebsocketEvents {
+        self.events.clone()
+    }
+
+    fn new(stream: WebsocketPump, rx_message: mpsc::UnboundedReceiver<WebsocketMessage>) -> Self {
+        let stream = Arc::new(stream);
+        let is_closed = Arc::new(AtomicBool::new(false));
+        Self {
+            writer: RealtimeWebsocketWriter {
+                stream: Arc::clone(&stream),
+                is_closed: Arc::clone(&is_closed),
+            },
+            events: RealtimeWebsocketEvents {
+                rx_message: Arc::new(Mutex::new(rx_message)),
+                is_closed,
+            },
+        }
+    }
+}
+
+impl RealtimeWebsocketWriter {
+    pub async fn send_audio_frame(&self, frame: RealtimeAudioFrame) -> Result<(), ApiError> {
+        self.send_json(RealtimeOutboundMessage::InputAudioDelta {
+            delta: frame.data,
+            sample_rate: frame.sample_rate,
+            num_channels: frame.num_channels,
+            samples_per_channel: frame.samples_per_channel,
+        })
+        .await
+    }
+
+    pub async fn send_conversation_item_create(&self, text: String) -> Result<(), ApiError> {
+        self.send_json(RealtimeOutboundMessage::ConversationItemCreate {
+            item: ConversationItem {
+                kind: "message".to_string(),
+                role: "user".to_string(),
+                content: vec![ConversationItemContent {
+                    kind: "text".to_string(),
+                    text,
+                }],
+            },
+        })
+        .await
+    }
+
+    pub async fn send_session_update(
+        &self,
+        backend_prompt: String,
+        conversation_id: Option<String>,
+    ) -> Result<(), ApiError> {
+        self.send_json(RealtimeOutboundMessage::SessionUpdate {
+            session: Some(SessionUpdateSession {
+                backend_prompt,
+                conversation_id,
+            }),
+        })
+        .await
+    }
+
+    pub async fn send_session_create(
+        &self,
+        backend_prompt: String,
+        conversation_id: Option<String>,
+    ) -> Result<(), ApiError> {
+        self.send_json(RealtimeOutboundMessage::SessionCreate {
+            session: SessionCreateSession {
+                backend_prompt,
+                conversation_id,
+            },
+        })
+        .await
+    }
+
+    pub async fn close(&self) -> Result<(), ApiError> {
+        if self.is_closed.swap(true, Ordering::SeqCst) {
+            return Ok(());
+        }
+        if let Err(err) = self.stream.close().await
+            && !matches!(err, WsError::ConnectionClosed | WsError::AlreadyClosed)
+        {
+            return Err(ApiError::Stream(format!(
+                "failed to close websocket: {err}"
+            )));
+        }
+        Ok(())
+    }
+
+    async fn send_json(&self, message: RealtimeOutboundMessage) -> Result<(), ApiError> {
+        let payload = serde_json::to_string(&message)
+            .map_err(|err| ApiError::Stream(format!("failed to encode realtime request: {err}")))?;
+        trace!("realtime websocket request: {payload}");
+
+        if self.is_closed.load(Ordering::SeqCst) {
+            return Err(ApiError::Stream(
+                "realtime websocket connection is closed".to_string(),
+            ));
+        }
+
+        self.stream
+            .send(Message::Text(payload.into()))
+            .await
+            .map_err(|err| ApiError::Stream(format!("failed to send realtime request: {err}")))?;
+        Ok(())
+    }
+}
+
+impl RealtimeWebsocketEvents {
+    pub async fn next_event(&self) -> Result<Option<RealtimeEvent>, ApiError> {
+        if self.is_closed.load(Ordering::SeqCst) {
+            return Ok(None);
+        }
+
+        loop {
+            let msg = match self.rx_message.lock().await.recv().await {
+                Some(Ok(msg)) => msg,
+                Some(Err(err)) => {
+                    self.is_closed.store(true, Ordering::SeqCst);
+                    return Err(ApiError::Stream(format!(
+                        "failed to read websocket message: {err}"
+                    )));
+                }
+                None => {
+                    self.is_closed.store(true, Ordering::SeqCst);
+                    return Ok(None);
+                }
+            };
+
+            match msg {
+                Message::Text(text) => {
+                    if let Some(event) = parse_realtime_event(&text) {
+                        return Ok(Some(event));
+                    }
+                }
+                Message::Close(_) => {
+                    self.is_closed.store(true, Ordering::SeqCst);
+                    return Ok(None);
+                }
+                Message::Binary(_) => {
+                    return Ok(Some(RealtimeEvent::Error(
+                        "unexpected binary realtime websocket event".to_string(),
+                    )));
+                }
+                Message::Frame(_) | Message::Ping(_) | Message::Pong(_) => {}
+            }
+        }
+    }
+}
+
+pub struct RealtimeWebsocketClient {
+    provider: Provider,
+}
+
+impl RealtimeWebsocketClient {
+    pub fn new(provider: Provider) -> Self {
+        Self { provider }
+    }
+
+    pub async fn connect(
+        &self,
+        config: RealtimeSessionConfig,
+        extra_headers: HeaderMap,
+        default_headers: HeaderMap,
+    ) -> Result<RealtimeWebsocketConnection, ApiError> {
+        ensure_rustls_crypto_provider();
+        let ws_url = websocket_url_from_api_url(
+            config.api_url.as_str(),
+            self.provider.query_params.as_ref(),
+        )?;
+
+        let mut request = ws_url
+            .as_str()
+            .into_client_request()
+            .map_err(|err| ApiError::Stream(format!("failed to build websocket request: {err}")))?;
+        let headers = merge_request_headers(&self.provider.headers, extra_headers, default_headers);
+        request.headers_mut().extend(headers);
+
+        info!("connecting realtime websocket: {ws_url}");
+        let (stream, _) =
+            tokio_tungstenite::connect_async_with_config(request, Some(websocket_config()), false)
+                .await
+                .map_err(|err| {
+                    ApiError::Stream(format!("failed to connect realtime websocket: {err}"))
+                })?;
+
+        let (stream, rx_message) = WebsocketPump::new(stream);
+        let connection = RealtimeWebsocketConnection::new(stream, rx_message);
+        connection
+            .send_session_create(config.prompt, config.conversation_id)
+            .await?;
+        Ok(connection)
+    }
+}
+
+fn merge_request_headers(
+    provider_headers: &HeaderMap,
+    extra_headers: HeaderMap,
+    default_headers: HeaderMap,
+) -> HeaderMap {
+    let mut headers = provider_headers.clone();
+    headers.extend(extra_headers);
+    for (name, value) in &default_headers {
+        if let http::header::Entry::Vacant(entry) = headers.entry(name) {
+            entry.insert(value.clone());
+        }
+    }
+    headers
+}
+
+fn websocket_config() -> WebSocketConfig {
+    WebSocketConfig::default()
+}
+
+fn websocket_url_from_api_url(
+    api_url: &str,
+    query_params: Option<&std::collections::HashMap<String, String>>,
+) -> Result<Url, ApiError> {
+    let mut url = Url::parse(api_url)
+        .map_err(|err| ApiError::Stream(format!("failed to parse realtime api_url: {err}")))?;
+
+    match url.scheme() {
+        "ws" | "wss" => {
+            if url.path().is_empty() || url.path() == "/" {
+                url.set_path("/ws");
+            }
+            Ok(url)
+        }
+        "http" | "https" => {
+            if url.path().is_empty() || url.path() == "/" {
+                url.set_path("/ws");
+            }
+            let scheme = if url.scheme() == "http" { "ws" } else { "wss" };
+            let _ = url.set_scheme(scheme);
+            Ok(url)
+        }
+        scheme => Err(ApiError::Stream(format!(
+            "unsupported realtime api_url scheme: {scheme}"
+        ))),
+    }?;
+
+    if let Some(params) = query_params
+        && !params.is_empty()
+    {
+        let mut url_query = url.query_pairs_mut();
+        for (key, value) in params {
+            url_query.append_pair(key, value);
+        }
+    }
+
+    Ok(url)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use futures::SinkExt;
+    use futures::StreamExt;
+    use http::HeaderValue;
+    use pretty_assertions::assert_eq;
+    use serde_json::Value;
+    use serde_json::json;
+    use std::collections::HashMap;
+    use std::time::Duration;
+    use tokio::net::TcpListener;
+    use tokio_tungstenite::accept_async;
+    use tokio_tungstenite::tungstenite::Message;
+
+    #[test]
+    fn parse_session_created_event() {
+        let payload = json!({
+            "type": "session.created",
+            "session": {"id": "sess_123"}
+        })
+        .to_string();
+
+        assert_eq!(
+            parse_realtime_event(payload.as_str()),
+            Some(RealtimeEvent::SessionCreated {
+                session_id: "sess_123".to_string()
+            })
+        );
+    }
+
+    #[test]
+    fn parse_audio_delta_event() {
+        let payload = json!({
+            "type": "response.output_audio.delta",
+            "delta": "AAA=",
+            "sample_rate": 48000,
+            "num_channels": 1,
+            "samples_per_channel": 960
+        })
+        .to_string();
+        assert_eq!(
+            parse_realtime_event(payload.as_str()),
+            Some(RealtimeEvent::AudioOut(RealtimeAudioFrame {
+                data: "AAA=".to_string(),
+                sample_rate: 48000,
+                num_channels: 1,
+                samples_per_channel: Some(960),
+            }))
+        );
+    }
+
+    #[test]
+    fn parse_conversation_item_added_event() {
+        let payload = json!({
+            "type": "conversation.item.added",
+            "item": {"type": "spawn_transcript", "seq": 7}
+        })
+        .to_string();
+        assert_eq!(
+            parse_realtime_event(payload.as_str()),
+            Some(RealtimeEvent::ConversationItemAdded(
+                json!({"type": "spawn_transcript", "seq": 7})
+            ))
+        );
+    }
+
+    #[test]
+    fn merge_request_headers_matches_http_precedence() {
+        let mut provider_headers = HeaderMap::new();
+        provider_headers.insert(
+            "originator",
+            HeaderValue::from_static("provider-originator"),
+        );
+        provider_headers.insert("x-priority", HeaderValue::from_static("provider"));
+
+        let mut extra_headers = HeaderMap::new();
+        extra_headers.insert("x-priority", HeaderValue::from_static("extra"));
+
+        let mut default_headers = HeaderMap::new();
+        default_headers.insert("originator", HeaderValue::from_static("default-originator"));
+        default_headers.insert("x-priority", HeaderValue::from_static("default"));
+        default_headers.insert("x-default-only", HeaderValue::from_static("default-only"));
+
+        let merged = merge_request_headers(&provider_headers, extra_headers, default_headers);
+
+        assert_eq!(
+            merged.get("originator"),
+            Some(&HeaderValue::from_static("provider-originator"))
+        );
+        assert_eq!(
+            merged.get("x-priority"),
+            Some(&HeaderValue::from_static("extra"))
+        );
+        assert_eq!(
+            merged.get("x-default-only"),
+            Some(&HeaderValue::from_static("default-only"))
+        );
+    }
+
+    #[test]
+    fn websocket_url_from_http_base_defaults_to_ws_path() {
+        let url = websocket_url_from_api_url("http://127.0.0.1:8011", None).expect("build ws url");
+        assert_eq!(url.as_str(), "ws://127.0.0.1:8011/ws");
+    }
+
+    #[test]
+    fn websocket_url_from_ws_base_defaults_to_ws_path() {
+        let url = websocket_url_from_api_url("wss://example.com", None).expect("build ws url");
+        assert_eq!(url.as_str(), "wss://example.com/ws");
+    }
+
+    #[test]
+    fn websocket_url_includes_provider_query_params() {
+        let mut query_params = HashMap::new();
+        query_params.insert("api-version".to_string(), "2024-10-01-preview".to_string());
+
+        let url = websocket_url_from_api_url("https://example.com/ws", Some(&query_params))
+            .expect("build ws url");
+        let api_version = url
+            .query_pairs()
+            .find(|(key, _)| key == "api-version")
+            .map(|(_, value)| value.into_owned());
+
+        assert_eq!(url.scheme(), "wss");
+        assert_eq!(api_version, Some("2024-10-01-preview".to_string()));
+    }
+
+    #[tokio::test]
+    async fn e2e_connect_and_exchange_events_against_mock_ws_server() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
+        let addr = listener.local_addr().expect("local addr");
+
+        let server = tokio::spawn(async move {
+            let (stream, _) = listener.accept().await.expect("accept");
+            let mut ws = accept_async(stream).await.expect("accept ws");
+
+            let first = ws
+                .next()
+                .await
+                .expect("first msg")
+                .expect("first msg ok")
+                .into_text()
+                .expect("text");
+            let first_json: Value = serde_json::from_str(&first).expect("json");
+            assert_eq!(first_json["type"], "session.create");
+            assert_eq!(
+                first_json["session"]["backend_prompt"],
+                Value::String("backend prompt".to_string())
+            );
+            assert_eq!(
+                first_json["session"]["conversation_id"],
+                Value::String("conv_1".to_string())
+            );
+
+            ws.send(Message::Text(
+                json!({
+                    "type": "session.created",
+                    "session": {"id": "sess_mock"}
+                })
+                .to_string()
+                .into(),
+            ))
+            .await
+            .expect("send session.created");
+
+            let second = ws
+                .next()
+                .await
+                .expect("second msg")
+                .expect("second msg ok")
+                .into_text()
+                .expect("text");
+            let second_json: Value = serde_json::from_str(&second).expect("json");
+            assert_eq!(second_json["type"], "response.input_audio.delta");
+
+            let third = ws
+                .next()
+                .await
+                .expect("third msg")
+                .expect("third msg ok")
+                .into_text()
+                .expect("text");
+            let third_json: Value = serde_json::from_str(&third).expect("json");
+            assert_eq!(third_json["type"], "conversation.item.create");
+            assert_eq!(third_json["item"]["content"][0]["text"], "hello agent");
+
+            ws.send(Message::Text(
+                json!({
+                    "type": "response.output_audio.delta",
+                    "delta": "AQID",
+                    "sample_rate": 48000,
+                    "num_channels": 1
+                })
+                .to_string()
+                .into(),
+            ))
+            .await
+            .expect("send audio");
+
+            ws.send(Message::Text(
+                json!({
+                    "type": "conversation.item.added",
+                    "item": {"type": "spawn_transcript", "seq": 2}
+                })
+                .to_string()
+                .into(),
+            ))
+            .await
+            .expect("send item added");
+        });
+
+        let provider = Provider {
+            name: "test".to_string(),
+            base_url: "http://localhost".to_string(),
+            query_params: Some(HashMap::new()),
+            headers: HeaderMap::new(),
+            retry: crate::provider::RetryConfig {
+                max_attempts: 1,
+                base_delay: Duration::from_millis(1),
+                retry_429: false,
+                retry_5xx: false,
+                retry_transport: false,
+            },
+            stream_idle_timeout: Duration::from_secs(5),
+        };
+        let client = RealtimeWebsocketClient::new(provider);
+        let connection = client
+            .connect(
+                RealtimeSessionConfig {
+                    api_url: format!("ws://{addr}"),
+                    prompt: "backend prompt".to_string(),
+                    conversation_id: Some("conv_1".to_string()),
+                },
+                HeaderMap::new(),
+                HeaderMap::new(),
+            )
+            .await
+            .expect("connect");
+
+        let created = connection
+            .next_event()
+            .await
+            .expect("next event")
+            .expect("event");
+        assert_eq!(
+            created,
+            RealtimeEvent::SessionCreated {
+                session_id: "sess_mock".to_string()
+            }
+        );
+
+        connection
+            .send_audio_frame(RealtimeAudioFrame {
+                data: "AQID".to_string(),
+                sample_rate: 48000,
+                num_channels: 1,
+                samples_per_channel: Some(960),
+            })
+            .await
+            .expect("send audio");
+        connection
+            .send_conversation_item_create("hello agent".to_string())
+            .await
+            .expect("send item");
+
+        let audio_event = connection
+            .next_event()
+            .await
+            .expect("next event")
+            .expect("event");
+        assert_eq!(
+            audio_event,
+            RealtimeEvent::AudioOut(RealtimeAudioFrame {
+                data: "AQID".to_string(),
+                sample_rate: 48000,
+                num_channels: 1,
+                samples_per_channel: None,
+            })
+        );
+
+        let added_event = connection
+            .next_event()
+            .await
+            .expect("next event")
+            .expect("event");
+        assert_eq!(
+            added_event,
+            RealtimeEvent::ConversationItemAdded(json!({
+                "type": "spawn_transcript",
+                "seq": 2
+            }))
+        );
+
+        connection.close().await.expect("close");
+        server.await.expect("server task");
+    }
+
+    #[tokio::test]
+    async fn send_does_not_block_while_next_event_waits_for_inbound_data() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
+        let addr = listener.local_addr().expect("local addr");
+
+        let server = tokio::spawn(async move {
+            let (stream, _) = listener.accept().await.expect("accept");
+            let mut ws = accept_async(stream).await.expect("accept ws");
+
+            let first = ws
+                .next()
+                .await
+                .expect("first msg")
+                .expect("first msg ok")
+                .into_text()
+                .expect("text");
+            let first_json: Value = serde_json::from_str(&first).expect("json");
+            assert_eq!(first_json["type"], "session.create");
+
+            let second = ws
+                .next()
+                .await
+                .expect("second msg")
+                .expect("second msg ok")
+                .into_text()
+                .expect("text");
+            let second_json: Value = serde_json::from_str(&second).expect("json");
+            assert_eq!(second_json["type"], "response.input_audio.delta");
+
+            ws.send(Message::Text(
+                json!({
+                    "type": "session.created",
+                    "session": {"id": "sess_after_send"}
+                })
+                .to_string()
+                .into(),
+            ))
+            .await
+            .expect("send session.created");
+        });
+
+        let provider = Provider {
+            name: "test".to_string(),
+            base_url: "http://localhost".to_string(),
+            query_params: Some(HashMap::new()),
+            headers: HeaderMap::new(),
+            retry: crate::provider::RetryConfig {
+                max_attempts: 1,
+                base_delay: Duration::from_millis(1),
+                retry_429: false,
+                retry_5xx: false,
+                retry_transport: false,
+            },
+            stream_idle_timeout: Duration::from_secs(5),
+        };
+        let client = RealtimeWebsocketClient::new(provider);
+        let connection = client
+            .connect(
+                RealtimeSessionConfig {
+                    api_url: format!("ws://{addr}"),
+                    prompt: "backend prompt".to_string(),
+                    conversation_id: Some("conv_1".to_string()),
+                },
+                HeaderMap::new(),
+                HeaderMap::new(),
+            )
+            .await
+            .expect("connect");
+
+        let (send_result, next_result) = tokio::join!(
+            async {
+                tokio::time::timeout(
+                    Duration::from_millis(200),
+                    connection.send_audio_frame(RealtimeAudioFrame {
+                        data: "AQID".to_string(),
+                        sample_rate: 48000,
+                        num_channels: 1,
+                        samples_per_channel: Some(960),
+                    }),
+                )
+                .await
+            },
+            connection.next_event()
+        );
+
+        send_result
+            .expect("send should not block on next_event")
+            .expect("send audio");
+        let next_event = next_result.expect("next event").expect("event");
+        assert_eq!(
+            next_event,
+            RealtimeEvent::SessionCreated {
+                session_id: "sess_after_send".to_string()
+            }
+        );
+
+        connection.close().await.expect("close");
+        server.await.expect("server task");
+    }
+}

codex-rs/codex-api/src/endpoint/realtime_websocket/mod.rs

@@ -0,0 +1,10 @@
+pub mod methods;
+pub mod protocol;
+
+pub use methods::RealtimeWebsocketClient;
+pub use methods::RealtimeWebsocketConnection;
+pub use methods::RealtimeWebsocketEvents;
+pub use methods::RealtimeWebsocketWriter;
+pub use protocol::RealtimeAudioFrame;
+pub use protocol::RealtimeEvent;
+pub use protocol::RealtimeSessionConfig;

codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs

@@ -0,0 +1,146 @@
+use serde::Deserialize;
+use serde::Serialize;
+use serde_json::Value;
+use tracing::debug;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RealtimeSessionConfig {
+    pub api_url: String,
+    pub prompt: String,
+    pub conversation_id: Option<String>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct RealtimeAudioFrame {
+    pub data: String,
+    pub sample_rate: u32,
+    pub num_channels: u16,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub samples_per_channel: Option<u32>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum RealtimeEvent {
+    SessionCreated { session_id: String },
+    SessionUpdated { backend_prompt: Option<String> },
+    AudioOut(RealtimeAudioFrame),
+    ConversationItemAdded(Value),
+    Error(String),
+}
+
+#[derive(Debug, Clone, Serialize)]
+#[serde(tag = "type")]
+pub(super) enum RealtimeOutboundMessage {
+    #[serde(rename = "response.input_audio.delta")]
+    InputAudioDelta {
+        delta: String,
+        sample_rate: u32,
+        num_channels: u16,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        samples_per_channel: Option<u32>,
+    },
+    #[serde(rename = "session.create")]
+    SessionCreate { session: SessionCreateSession },
+    #[serde(rename = "session.update")]
+    SessionUpdate {
+        #[serde(skip_serializing_if = "Option::is_none")]
+        session: Option<SessionUpdateSession>,
+    },
+    #[serde(rename = "conversation.item.create")]
+    ConversationItemCreate { item: ConversationItem },
+}
+
+#[derive(Debug, Clone, Serialize)]
+pub(super) struct SessionUpdateSession {
+    pub(super) backend_prompt: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(super) conversation_id: Option<String>,
+}
+
+#[derive(Debug, Clone, Serialize)]
+pub(super) struct SessionCreateSession {
+    pub(super) backend_prompt: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(super) conversation_id: Option<String>,
+}
+
+#[derive(Debug, Clone, Serialize)]
+pub(super) struct ConversationItem {
+    #[serde(rename = "type")]
+    pub(super) kind: String,
+    pub(super) role: String,
+    pub(super) content: Vec<ConversationItemContent>,
+}
+
+#[derive(Debug, Clone, Serialize)]
+pub(super) struct ConversationItemContent {
+    #[serde(rename = "type")]
+    pub(super) kind: String,
+    pub(super) text: String,
+}
+
+pub(super) fn parse_realtime_event(payload: &str) -> Option<RealtimeEvent> {
+    let parsed: Value = match serde_json::from_str(payload) {
+        Ok(value) => value,
+        Err(err) => {
+            debug!("failed to parse realtime event: {err}, data: {payload}");
+            return None;
+        }
+    };
+
+    let event_type = parsed.get("type")?.as_str()?;
+    match event_type {
+        "session.created" => {
+            let session_id = parsed
+                .pointer("/session/id")
+                .and_then(Value::as_str)
+                .or_else(|| parsed.get("session_id").and_then(Value::as_str))?;
+            Some(RealtimeEvent::SessionCreated {
+                session_id: session_id.to_string(),
+            })
+        }
+        "session.updated" => Some(RealtimeEvent::SessionUpdated {
+            backend_prompt: parsed
+                .pointer("/session/backend_prompt")
+                .and_then(Value::as_str)
+                .map(ToString::to_string),
+        }),
+        "response.output_audio.delta" => {
+            let data = parsed
+                .get("delta")
+                .and_then(Value::as_str)
+                .or_else(|| parsed.get("data").and_then(Value::as_str))?;
+            let sample_rate = parsed
+                .get("sample_rate")
+                .and_then(Value::as_u64)
+                .and_then(|value| u32::try_from(value).ok())?;
+            let num_channels = parsed
+                .get("num_channels")
+                .and_then(Value::as_u64)
+                .and_then(|value| u16::try_from(value).ok())?;
+            let samples_per_channel = parsed
+                .get("samples_per_channel")
+                .and_then(Value::as_u64)
+                .and_then(|value| u32::try_from(value).ok());
+            Some(RealtimeEvent::AudioOut(RealtimeAudioFrame {
+                data: data.to_string(),
+                sample_rate,
+                num_channels,
+                samples_per_channel,
+            }))
+        }
+        "conversation.item.added" => parsed
+            .get("item")
+            .cloned()
+            .map(RealtimeEvent::ConversationItemAdded),
+        "error" => {
+            let message = parsed
+                .get("message")
+                .and_then(Value::as_str)
+                .map(ToString::to_string)
+                .or_else(|| parsed.get("error").map(ToString::to_string))?;
+            Some(RealtimeEvent::Error(message))
+        }
+        _ => None,
+    }
+}

codex-rs/codex-api/src/endpoint/responses_websocket.rs

@@ -3,6 +3,8 @@ use crate::auth::add_auth_headers_to_header_map;
 use crate::common::ResponseEvent;
 use crate::common::ResponseStream;
 use crate::common::ResponsesWsRequest;
+use crate::endpoint::websocket_pump::WebsocketMessage;
+use crate::endpoint::websocket_pump::WebsocketPump;
 use crate::error::ApiError;
 use crate::provider::Provider;
 use crate::rate_limits::parse_rate_limit_event;
@@ -11,8 +13,6 @@ use crate::sse::responses::process_responses_event;
 use crate::telemetry::WebsocketTelemetry;
 use codex_client::TransportError;
 use codex_utils_rustls_provider::ensure_rustls_crypto_provider;
-use futures::SinkExt;
-use futures::StreamExt;
 use http::HeaderMap;
 use http::HeaderName;
 use http::HeaderValue;
@@ -23,10 +23,8 @@ use serde_json::map::Map as JsonMap;
 use std::sync::Arc;
 use std::sync::OnceLock;
 use std::time::Duration;
-use tokio::net::TcpStream;
 use tokio::sync::Mutex;
 use tokio::sync::mpsc;
-use tokio::sync::oneshot;
 use tokio::time::Instant;
 use tokio_tungstenite::MaybeTlsStream;
 use tokio_tungstenite::WebSocketStream;
@@ -43,110 +41,22 @@ use tungstenite::protocol::WebSocketConfig;
 use url::Url;
 
 struct WsStream {
-    tx_command: mpsc::Sender<WsCommand>,
-    rx_message: mpsc::UnboundedReceiver<Result<Message, WsError>>,
-    pump_task: tokio::task::JoinHandle<()>,
-}
-
-enum WsCommand {
-    Send {
-        message: Message,
-        tx_result: oneshot::Sender<Result<(), WsError>>,
-    },
-    Close {
-        tx_result: oneshot::Sender<Result<(), WsError>>,
-    },
+    pump: WebsocketPump,
+    rx_message: mpsc::UnboundedReceiver<WebsocketMessage>,
 }
 
 impl WsStream {
-    fn new(inner: WebSocketStream<MaybeTlsStream<TcpStream>>) -> Self {
-        let (tx_command, mut rx_command) = mpsc::channel::<WsCommand>(32);
-        let (tx_message, rx_message) = mpsc::unbounded_channel::<Result<Message, WsError>>();
-
-        let pump_task = tokio::spawn(async move {
-            let mut inner = inner;
-            loop {
-                tokio::select! {
-                    command = rx_command.recv() => {
-                        let Some(command) = command else {
-                            break;
-                        };
-                        match command {
-                            WsCommand::Send { message, tx_result } => {
-                                let result = inner.send(message).await;
-                                let should_break = result.is_err();
-                                let _ = tx_result.send(result);
-                                if should_break {
-                                    break;
-                                }
-                            }
-                            WsCommand::Close { tx_result } => {
-                                let result = inner.close(None).await;
-                                let _ = tx_result.send(result);
-                                break;
-                            }
-                        }
-                    }
-                    message = inner.next() => {
-                        let Some(message) = message else {
-                            break;
-                        };
-                        match message {
-                            Ok(Message::Ping(payload)) => {
-                                if let Err(err) = inner.send(Message::Pong(payload)).await {
-                                    let _ = tx_message.send(Err(err));
-                                    break;
-                                }
-                            }
-                            Ok(Message::Pong(_)) => {}
-                            Ok(message @ (Message::Text(_)
-                            | Message::Binary(_)
-                            | Message::Close(_)
-                            | Message::Frame(_))) => {
-                                let is_close = matches!(message, Message::Close(_));
-                                if tx_message.send(Ok(message)).is_err() {
-                                    break;
-                                }
-                                if is_close {
-                                    break;
-                                }
-                            }
-                            Err(err) => {
-                                let _ = tx_message.send(Err(err));
-                                break;
-                            }
-                        }
-                    }
-                }
-            }
-        });
-
-        Self {
-            tx_command,
-            rx_message,
-            pump_task,
-        }
-    }
-
-    async fn request(
-        &self,
-        make_command: impl FnOnce(oneshot::Sender<Result<(), WsError>>) -> WsCommand,
-    ) -> Result<(), WsError> {
-        let (tx_result, rx_result) = oneshot::channel();
-        if self.tx_command.send(make_command(tx_result)).await.is_err() {
-            return Err(WsError::ConnectionClosed);
-        }
-        rx_result.await.unwrap_or(Err(WsError::ConnectionClosed))
+    fn new(inner: WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>>) -> Self {
+        let (pump, rx_message) = WebsocketPump::new(inner);
+        Self { pump, rx_message }
     }
 
     async fn send(&self, message: Message) -> Result<(), WsError> {
-        self.request(|tx_result| WsCommand::Send { message, tx_result })
-            .await
+        self.pump.send(message).await
     }
 
     async fn close(&self) -> Result<(), WsError> {
-        self.request(|tx_result| WsCommand::Close { tx_result })
-            .await
+        self.pump.close().await
     }
 
     async fn next(&mut self) -> Option<Result<Message, WsError>> {
@@ -154,12 +64,6 @@ impl WsStream {
     }
 }
 
-impl Drop for WsStream {
-    fn drop(&mut self) {
-        self.pump_task.abort();
-    }
-}
-
 const X_CODEX_TURN_STATE_HEADER: &str = "x-codex-turn-state";
 const X_MODELS_ETAG_HEADER: &str = "x-models-etag";
 const X_REASONING_INCLUDED_HEADER: &str = "x-reasoning-included";

codex-rs/codex-api/src/endpoint/websocket_pump.rs

@@ -0,0 +1,128 @@
+use futures::SinkExt;
+use futures::StreamExt;
+use tokio::net::TcpStream;
+use tokio::sync::mpsc;
+use tokio::sync::oneshot;
+use tokio_tungstenite::MaybeTlsStream;
+use tokio_tungstenite::WebSocketStream;
+use tokio_tungstenite::tungstenite::Error as WsError;
+use tokio_tungstenite::tungstenite::Message;
+
+pub(crate) type WebsocketMessage = Result<Message, WsError>;
+
+pub(crate) struct WebsocketPump {
+    tx_command: mpsc::Sender<WsCommand>,
+    pump_task: tokio::task::JoinHandle<()>,
+}
+
+enum WsCommand {
+    Send {
+        message: Message,
+        tx_result: oneshot::Sender<Result<(), WsError>>,
+    },
+    Close {
+        tx_result: oneshot::Sender<Result<(), WsError>>,
+    },
+}
+
+impl WebsocketPump {
+    pub(crate) fn new(
+        inner: WebSocketStream<MaybeTlsStream<TcpStream>>,
+    ) -> (Self, mpsc::UnboundedReceiver<WebsocketMessage>) {
+        let (tx_command, mut rx_command) = mpsc::channel::<WsCommand>(32);
+        let (tx_message, rx_message) = mpsc::unbounded_channel::<WebsocketMessage>();
+
+        let pump_task = tokio::spawn(async move {
+            let mut inner = inner;
+            loop {
+                tokio::select! {
+                    command = rx_command.recv() => {
+                        let Some(command) = command else {
+                            break;
+                        };
+                        match command {
+                            WsCommand::Send { message, tx_result } => {
+                                let result = inner.send(message).await;
+                                let should_break = result.is_err();
+                                let _ = tx_result.send(result);
+                                if should_break {
+                                    break;
+                                }
+                            }
+                            WsCommand::Close { tx_result } => {
+                                let result = inner.close(None).await;
+                                let _ = tx_result.send(result);
+                                break;
+                            }
+                        }
+                    }
+                    message = inner.next() => {
+                        let Some(message) = message else {
+                            break;
+                        };
+                        match message {
+                            Ok(Message::Ping(payload)) => {
+                                if let Err(err) = inner.send(Message::Pong(payload)).await {
+                                    let _ = tx_message.send(Err(err));
+                                    break;
+                                }
+                            }
+                            Ok(Message::Pong(_)) => {}
+                            Ok(message @ (Message::Text(_)
+                                | Message::Binary(_)
+                                | Message::Close(_)
+                                | Message::Frame(_))) => {
+                                let is_close = matches!(message, Message::Close(_));
+                                if tx_message.send(Ok(message)).is_err() {
+                                    break;
+                                }
+                                if is_close {
+                                    break;
+                                }
+                            }
+                            Err(err) => {
+                                let _ = tx_message.send(Err(err));
+                                break;
+                            }
+                        }
+                    }
+                }
+            }
+        });
+
+        (
+            Self {
+                tx_command,
+                pump_task,
+            },
+            rx_message,
+        )
+    }
+
+    pub(crate) async fn send(&self, message: Message) -> Result<(), WsError> {
+        self.request(|tx_result| WsCommand::Send { message, tx_result })
+            .await
+    }
+
+    pub(crate) async fn close(&self) -> Result<(), WsError> {
+        self.request(|tx_result| WsCommand::Close { tx_result })
+            .await
+    }
+
+    async fn request(
+        &self,
+        make_command: impl FnOnce(oneshot::Sender<Result<(), WsError>>) -> WsCommand,
+    ) -> Result<(), WsError> {
+        let (tx_result, rx_result) = oneshot::channel();
+        if self.tx_command.send(make_command(tx_result)).await.is_err() {
+            return Err(WsError::ConnectionClosed);
+        }
+        rx_result.await.unwrap_or(Err(WsError::ConnectionClosed))
+    }
+}
+
+impl Drop for WebsocketPump {
+    fn drop(&mut self) {
+        self.pump_task.abort();
+    }
+}

codex-rs/codex-api/src/lib.rs

@@ -29,6 +29,11 @@ pub use crate::endpoint::aggregate::AggregateStreamExt;
 pub use crate::endpoint::compact::CompactClient;
 pub use crate::endpoint::memories::MemoriesClient;
 pub use crate::endpoint::models::ModelsClient;
+pub use crate::endpoint::realtime_websocket::RealtimeAudioFrame;
+pub use crate::endpoint::realtime_websocket::RealtimeEvent;
+pub use crate::endpoint::realtime_websocket::RealtimeSessionConfig;
+pub use crate::endpoint::realtime_websocket::RealtimeWebsocketClient;
+pub use crate::endpoint::realtime_websocket::RealtimeWebsocketConnection;
 pub use crate::endpoint::responses::ResponsesClient;
 pub use crate::endpoint::responses::ResponsesOptions;
 pub use crate::endpoint::responses_websocket::ResponsesWebsocketClient;

codex-rs/codex-api/tests/common/mod.rs

@@ -0,0 +1 @@
+pub mod ws_harness;

codex-rs/codex-api/tests/common/ws_harness.rs

@@ -0,0 +1,43 @@
+use std::collections::HashMap;
+use std::future::Future;
+use std::time::Duration;
+
+use codex_api::Provider;
+use codex_api::provider::RetryConfig;
+use http::HeaderMap;
+use tokio::net::TcpListener;
+use tokio::task::JoinHandle;
+use tokio_tungstenite::WebSocketStream;
+use tokio_tungstenite::accept_async;
+
+pub(crate) async fn spawn_ws_server<F, Fut>(handler: F) -> (String, JoinHandle<()>)
+where
+    F: FnOnce(WebSocketStream<tokio::net::TcpStream>) -> Fut + Send + 'static,
+    Fut: Future<Output = ()> + Send + 'static,
+{
+    let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
+    let addr = listener.local_addr().expect("local addr");
+    let server = tokio::spawn(async move {
+        let (stream, _) = listener.accept().await.expect("accept");
+        let ws = accept_async(stream).await.expect("accept ws");
+        handler(ws).await;
+    });
+    (format!("ws://{addr}"), server)
+}
+
+pub(crate) fn test_provider() -> Provider {
+    Provider {
+        name: "test".to_string(),
+        base_url: "http://localhost".to_string(),
+        query_params: Some(HashMap::new()),
+        headers: HeaderMap::new(),
+        retry: RetryConfig {
+            max_attempts: 1,
+            base_delay: Duration::from_millis(1),
+            retry_429: false,
+            retry_5xx: false,
+            retry_transport: false,
+        },
+        stream_idle_timeout: Duration::from_secs(5),
+    }
+}

codex-rs/codex-api/tests/realtime_websocket_e2e.rs

@@ -0,0 +1,315 @@
+mod common;
+
+use std::time::Duration;
+
+use codex_api::RealtimeAudioFrame;
+use codex_api::RealtimeEvent;
+use codex_api::RealtimeSessionConfig;
+use codex_api::RealtimeWebsocketClient;
+use common::ws_harness;
+use futures::SinkExt;
+use futures::StreamExt;
+use http::HeaderMap;
+use serde_json::Value;
+use serde_json::json;
+use tokio_tungstenite::tungstenite::Message;
+
+#[tokio::test]
+async fn realtime_ws_e2e_session_create_and_event_flow() {
+    let (api_url, server) = ws_harness::spawn_ws_server(|mut ws| async move {
+        let first = ws
+            .next()
+            .await
+            .expect("first msg")
+            .expect("first msg ok")
+            .into_text()
+            .expect("text");
+        let first_json: Value = serde_json::from_str(&first).expect("json");
+        assert_eq!(first_json["type"], "session.create");
+        assert_eq!(
+            first_json["session"]["backend_prompt"],
+            Value::String("backend prompt".to_string())
+        );
+        assert_eq!(
+            first_json["session"]["conversation_id"],
+            Value::String("conv_123".to_string())
+        );
+
+        ws.send(Message::Text(
+            json!({
+                "type": "session.created",
+                "session": {"id": "sess_mock"}
+            })
+            .to_string()
+            .into(),
+        ))
+        .await
+        .expect("send session.created");
+
+        let second = ws
+            .next()
+            .await
+            .expect("second msg")
+            .expect("second msg ok")
+            .into_text()
+            .expect("text");
+        let second_json: Value = serde_json::from_str(&second).expect("json");
+        assert_eq!(second_json["type"], "response.input_audio.delta");
+
+        ws.send(Message::Text(
+            json!({
+                "type": "response.output_audio.delta",
+                "delta": "AQID",
+                "sample_rate": 48000,
+                "num_channels": 1
+            })
+            .to_string()
+            .into(),
+        ))
+        .await
+        .expect("send audio out");
+    })
+    .await;
+
+    let client = RealtimeWebsocketClient::new(ws_harness::test_provider());
+    let connection = client
+        .connect(
+            RealtimeSessionConfig {
+                api_url,
+                prompt: "backend prompt".to_string(),
+                conversation_id: Some("conv_123".to_string()),
+            },
+            HeaderMap::new(),
+            HeaderMap::new(),
+        )
+        .await
+        .expect("connect");
+
+    let created = connection
+        .next_event()
+        .await
+        .expect("next event")
+        .expect("event");
+    assert_eq!(
+        created,
+        RealtimeEvent::SessionCreated {
+            session_id: "sess_mock".to_string()
+        }
+    );
+
+    connection
+        .send_audio_frame(RealtimeAudioFrame {
+            data: "AQID".to_string(),
+            sample_rate: 48000,
+            num_channels: 1,
+            samples_per_channel: Some(960),
+        })
+        .await
+        .expect("send audio");
+
+    let audio_event = connection
+        .next_event()
+        .await
+        .expect("next event")
+        .expect("event");
+    assert_eq!(
+        audio_event,
+        RealtimeEvent::AudioOut(RealtimeAudioFrame {
+            data: "AQID".to_string(),
+            sample_rate: 48000,
+            num_channels: 1,
+            samples_per_channel: None,
+        })
+    );
+
+    connection.close().await.expect("close");
+    server.await.expect("server task");
+}
+
+#[tokio::test]
+async fn realtime_ws_e2e_send_while_next_event_waits() {
+    let (api_url, server) = ws_harness::spawn_ws_server(|mut ws| async move {
+        let first = ws
+            .next()
+            .await
+            .expect("first msg")
+            .expect("first msg ok")
+            .into_text()
+            .expect("text");
+        let first_json: Value = serde_json::from_str(&first).expect("json");
+        assert_eq!(first_json["type"], "session.create");
+
+        let second = ws
+            .next()
+            .await
+            .expect("second msg")
+            .expect("second msg ok")
+            .into_text()
+            .expect("text");
+        let second_json: Value = serde_json::from_str(&second).expect("json");
+        assert_eq!(second_json["type"], "response.input_audio.delta");
+
+        ws.send(Message::Text(
+            json!({
+                "type": "session.created",
+                "session": {"id": "sess_after_send"}
+            })
+            .to_string()
+            .into(),
+        ))
+        .await
+        .expect("send session.created");
+    })
+    .await;
+
+    let client = RealtimeWebsocketClient::new(ws_harness::test_provider());
+    let connection = client
+        .connect(
+            RealtimeSessionConfig {
+                api_url,
+                prompt: "backend prompt".to_string(),
+                conversation_id: Some("conv_123".to_string()),
+            },
+            HeaderMap::new(),
+            HeaderMap::new(),
+        )
+        .await
+        .expect("connect");
+
+    let (send_result, next_result) = tokio::join!(
+        async {
+            tokio::time::timeout(
+                Duration::from_millis(200),
+                connection.send_audio_frame(RealtimeAudioFrame {
+                    data: "AQID".to_string(),
+                    sample_rate: 48000,
+                    num_channels: 1,
+                    samples_per_channel: Some(960),
+                }),
+            )
+            .await
+        },
+        connection.next_event()
+    );
+
+    send_result
+        .expect("send should not block on next_event")
+        .expect("send audio");
+    let next_event = next_result.expect("next event").expect("event");
+    assert_eq!(
+        next_event,
+        RealtimeEvent::SessionCreated {
+            session_id: "sess_after_send".to_string()
+        }
+    );
+
+    connection.close().await.expect("close");
+    server.await.expect("server task");
+}
+
+#[tokio::test]
+async fn realtime_ws_e2e_disconnected_emitted_once() {
+    let (api_url, server) = ws_harness::spawn_ws_server(|mut ws| async move {
+        let first = ws
+            .next()
+            .await
+            .expect("first msg")
+            .expect("first msg ok")
+            .into_text()
+            .expect("text");
+        let first_json: Value = serde_json::from_str(&first).expect("json");
+        assert_eq!(first_json["type"], "session.create");
+
+        ws.send(Message::Close(None)).await.expect("send close");
+    })
+    .await;
+
+    let client = RealtimeWebsocketClient::new(ws_harness::test_provider());
+    let connection = client
+        .connect(
+            RealtimeSessionConfig {
+                api_url,
+                prompt: "backend prompt".to_string(),
+                conversation_id: Some("conv_123".to_string()),
+            },
+            HeaderMap::new(),
+            HeaderMap::new(),
+        )
+        .await
+        .expect("connect");
+
+    let first = connection.next_event().await.expect("next event");
+    assert_eq!(first, None);
+
+    let second = connection.next_event().await.expect("next event");
+    assert_eq!(second, None);
+
+    server.await.expect("server task");
+}
+
+#[tokio::test]
+async fn realtime_ws_e2e_ignores_unknown_text_events() {
+    let (api_url, server) = ws_harness::spawn_ws_server(|mut ws| async move {
+        let first = ws
+            .next()
+            .await
+            .expect("first msg")
+            .expect("first msg ok")
+            .into_text()
+            .expect("text");
+        let first_json: Value = serde_json::from_str(&first).expect("json");
+        assert_eq!(first_json["type"], "session.create");
+
+        ws.send(Message::Text(
+            json!({
+                "type": "response.created",
+                "response": {"id": "resp_unknown"}
+            })
+            .to_string()
+            .into(),
+        ))
+        .await
+        .expect("send unknown event");
+
+        ws.send(Message::Text(
+            json!({
+                "type": "session.created",
+                "session": {"id": "sess_after_unknown"}
+            })
+            .to_string()
+            .into(),
+        ))
+        .await
+        .expect("send session.created");
+    })
+    .await;
+
+    let client = RealtimeWebsocketClient::new(ws_harness::test_provider());
+    let connection = client
+        .connect(
+            RealtimeSessionConfig {
+                api_url,
+                prompt: "backend prompt".to_string(),
+                conversation_id: Some("conv_123".to_string()),
+            },
+            HeaderMap::new(),
+            HeaderMap::new(),
+        )
+        .await
+        .expect("connect");
+
+    let event = connection
+        .next_event()
+        .await
+        .expect("next event")
+        .expect("event");
+    assert_eq!(
+        event,
+        RealtimeEvent::SessionCreated {
+            session_id: "sess_after_unknown".to_string()
+        }
+    );
+
+    connection.close().await.expect("close");
+    server.await.expect("server task");
+}