From 2253a9d1d7832cacb86cebf48c267eb58d039603 Mon Sep 17 00:00:00 2001 From: Ahmed Ibrahim Date: Thu, 12 Mar 2026 23:50:30 -0700 Subject: [PATCH] Add realtime transcription mode for websocket sessions (#14556) - add experimental_realtime_ws_mode (conversational/transcription) and plumb it into realtime conversation session config - switch realtime websocket intent and session.update payload shape based on mode - update config schema and realtime/config tests --------- Co-authored-by: Codex --- .../endpoint/realtime_websocket/methods.rs | 429 ++++++++++++++++-- .../src/endpoint/realtime_websocket/mod.rs | 1 + .../endpoint/realtime_websocket/protocol.rs | 13 +- codex-rs/codex-api/src/lib.rs | 1 + .../codex-api/tests/realtime_websocket_e2e.rs | 6 + codex-rs/core/config.schema.json | 15 + codex-rs/core/src/config/config_tests.rs | 32 ++ codex-rs/core/src/config/mod.rs | 15 + codex-rs/core/src/realtime_conversation.rs | 33 +- 9 files changed, 482 insertions(+), 63 deletions(-) diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs index c60564d1b..cb710b30b 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs @@ -7,6 +7,7 @@ use crate::endpoint::realtime_websocket::protocol::RealtimeEvent; use crate::endpoint::realtime_websocket::protocol::RealtimeEventParser; use crate::endpoint::realtime_websocket::protocol::RealtimeOutboundMessage; use crate::endpoint::realtime_websocket::protocol::RealtimeSessionConfig; +use crate::endpoint::realtime_websocket::protocol::RealtimeSessionMode; use crate::endpoint::realtime_websocket::protocol::RealtimeTranscriptDelta; use crate::endpoint::realtime_websocket::protocol::RealtimeTranscriptEntry; use crate::endpoint::realtime_websocket::protocol::SessionAudio; @@ -52,6 +53,16 @@ const REALTIME_V2_SESSION_TYPE: &str = "realtime"; const REALTIME_V2_CODEX_TOOL_NAME: &str = "codex"; const REALTIME_V2_CODEX_TOOL_DESCRIPTION: &str = "Delegate work to Codex and return the result."; +fn normalized_session_mode( + event_parser: RealtimeEventParser, + session_mode: RealtimeSessionMode, +) -> RealtimeSessionMode { + match event_parser { + RealtimeEventParser::V1 => RealtimeSessionMode::Conversational, + RealtimeEventParser::RealtimeV2 => session_mode, + } +} + struct WsStream { tx_command: mpsc::Sender, pump_task: tokio::task::JoinHandle<()>, @@ -289,12 +300,16 @@ impl RealtimeWebsocketWriter { } pub async fn send_conversation_item_create(&self, text: String) -> Result<(), ApiError> { + let content_kind = match self.event_parser { + RealtimeEventParser::V1 => "text", + RealtimeEventParser::RealtimeV2 => "input_text", + }; self.send_json(RealtimeOutboundMessage::ConversationItemCreate { item: ConversationItemPayload::Message(ConversationMessageItem { kind: "message".to_string(), role: "user".to_string(), content: vec![ConversationItemContent { - kind: "text".to_string(), + kind: content_kind.to_string(), text, }], }), @@ -326,34 +341,51 @@ impl RealtimeWebsocketWriter { self.send_json(message).await } - pub async fn send_session_update(&self, instructions: String) -> Result<(), ApiError> { - let (session_kind, tools) = match self.event_parser { - RealtimeEventParser::V1 => (REALTIME_V1_SESSION_TYPE.to_string(), None), - RealtimeEventParser::RealtimeV2 => ( - REALTIME_V2_SESSION_TYPE.to_string(), - Some(vec![SessionFunctionTool { - kind: "function".to_string(), - name: REALTIME_V2_CODEX_TOOL_NAME.to_string(), - description: REALTIME_V2_CODEX_TOOL_DESCRIPTION.to_string(), - parameters: json!({ - "type": "object", - "properties": { - "prompt": { - "type": "string", - "description": "Prompt text for the delegated Codex task." - } - }, - "required": ["prompt"], - "additionalProperties": false + pub async fn send_session_update( + &self, + instructions: String, + session_mode: RealtimeSessionMode, + ) -> Result<(), ApiError> { + let session_mode = normalized_session_mode(self.event_parser, session_mode); + let (session_kind, session_instructions, output_audio) = match session_mode { + RealtimeSessionMode::Conversational => { + let kind = match self.event_parser { + RealtimeEventParser::V1 => REALTIME_V1_SESSION_TYPE.to_string(), + RealtimeEventParser::RealtimeV2 => REALTIME_V2_SESSION_TYPE.to_string(), + }; + ( + kind, + Some(instructions), + Some(SessionAudioOutput { + voice: REALTIME_AUDIO_VOICE.to_string(), }), - }]), - ), + ) + } + RealtimeSessionMode::Transcription => ("transcription".to_string(), None, None), + }; + let tools = match self.event_parser { + RealtimeEventParser::RealtimeV2 => Some(vec![SessionFunctionTool { + kind: "function".to_string(), + name: REALTIME_V2_CODEX_TOOL_NAME.to_string(), + description: REALTIME_V2_CODEX_TOOL_DESCRIPTION.to_string(), + parameters: json!({ + "type": "object", + "properties": { + "prompt": { + "type": "string", + "description": "Prompt text for the delegated Codex task." + } + }, + "required": ["prompt"], + "additionalProperties": false + }), + }]), + RealtimeEventParser::V1 => None, }; - self.send_json(RealtimeOutboundMessage::SessionUpdate { session: SessionUpdateSession { kind: session_kind, - instructions, + instructions: session_instructions, audio: SessionAudio { input: SessionAudioInput { format: SessionAudioFormat { @@ -361,9 +393,7 @@ impl RealtimeWebsocketWriter { rate: REALTIME_AUDIO_SAMPLE_RATE, }, }, - output: SessionAudioOutput { - voice: REALTIME_AUDIO_VOICE.to_string(), - }, + output: output_audio, }, tools, }, @@ -514,6 +544,8 @@ impl RealtimeWebsocketClient { self.provider.base_url.as_str(), self.provider.query_params.as_ref(), config.model.as_deref(), + config.event_parser, + config.session_mode, )?; let mut request = ws_url @@ -555,7 +587,7 @@ impl RealtimeWebsocketClient { ); connection .writer - .send_session_update(config.instructions) + .send_session_update(config.instructions, config.session_mode) .await?; Ok(connection) } @@ -600,6 +632,8 @@ fn websocket_url_from_api_url( api_url: &str, query_params: Option<&HashMap>, model: Option<&str>, + event_parser: RealtimeEventParser, + _session_mode: RealtimeSessionMode, ) -> Result { let mut url = Url::parse(api_url) .map_err(|err| ApiError::Stream(format!("failed to parse realtime api_url: {err}")))?; @@ -619,9 +653,20 @@ fn websocket_url_from_api_url( } } - { + let intent = match event_parser { + RealtimeEventParser::V1 => Some("quicksilver"), + RealtimeEventParser::RealtimeV2 => None, + }; + let has_extra_query_params = query_params.is_some_and(|query_params| { + query_params + .iter() + .any(|(key, _)| key != "intent" && !(key == "model" && model.is_some())) + }); + if intent.is_some() || model.is_some() || has_extra_query_params { let mut query = url.query_pairs_mut(); - query.append_pair("intent", "quicksilver"); + if let Some(intent) = intent { + query.append_pair("intent", intent); + } if let Some(model) = model { query.append_pair("model", model); } @@ -902,8 +947,14 @@ mod tests { #[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, None).expect("build ws url"); + let url = websocket_url_from_api_url( + "http://127.0.0.1:8011", + None, + None, + RealtimeEventParser::V1, + RealtimeSessionMode::Conversational, + ) + .expect("build ws url"); assert_eq!( url.as_str(), "ws://127.0.0.1:8011/v1/realtime?intent=quicksilver" @@ -912,9 +963,14 @@ mod tests { #[test] fn websocket_url_from_ws_base_defaults_to_ws_path() { - let url = - websocket_url_from_api_url("wss://example.com", None, Some("realtime-test-model")) - .expect("build ws url"); + let url = websocket_url_from_api_url( + "wss://example.com", + None, + Some("realtime-test-model"), + RealtimeEventParser::V1, + RealtimeSessionMode::Conversational, + ) + .expect("build ws url"); assert_eq!( url.as_str(), "wss://example.com/v1/realtime?intent=quicksilver&model=realtime-test-model" @@ -923,8 +979,14 @@ mod tests { #[test] fn websocket_url_from_v1_base_appends_realtime_path() { - let url = websocket_url_from_api_url("https://api.openai.com/v1", None, Some("snapshot")) - .expect("build ws url"); + let url = websocket_url_from_api_url( + "https://api.openai.com/v1", + None, + Some("snapshot"), + RealtimeEventParser::V1, + RealtimeSessionMode::Conversational, + ) + .expect("build ws url"); assert_eq!( url.as_str(), "wss://api.openai.com/v1/realtime?intent=quicksilver&model=snapshot" @@ -933,9 +995,14 @@ mod tests { #[test] fn websocket_url_from_nested_v1_base_appends_realtime_path() { - let url = - websocket_url_from_api_url("https://example.com/openai/v1", None, Some("snapshot")) - .expect("build ws url"); + let url = websocket_url_from_api_url( + "https://example.com/openai/v1", + None, + Some("snapshot"), + RealtimeEventParser::V1, + RealtimeSessionMode::Conversational, + ) + .expect("build ws url"); assert_eq!( url.as_str(), "wss://example.com/openai/v1/realtime?intent=quicksilver&model=snapshot" @@ -951,6 +1018,8 @@ mod tests { ("intent".to_string(), "ignored".to_string()), ])), Some("snapshot"), + RealtimeEventParser::V1, + RealtimeSessionMode::Conversational, ) .expect("build ws url"); assert_eq!( @@ -959,6 +1028,54 @@ mod tests { ); } + #[test] + fn websocket_url_v1_ignores_transcription_mode() { + let url = websocket_url_from_api_url( + "https://example.com", + None, + None, + RealtimeEventParser::V1, + RealtimeSessionMode::Transcription, + ) + .expect("build ws url"); + assert_eq!( + url.as_str(), + "wss://example.com/v1/realtime?intent=quicksilver" + ); + } + + #[test] + fn websocket_url_omits_intent_for_realtime_v2_conversational_mode() { + let url = websocket_url_from_api_url( + "https://example.com/v1/realtime?foo=bar", + Some(&HashMap::from([ + ("trace".to_string(), "1".to_string()), + ("intent".to_string(), "ignored".to_string()), + ])), + Some("snapshot"), + RealtimeEventParser::RealtimeV2, + RealtimeSessionMode::Conversational, + ) + .expect("build ws url"); + assert_eq!( + url.as_str(), + "wss://example.com/v1/realtime?foo=bar&model=snapshot&trace=1" + ); + } + + #[test] + fn websocket_url_omits_intent_for_realtime_v2_transcription_mode() { + let url = websocket_url_from_api_url( + "https://example.com", + None, + None, + RealtimeEventParser::RealtimeV2, + RealtimeSessionMode::Transcription, + ) + .expect("build ws url"); + assert_eq!(url.as_str(), "wss://example.com/v1/realtime"); + } + #[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"); @@ -1124,6 +1241,7 @@ mod tests { model: Some("realtime-test-model".to_string()), session_id: Some("conv_1".to_string()), event_parser: RealtimeEventParser::V1, + session_mode: RealtimeSessionMode::Conversational, }, HeaderMap::new(), HeaderMap::new(), @@ -1301,14 +1419,36 @@ mod tests { assert_eq!(second_json["type"], "conversation.item.create"); assert_eq!( second_json["item"]["type"], + Value::String("message".to_string()) + ); + assert_eq!( + second_json["item"]["content"][0]["type"], + Value::String("input_text".to_string()) + ); + assert_eq!( + second_json["item"]["content"][0]["text"], + Value::String("delegate this".to_string()) + ); + + 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"]["type"], Value::String("function_call_output".to_string()) ); assert_eq!( - second_json["item"]["call_id"], + third_json["item"]["call_id"], Value::String("call_1".to_string()) ); assert_eq!( - second_json["item"]["output"], + third_json["item"]["output"], Value::String("delegated result".to_string()) ); }); @@ -1335,6 +1475,7 @@ mod tests { model: Some("realtime-test-model".to_string()), session_id: Some("conv_1".to_string()), event_parser: RealtimeEventParser::RealtimeV2, + session_mode: RealtimeSessionMode::Conversational, }, HeaderMap::new(), HeaderMap::new(), @@ -1355,6 +1496,10 @@ mod tests { } ); + connection + .send_conversation_item_create("delegate this".to_string()) + .await + .expect("send text item"); connection .send_conversation_handoff_append("call_1".to_string(), "delegated result".to_string()) .await @@ -1364,6 +1509,205 @@ mod tests { server.await.expect("server task"); } + #[tokio::test] + async fn transcription_mode_session_update_omits_output_audio_and_instructions() { + 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.update"); + assert_eq!( + first_json["session"]["type"], + Value::String("transcription".to_string()) + ); + assert!(first_json["session"].get("instructions").is_none()); + assert!(first_json["session"]["audio"].get("output").is_none()); + assert_eq!( + first_json["session"]["tools"][0]["name"], + Value::String("codex".to_string()) + ); + + ws.send(Message::Text( + json!({ + "type": "session.updated", + "session": {"id": "sess_transcription"} + }) + .to_string() + .into(), + )) + .await + .expect("send session.updated"); + + 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"], "input_audio_buffer.append"); + }); + + let provider = Provider { + name: "test".to_string(), + base_url: format!("http://{addr}"), + 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 { + instructions: "backend prompt".to_string(), + model: Some("realtime-test-model".to_string()), + session_id: Some("conv_1".to_string()), + event_parser: RealtimeEventParser::RealtimeV2, + session_mode: RealtimeSessionMode::Transcription, + }, + HeaderMap::new(), + HeaderMap::new(), + ) + .await + .expect("connect"); + + let created = connection + .next_event() + .await + .expect("next event") + .expect("event"); + assert_eq!( + created, + RealtimeEvent::SessionUpdated { + session_id: "sess_transcription".to_string(), + instructions: None, + } + ); + + connection + .send_audio_frame(RealtimeAudioFrame { + data: "AQID".to_string(), + sample_rate: 24_000, + num_channels: 1, + samples_per_channel: Some(480), + }) + .await + .expect("send audio"); + + connection.close().await.expect("close"); + server.await.expect("server task"); + } + + #[tokio::test] + async fn v1_transcription_mode_is_treated_as_conversational() { + 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.update"); + assert_eq!( + first_json["session"]["type"], + Value::String("quicksilver".to_string()) + ); + assert_eq!( + first_json["session"]["instructions"], + Value::String("backend prompt".to_string()) + ); + assert_eq!( + first_json["session"]["audio"]["output"]["voice"], + Value::String("fathom".to_string()) + ); + assert!(first_json["session"].get("tools").is_none()); + + ws.send(Message::Text( + json!({ + "type": "session.updated", + "session": {"id": "sess_v1_mode"} + }) + .to_string() + .into(), + )) + .await + .expect("send session.updated"); + }); + + let provider = Provider { + name: "test".to_string(), + base_url: format!("http://{addr}"), + 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 { + instructions: "backend prompt".to_string(), + model: Some("realtime-test-model".to_string()), + session_id: Some("conv_1".to_string()), + event_parser: RealtimeEventParser::V1, + session_mode: RealtimeSessionMode::Transcription, + }, + HeaderMap::new(), + HeaderMap::new(), + ) + .await + .expect("connect"); + + let created = connection + .next_event() + .await + .expect("next event") + .expect("event"); + assert_eq!( + created, + RealtimeEvent::SessionUpdated { + session_id: "sess_v1_mode".to_string(), + instructions: None, + } + ); + + 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"); @@ -1427,6 +1771,7 @@ mod tests { model: Some("realtime-test-model".to_string()), session_id: Some("conv_1".to_string()), event_parser: RealtimeEventParser::V1, + session_mode: RealtimeSessionMode::Conversational, }, HeaderMap::new(), HeaderMap::new(), diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/mod.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/mod.rs index 5672e0175..f307e6091 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/mod.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/mod.rs @@ -12,3 +12,4 @@ pub use methods::RealtimeWebsocketEvents; pub use methods::RealtimeWebsocketWriter; pub use protocol::RealtimeEventParser; pub use protocol::RealtimeSessionConfig; +pub use protocol::RealtimeSessionMode; diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs index 6479b3ec6..028f51cf3 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs @@ -14,12 +14,19 @@ pub enum RealtimeEventParser { RealtimeV2, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RealtimeSessionMode { + Conversational, + Transcription, +} + #[derive(Debug, Clone, PartialEq, Eq)] pub struct RealtimeSessionConfig { pub instructions: String, pub model: Option, pub session_id: Option, pub event_parser: RealtimeEventParser, + pub session_mode: RealtimeSessionMode, } #[derive(Debug, Clone, Serialize)] @@ -42,7 +49,8 @@ pub(super) enum RealtimeOutboundMessage { pub(super) struct SessionUpdateSession { #[serde(rename = "type")] pub(super) kind: String, - pub(super) instructions: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub(super) instructions: Option, pub(super) audio: SessionAudio, #[serde(skip_serializing_if = "Option::is_none")] pub(super) tools: Option>, @@ -51,7 +59,8 @@ pub(super) struct SessionUpdateSession { #[derive(Debug, Clone, Serialize)] pub(super) struct SessionAudio { pub(super) input: SessionAudioInput, - pub(super) output: SessionAudioOutput, + #[serde(skip_serializing_if = "Option::is_none")] + pub(super) output: Option, } #[derive(Debug, Clone, Serialize)] diff --git a/codex-rs/codex-api/src/lib.rs b/codex-rs/codex-api/src/lib.rs index 35ae983b9..a1588a983 100644 --- a/codex-rs/codex-api/src/lib.rs +++ b/codex-rs/codex-api/src/lib.rs @@ -29,6 +29,7 @@ pub use crate::endpoint::memories::MemoriesClient; pub use crate::endpoint::models::ModelsClient; pub use crate::endpoint::realtime_websocket::RealtimeEventParser; pub use crate::endpoint::realtime_websocket::RealtimeSessionConfig; +pub use crate::endpoint::realtime_websocket::RealtimeSessionMode; pub use crate::endpoint::realtime_websocket::RealtimeWebsocketClient; pub use crate::endpoint::realtime_websocket::RealtimeWebsocketConnection; pub use crate::endpoint::responses::ResponsesClient; diff --git a/codex-rs/codex-api/tests/realtime_websocket_e2e.rs b/codex-rs/codex-api/tests/realtime_websocket_e2e.rs index d6d73c0f0..30786ad92 100644 --- a/codex-rs/codex-api/tests/realtime_websocket_e2e.rs +++ b/codex-rs/codex-api/tests/realtime_websocket_e2e.rs @@ -6,6 +6,7 @@ use codex_api::RealtimeAudioFrame; use codex_api::RealtimeEvent; use codex_api::RealtimeEventParser; use codex_api::RealtimeSessionConfig; +use codex_api::RealtimeSessionMode; use codex_api::RealtimeWebsocketClient; use codex_api::provider::Provider; use codex_api::provider::RetryConfig; @@ -142,6 +143,7 @@ async fn realtime_ws_e2e_session_create_and_event_flow() { model: Some("realtime-test-model".to_string()), session_id: Some("conv_123".to_string()), event_parser: RealtimeEventParser::V1, + session_mode: RealtimeSessionMode::Conversational, }, HeaderMap::new(), HeaderMap::new(), @@ -235,6 +237,7 @@ async fn realtime_ws_e2e_send_while_next_event_waits() { model: Some("realtime-test-model".to_string()), session_id: Some("conv_123".to_string()), event_parser: RealtimeEventParser::V1, + session_mode: RealtimeSessionMode::Conversational, }, HeaderMap::new(), HeaderMap::new(), @@ -299,6 +302,7 @@ async fn realtime_ws_e2e_disconnected_emitted_once() { model: Some("realtime-test-model".to_string()), session_id: Some("conv_123".to_string()), event_parser: RealtimeEventParser::V1, + session_mode: RealtimeSessionMode::Conversational, }, HeaderMap::new(), HeaderMap::new(), @@ -360,6 +364,7 @@ async fn realtime_ws_e2e_ignores_unknown_text_events() { model: Some("realtime-test-model".to_string()), session_id: Some("conv_123".to_string()), event_parser: RealtimeEventParser::V1, + session_mode: RealtimeSessionMode::Conversational, }, HeaderMap::new(), HeaderMap::new(), @@ -424,6 +429,7 @@ async fn realtime_ws_e2e_realtime_v2_parser_emits_handoff_requested() { model: Some("realtime-test-model".to_string()), session_id: Some("conv_123".to_string()), event_parser: RealtimeEventParser::RealtimeV2, + session_mode: RealtimeSessionMode::Conversational, }, HeaderMap::new(), HeaderMap::new(), diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index 3b2f07f85..40949dfa6 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -1342,6 +1342,13 @@ }, "type": "object" }, + "RealtimeWsMode": { + "enum": [ + "conversational", + "transcription" + ], + "type": "string" + }, "ReasoningEffort": { "description": "See https://platform.openai.com/docs/guides/reasoning?api-mode=responses#get-started-with-reasoning", "enum": [ @@ -1816,6 +1823,14 @@ "description": "Experimental / do not use. Overrides only the realtime conversation websocket transport base URL (the `Op::RealtimeConversation` `/v1/realtime` connection) without changing normal provider HTTP requests.", "type": "string" }, + "experimental_realtime_ws_mode": { + "allOf": [ + { + "$ref": "#/definitions/RealtimeWsMode" + } + ], + "description": "Experimental / do not use. Selects the realtime websocket intent mode. `conversational` is speech-to-speech while `transcription` is transcript-only." + }, "experimental_realtime_ws_model": { "description": "Experimental / do not use. Selects the realtime websocket model/snapshot used for the `Op::RealtimeConversation` connection.", "type": "string" diff --git a/codex-rs/core/src/config/config_tests.rs b/codex-rs/core/src/config/config_tests.rs index 00e5a1756..0fe4b5815 100644 --- a/codex-rs/core/src/config/config_tests.rs +++ b/codex-rs/core/src/config/config_tests.rs @@ -4129,6 +4129,7 @@ fn test_precedence_fixture_with_o3_profile() -> std::io::Result<()> { experimental_realtime_start_instructions: None, experimental_realtime_ws_base_url: None, experimental_realtime_ws_model: None, + experimental_realtime_ws_mode: RealtimeWsMode::Conversational, experimental_realtime_ws_backend_prompt: None, experimental_realtime_ws_startup_context: None, base_instructions: None, @@ -4265,6 +4266,7 @@ fn test_precedence_fixture_with_gpt3_profile() -> std::io::Result<()> { experimental_realtime_start_instructions: None, experimental_realtime_ws_base_url: None, experimental_realtime_ws_model: None, + experimental_realtime_ws_mode: RealtimeWsMode::Conversational, experimental_realtime_ws_backend_prompt: None, experimental_realtime_ws_startup_context: None, base_instructions: None, @@ -4399,6 +4401,7 @@ fn test_precedence_fixture_with_zdr_profile() -> std::io::Result<()> { experimental_realtime_start_instructions: None, experimental_realtime_ws_base_url: None, experimental_realtime_ws_model: None, + experimental_realtime_ws_mode: RealtimeWsMode::Conversational, experimental_realtime_ws_backend_prompt: None, experimental_realtime_ws_startup_context: None, base_instructions: None, @@ -4519,6 +4522,7 @@ fn test_precedence_fixture_with_gpt5_profile() -> std::io::Result<()> { experimental_realtime_start_instructions: None, experimental_realtime_ws_base_url: None, experimental_realtime_ws_model: None, + experimental_realtime_ws_mode: RealtimeWsMode::Conversational, experimental_realtime_ws_backend_prompt: None, experimental_realtime_ws_startup_context: None, base_instructions: None, @@ -5566,6 +5570,34 @@ experimental_realtime_ws_model = "realtime-test-model" Ok(()) } +#[test] +fn experimental_realtime_ws_mode_loads_from_config_toml() -> std::io::Result<()> { + let cfg: ConfigToml = toml::from_str( + r#" +experimental_realtime_ws_mode = "transcription" +"#, + ) + .expect("TOML deserialization should succeed"); + + assert_eq!( + cfg.experimental_realtime_ws_mode, + Some(RealtimeWsMode::Transcription) + ); + + let codex_home = TempDir::new()?; + let config = Config::load_from_base_config_with_overrides( + cfg, + ConfigOverrides::default(), + codex_home.path().to_path_buf(), + )?; + + assert_eq!( + config.experimental_realtime_ws_mode, + RealtimeWsMode::Transcription + ); + Ok(()) +} + #[test] fn realtime_audio_loads_from_config_toml() -> std::io::Result<()> { let cfg: ConfigToml = toml::from_str( diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index e4e90fca4..d239c91bb 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/codex-rs/core/src/config/mod.rs @@ -463,6 +463,9 @@ pub struct Config { /// Experimental / do not use. Selects the realtime websocket model/snapshot /// used for the `Op::RealtimeConversation` connection. pub experimental_realtime_ws_model: Option, + /// Experimental / do not use. Selects the realtime websocket intent mode. + /// `conversational` is speech-to-speech while `transcription` is transcript-only. + pub experimental_realtime_ws_mode: RealtimeWsMode, /// Experimental / do not use. Overrides only the realtime conversation /// websocket transport instructions (the `Op::RealtimeConversation` /// `/ws` session.update instructions) without changing normal prompts. @@ -1238,6 +1241,9 @@ pub struct ConfigToml { /// Experimental / do not use. Selects the realtime websocket model/snapshot /// used for the `Op::RealtimeConversation` connection. pub experimental_realtime_ws_model: Option, + /// Experimental / do not use. Selects the realtime websocket intent mode. + /// `conversational` is speech-to-speech while `transcription` is transcript-only. + pub experimental_realtime_ws_mode: Option, /// Experimental / do not use. Overrides only the realtime conversation /// websocket transport instructions (the `Op::RealtimeConversation` /// `/ws` session.update instructions) without changing normal prompts. @@ -1383,6 +1389,14 @@ pub struct RealtimeAudioConfig { pub speaker: Option, } +#[derive(Serialize, Deserialize, Debug, Clone, Copy, Default, PartialEq, Eq, JsonSchema)] +#[serde(rename_all = "snake_case")] +pub enum RealtimeWsMode { + #[default] + Conversational, + Transcription, +} + #[derive(Serialize, Deserialize, Debug, Clone, Default, PartialEq, Eq, JsonSchema)] #[schemars(deny_unknown_fields)] pub struct RealtimeAudioToml { @@ -2462,6 +2476,7 @@ impl Config { }), experimental_realtime_ws_base_url: cfg.experimental_realtime_ws_base_url, experimental_realtime_ws_model: cfg.experimental_realtime_ws_model, + experimental_realtime_ws_mode: cfg.experimental_realtime_ws_mode.unwrap_or_default(), experimental_realtime_ws_backend_prompt: cfg.experimental_realtime_ws_backend_prompt, experimental_realtime_ws_startup_context: cfg.experimental_realtime_ws_startup_context, experimental_realtime_start_instructions: cfg.experimental_realtime_start_instructions, diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index bc822a9bf..03407a427 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -15,6 +15,7 @@ use codex_api::RealtimeAudioFrame; use codex_api::RealtimeEvent; use codex_api::RealtimeEventParser; use codex_api::RealtimeSessionConfig; +use codex_api::RealtimeSessionMode; use codex_api::RealtimeWebsocketClient; use codex_api::endpoint::realtime_websocket::RealtimeWebsocketEvents; use codex_api::endpoint::realtime_websocket::RealtimeWebsocketWriter; @@ -116,10 +117,7 @@ impl RealtimeConversationManager { &self, api_provider: ApiProvider, extra_headers: Option, - prompt: String, - model: Option, - session_id: Option, - event_parser: RealtimeEventParser, + session_config: RealtimeSessionConfig, ) -> CodexResult<(Receiver, Arc)> { let previous_state = { let mut guard = self.state.lock().await; @@ -131,12 +129,6 @@ impl RealtimeConversationManager { let _ = state.task.await; } - let session_config = RealtimeSessionConfig { - instructions: prompt, - model, - session_id, - event_parser, - }; let client = RealtimeWebsocketClient::new(api_provider); let connection = client .connect( @@ -307,23 +299,26 @@ pub(crate) async fn handle_start( } else { RealtimeEventParser::V1 }; - + let session_mode = match config.experimental_realtime_ws_mode { + crate::config::RealtimeWsMode::Conversational => RealtimeSessionMode::Conversational, + crate::config::RealtimeWsMode::Transcription => RealtimeSessionMode::Transcription, + }; let requested_session_id = params .session_id .or_else(|| Some(sess.conversation_id.to_string())); + let session_config = RealtimeSessionConfig { + instructions: prompt, + model, + session_id: requested_session_id.clone(), + event_parser, + session_mode, + }; let extra_headers = realtime_request_headers(requested_session_id.as_deref(), realtime_api_key.as_str())?; info!("starting realtime conversation"); let (events_rx, realtime_active) = match sess .conversation - .start( - api_provider, - extra_headers, - prompt, - model, - requested_session_id.clone(), - event_parser, - ) + .start(api_provider, extra_headers, session_config) .await { Ok(events_rx) => events_rx,