From bc01d653d8bed421f5cf40d1525c66aa5c5b7ef5 Mon Sep 17 00:00:00 2001 From: Ahmed Ibrahim Date: Sun, 15 Mar 2026 15:53:47 -0700 Subject: [PATCH] Split realtime websocket methods by version Co-authored-by: Codex --- .../endpoint/realtime_websocket/methods.rs | 206 ++---------------- .../realtime_websocket/methods_common.rs | 67 ++++++ .../endpoint/realtime_websocket/methods_v1.rs | 65 ++++++ .../endpoint/realtime_websocket/methods_v2.rs | 128 +++++++++++ .../src/endpoint/realtime_websocket/mod.rs | 3 + 5 files changed, 283 insertions(+), 186 deletions(-) create mode 100644 codex-rs/codex-api/src/endpoint/realtime_websocket/methods_common.rs create mode 100644 codex-rs/codex-api/src/endpoint/realtime_websocket/methods_v1.rs create mode 100644 codex-rs/codex-api/src/endpoint/realtime_websocket/methods_v2.rs 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 dbd9a5296f..22fb3276a1 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs @@ -1,7 +1,8 @@ -use crate::endpoint::realtime_websocket::protocol::ConversationFunctionCallOutputItem; -use crate::endpoint::realtime_websocket::protocol::ConversationItemContent; -use crate::endpoint::realtime_websocket::protocol::ConversationItemPayload; -use crate::endpoint::realtime_websocket::protocol::ConversationMessageItem; +use crate::endpoint::realtime_websocket::methods_common::conversation_handoff_append_message; +use crate::endpoint::realtime_websocket::methods_common::conversation_item_create_message; +use crate::endpoint::realtime_websocket::methods_common::normalized_session_mode; +use crate::endpoint::realtime_websocket::methods_common::session_update_session; +use crate::endpoint::realtime_websocket::methods_common::websocket_intent; use crate::endpoint::realtime_websocket::protocol::RealtimeAudioFrame; use crate::endpoint::realtime_websocket::protocol::RealtimeEvent; use crate::endpoint::realtime_websocket::protocol::RealtimeEventParser; @@ -10,16 +11,6 @@ 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; -use crate::endpoint::realtime_websocket::protocol::SessionAudioFormat; -use crate::endpoint::realtime_websocket::protocol::SessionAudioInput; -use crate::endpoint::realtime_websocket::protocol::SessionAudioOutput; -use crate::endpoint::realtime_websocket::protocol::SessionAudioOutputFormat; -use crate::endpoint::realtime_websocket::protocol::SessionAudioVoice; -use crate::endpoint::realtime_websocket::protocol::SessionFunctionTool; -use crate::endpoint::realtime_websocket::protocol::SessionNoiseReduction; -use crate::endpoint::realtime_websocket::protocol::SessionTurnDetection; -use crate::endpoint::realtime_websocket::protocol::SessionUpdateSession; use crate::endpoint::realtime_websocket::protocol::parse_realtime_event; use crate::error::ApiError; use crate::provider::Provider; @@ -29,7 +20,6 @@ use futures::SinkExt; use futures::StreamExt; use http::HeaderMap; use http::HeaderValue; -use serde_json::json; use std::collections::HashMap; use std::sync::Arc; use std::sync::atomic::AtomicBool; @@ -50,27 +40,6 @@ use tracing::trace; use tungstenite::protocol::WebSocketConfig; use url::Url; -const REALTIME_AUDIO_SAMPLE_RATE: u32 = 24_000; -const REALTIME_AUDIO_FORMAT: &str = "audio/pcm"; -const REALTIME_V2_NOISE_REDUCTION: &str = "near_field"; -const REALTIME_V2_TURN_DETECTION: &str = "server_vad"; -const REALTIME_V2_OUTPUT_MODALITY_AUDIO: &str = "audio"; -const REALTIME_V2_TOOL_CHOICE: &str = "auto"; -const REALTIME_V1_SESSION_TYPE: &str = "quicksilver"; -const REALTIME_V2_SESSION_TYPE: &str = "realtime"; -const REALTIME_V2_CODEX_TOOL_NAME: &str = "codex"; -const REALTIME_V2_CODEX_TOOL_DESCRIPTION: &str = "Delegate a request to Codex and return the final result to the user. Use this as the default action. If the user asks to do something next, later, after this, or once current work finishes, call this tool so the work is actually queued instead of merely promising to do it later."; - -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<()>, @@ -323,21 +292,8 @@ 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: content_kind.to_string(), - text, - }], - }), - }) - .await + self.send_json(conversation_item_create_message(self.event_parser, text)) + .await } pub async fn send_conversation_handoff_append( @@ -345,23 +301,12 @@ impl RealtimeWebsocketWriter { handoff_id: String, output_text: String, ) -> Result<(), ApiError> { - let message = match self.event_parser { - RealtimeEventParser::V1 => RealtimeOutboundMessage::ConversationHandoffAppend { - handoff_id, - output_text, - }, - RealtimeEventParser::RealtimeV2 => RealtimeOutboundMessage::ConversationItemCreate { - item: ConversationItemPayload::FunctionCallOutput( - ConversationFunctionCallOutputItem { - kind: "function_call_output".to_string(), - call_id: handoff_id, - output: output_text, - }, - ), - }, - }; - - self.send_json(message).await + self.send_json(conversation_handoff_append_message( + self.event_parser, + handoff_id, + output_text, + )) + .await } pub async fn send_response_create(&self) -> Result<(), ApiError> { @@ -389,125 +334,17 @@ impl RealtimeWebsocketWriter { session_mode: RealtimeSessionMode, ) -> Result<(), ApiError> { let session_mode = normalized_session_mode(self.event_parser, session_mode); - let (session_kind, session_instructions, output_modalities, input_audio, 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(), - }; - let input = match self.event_parser { - RealtimeEventParser::V1 => SessionAudioInput { - format: SessionAudioFormat { - kind: REALTIME_AUDIO_FORMAT.to_string(), - rate: REALTIME_AUDIO_SAMPLE_RATE, - }, - noise_reduction: None, - turn_detection: None, - }, - RealtimeEventParser::RealtimeV2 => SessionAudioInput { - format: SessionAudioFormat { - kind: REALTIME_AUDIO_FORMAT.to_string(), - rate: REALTIME_AUDIO_SAMPLE_RATE, - }, - noise_reduction: Some(SessionNoiseReduction { - kind: REALTIME_V2_NOISE_REDUCTION.to_string(), - }), - turn_detection: Some(SessionTurnDetection { - kind: REALTIME_V2_TURN_DETECTION.to_string(), - interrupt_response: true, - create_response: true, - }), - }, - }; - let output = match self.event_parser { - RealtimeEventParser::V1 => SessionAudioOutput { - format: None, - voice: SessionAudioVoice::Fathom, - }, - RealtimeEventParser::RealtimeV2 => SessionAudioOutput { - format: Some(SessionAudioOutputFormat { - kind: REALTIME_AUDIO_FORMAT.to_string(), - rate: REALTIME_AUDIO_SAMPLE_RATE, - }), - voice: SessionAudioVoice::Marin, - }, - }; - let output_modalities = match self.event_parser { - RealtimeEventParser::V1 => None, - RealtimeEventParser::RealtimeV2 => { - Some(vec![REALTIME_V2_OUTPUT_MODALITY_AUDIO.to_string()]) - } - }; - ( - kind, - Some(instructions), - output_modalities, - input, - Some(output), - ) - } - RealtimeSessionMode::Transcription => ( - "transcription".to_string(), - None, - None, - SessionAudioInput { - format: SessionAudioFormat { - kind: REALTIME_AUDIO_FORMAT.to_string(), - rate: REALTIME_AUDIO_SAMPLE_RATE, - }, - noise_reduction: None, - turn_detection: None, - }, - None, - ), - }; - let (tools, tool_choice) = match (self.event_parser, session_mode) { - (RealtimeEventParser::RealtimeV2, RealtimeSessionMode::Conversational) => ( - 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": "The user request to delegate to Codex." - } - }, - "required": ["prompt"], - "additionalProperties": false - }), - }]), - Some(REALTIME_V2_TOOL_CHOICE.to_string()), - ), - (RealtimeEventParser::RealtimeV2, RealtimeSessionMode::Transcription) - | (RealtimeEventParser::V1, RealtimeSessionMode::Conversational) - | (RealtimeEventParser::V1, RealtimeSessionMode::Transcription) => (None, None), - }; + let session = session_update_session(self.event_parser, instructions, session_mode); debug!( event_parser = ?self.event_parser, session_mode = ?session_mode, - instructions_len = session_instructions.as_ref().map(String::len).unwrap_or_default(), - has_output_audio = output_audio.is_some(), - has_tools = tools.is_some(), + instructions_len = session.instructions.as_ref().map(String::len).unwrap_or_default(), + has_output_audio = session.audio.output.is_some(), + has_tools = session.tools.is_some(), "realtime websocket prepared session.update" ); - self.send_json(RealtimeOutboundMessage::SessionUpdate { - session: SessionUpdateSession { - kind: session_kind, - instructions: session_instructions, - output_modalities, - audio: SessionAudio { - input: input_audio, - output: output_audio, - }, - tools, - tool_choice, - }, - }) - .await + self.send_json(RealtimeOutboundMessage::SessionUpdate { session }) + .await } pub async fn close(&self) -> Result<(), ApiError> { @@ -775,10 +612,7 @@ fn websocket_url_from_api_url( } } - let intent = match event_parser { - RealtimeEventParser::V1 => Some("quicksilver"), - RealtimeEventParser::RealtimeV2 => None, - }; + let intent = websocket_intent(event_parser); let has_extra_query_params = query_params.is_some_and(|query_params| { query_params .iter() diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods_common.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods_common.rs new file mode 100644 index 0000000000..4a5013c656 --- /dev/null +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods_common.rs @@ -0,0 +1,67 @@ +use crate::endpoint::realtime_websocket::methods_v1::conversation_handoff_append_message as v1_conversation_handoff_append_message; +use crate::endpoint::realtime_websocket::methods_v1::conversation_item_create_message as v1_conversation_item_create_message; +use crate::endpoint::realtime_websocket::methods_v1::session_update_session as v1_session_update_session; +use crate::endpoint::realtime_websocket::methods_v1::websocket_intent as v1_websocket_intent; +use crate::endpoint::realtime_websocket::methods_v2::conversation_handoff_append_message as v2_conversation_handoff_append_message; +use crate::endpoint::realtime_websocket::methods_v2::conversation_item_create_message as v2_conversation_item_create_message; +use crate::endpoint::realtime_websocket::methods_v2::session_update_session as v2_session_update_session; +use crate::endpoint::realtime_websocket::methods_v2::websocket_intent as v2_websocket_intent; +use crate::endpoint::realtime_websocket::protocol::RealtimeEventParser; +use crate::endpoint::realtime_websocket::protocol::RealtimeOutboundMessage; +use crate::endpoint::realtime_websocket::protocol::RealtimeSessionMode; +use crate::endpoint::realtime_websocket::protocol::SessionUpdateSession; + +pub(super) const REALTIME_AUDIO_SAMPLE_RATE: u32 = 24_000; +pub(super) const REALTIME_AUDIO_FORMAT: &str = "audio/pcm"; + +pub(super) fn normalized_session_mode( + event_parser: RealtimeEventParser, + session_mode: RealtimeSessionMode, +) -> RealtimeSessionMode { + match event_parser { + RealtimeEventParser::V1 => RealtimeSessionMode::Conversational, + RealtimeEventParser::RealtimeV2 => session_mode, + } +} + +pub(super) fn conversation_item_create_message( + event_parser: RealtimeEventParser, + text: String, +) -> RealtimeOutboundMessage { + match event_parser { + RealtimeEventParser::V1 => v1_conversation_item_create_message(text), + RealtimeEventParser::RealtimeV2 => v2_conversation_item_create_message(text), + } +} + +pub(super) fn conversation_handoff_append_message( + event_parser: RealtimeEventParser, + handoff_id: String, + output_text: String, +) -> RealtimeOutboundMessage { + match event_parser { + RealtimeEventParser::V1 => v1_conversation_handoff_append_message(handoff_id, output_text), + RealtimeEventParser::RealtimeV2 => { + v2_conversation_handoff_append_message(handoff_id, output_text) + } + } +} + +pub(super) fn session_update_session( + event_parser: RealtimeEventParser, + instructions: String, + session_mode: RealtimeSessionMode, +) -> SessionUpdateSession { + let session_mode = normalized_session_mode(event_parser, session_mode); + match event_parser { + RealtimeEventParser::V1 => v1_session_update_session(instructions), + RealtimeEventParser::RealtimeV2 => v2_session_update_session(instructions, session_mode), + } +} + +pub(super) fn websocket_intent(event_parser: RealtimeEventParser) -> Option<&'static str> { + match event_parser { + RealtimeEventParser::V1 => v1_websocket_intent(), + RealtimeEventParser::RealtimeV2 => v2_websocket_intent(), + } +} diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods_v1.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods_v1.rs new file mode 100644 index 0000000000..429d06b005 --- /dev/null +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods_v1.rs @@ -0,0 +1,65 @@ +use crate::endpoint::realtime_websocket::methods_common::REALTIME_AUDIO_FORMAT; +use crate::endpoint::realtime_websocket::methods_common::REALTIME_AUDIO_SAMPLE_RATE; +use crate::endpoint::realtime_websocket::protocol::ConversationItemContent; +use crate::endpoint::realtime_websocket::protocol::ConversationItemPayload; +use crate::endpoint::realtime_websocket::protocol::ConversationMessageItem; +use crate::endpoint::realtime_websocket::protocol::RealtimeOutboundMessage; +use crate::endpoint::realtime_websocket::protocol::SessionAudio; +use crate::endpoint::realtime_websocket::protocol::SessionAudioFormat; +use crate::endpoint::realtime_websocket::protocol::SessionAudioInput; +use crate::endpoint::realtime_websocket::protocol::SessionAudioOutput; +use crate::endpoint::realtime_websocket::protocol::SessionAudioVoice; +use crate::endpoint::realtime_websocket::protocol::SessionUpdateSession; + +const REALTIME_V1_SESSION_TYPE: &str = "quicksilver"; + +pub(super) fn conversation_item_create_message(text: String) -> RealtimeOutboundMessage { + RealtimeOutboundMessage::ConversationItemCreate { + item: ConversationItemPayload::Message(ConversationMessageItem { + kind: "message".to_string(), + role: "user".to_string(), + content: vec![ConversationItemContent { + kind: "text".to_string(), + text, + }], + }), + } +} + +pub(super) fn conversation_handoff_append_message( + handoff_id: String, + output_text: String, +) -> RealtimeOutboundMessage { + RealtimeOutboundMessage::ConversationHandoffAppend { + handoff_id, + output_text, + } +} + +pub(super) fn session_update_session(instructions: String) -> SessionUpdateSession { + SessionUpdateSession { + kind: REALTIME_V1_SESSION_TYPE.to_string(), + instructions: Some(instructions), + output_modalities: None, + audio: SessionAudio { + input: SessionAudioInput { + format: SessionAudioFormat { + kind: REALTIME_AUDIO_FORMAT.to_string(), + rate: REALTIME_AUDIO_SAMPLE_RATE, + }, + noise_reduction: None, + turn_detection: None, + }, + output: Some(SessionAudioOutput { + format: None, + voice: SessionAudioVoice::Fathom, + }), + }, + tools: None, + tool_choice: None, + } +} + +pub(super) fn websocket_intent() -> Option<&'static str> { + Some("quicksilver") +} diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods_v2.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods_v2.rs new file mode 100644 index 0000000000..50de80610e --- /dev/null +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods_v2.rs @@ -0,0 +1,128 @@ +use crate::endpoint::realtime_websocket::methods_common::REALTIME_AUDIO_FORMAT; +use crate::endpoint::realtime_websocket::methods_common::REALTIME_AUDIO_SAMPLE_RATE; +use crate::endpoint::realtime_websocket::protocol::ConversationFunctionCallOutputItem; +use crate::endpoint::realtime_websocket::protocol::ConversationItemContent; +use crate::endpoint::realtime_websocket::protocol::ConversationItemPayload; +use crate::endpoint::realtime_websocket::protocol::ConversationMessageItem; +use crate::endpoint::realtime_websocket::protocol::RealtimeOutboundMessage; +use crate::endpoint::realtime_websocket::protocol::RealtimeSessionMode; +use crate::endpoint::realtime_websocket::protocol::SessionAudio; +use crate::endpoint::realtime_websocket::protocol::SessionAudioFormat; +use crate::endpoint::realtime_websocket::protocol::SessionAudioInput; +use crate::endpoint::realtime_websocket::protocol::SessionAudioOutput; +use crate::endpoint::realtime_websocket::protocol::SessionAudioOutputFormat; +use crate::endpoint::realtime_websocket::protocol::SessionAudioVoice; +use crate::endpoint::realtime_websocket::protocol::SessionFunctionTool; +use crate::endpoint::realtime_websocket::protocol::SessionNoiseReduction; +use crate::endpoint::realtime_websocket::protocol::SessionTurnDetection; +use crate::endpoint::realtime_websocket::protocol::SessionUpdateSession; +use serde_json::json; + +const REALTIME_V2_NOISE_REDUCTION: &str = "near_field"; +const REALTIME_V2_TURN_DETECTION: &str = "server_vad"; +const REALTIME_V2_OUTPUT_MODALITY_AUDIO: &str = "audio"; +const REALTIME_V2_TOOL_CHOICE: &str = "auto"; +const REALTIME_V2_SESSION_TYPE: &str = "realtime"; +const REALTIME_V2_CODEX_TOOL_NAME: &str = "codex"; +const REALTIME_V2_CODEX_TOOL_DESCRIPTION: &str = "Delegate a request to Codex and return the final result to the user. Use this as the default action. If the user asks to do something next, later, after this, or once current work finishes, call this tool so the work is actually queued instead of merely promising to do it later."; + +pub(super) fn conversation_item_create_message(text: String) -> RealtimeOutboundMessage { + RealtimeOutboundMessage::ConversationItemCreate { + item: ConversationItemPayload::Message(ConversationMessageItem { + kind: "message".to_string(), + role: "user".to_string(), + content: vec![ConversationItemContent { + kind: "input_text".to_string(), + text, + }], + }), + } +} + +pub(super) fn conversation_handoff_append_message( + handoff_id: String, + output_text: String, +) -> RealtimeOutboundMessage { + RealtimeOutboundMessage::ConversationItemCreate { + item: ConversationItemPayload::FunctionCallOutput(ConversationFunctionCallOutputItem { + kind: "function_call_output".to_string(), + call_id: handoff_id, + output: output_text, + }), + } +} + +pub(super) fn session_update_session( + instructions: String, + session_mode: RealtimeSessionMode, +) -> SessionUpdateSession { + match session_mode { + RealtimeSessionMode::Conversational => SessionUpdateSession { + kind: REALTIME_V2_SESSION_TYPE.to_string(), + instructions: Some(instructions), + output_modalities: Some(vec![REALTIME_V2_OUTPUT_MODALITY_AUDIO.to_string()]), + audio: SessionAudio { + input: SessionAudioInput { + format: SessionAudioFormat { + kind: REALTIME_AUDIO_FORMAT.to_string(), + rate: REALTIME_AUDIO_SAMPLE_RATE, + }, + noise_reduction: Some(SessionNoiseReduction { + kind: REALTIME_V2_NOISE_REDUCTION.to_string(), + }), + turn_detection: Some(SessionTurnDetection { + kind: REALTIME_V2_TURN_DETECTION.to_string(), + interrupt_response: true, + create_response: true, + }), + }, + output: Some(SessionAudioOutput { + format: Some(SessionAudioOutputFormat { + kind: REALTIME_AUDIO_FORMAT.to_string(), + rate: REALTIME_AUDIO_SAMPLE_RATE, + }), + voice: SessionAudioVoice::Marin, + }), + }, + tools: 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": "The user request to delegate to Codex." + } + }, + "required": ["prompt"], + "additionalProperties": false + }), + }]), + tool_choice: Some(REALTIME_V2_TOOL_CHOICE.to_string()), + }, + RealtimeSessionMode::Transcription => SessionUpdateSession { + kind: "transcription".to_string(), + instructions: None, + output_modalities: None, + audio: SessionAudio { + input: SessionAudioInput { + format: SessionAudioFormat { + kind: REALTIME_AUDIO_FORMAT.to_string(), + rate: REALTIME_AUDIO_SAMPLE_RATE, + }, + noise_reduction: None, + turn_detection: None, + }, + output: None, + }, + tools: None, + tool_choice: None, + }, + } +} + +pub(super) fn websocket_intent() -> Option<&'static str> { + None +} 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 f307e60914..d13585034a 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/mod.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/mod.rs @@ -1,4 +1,7 @@ pub mod methods; +mod methods_common; +mod methods_v1; +mod methods_v2; pub mod protocol; mod protocol_common; mod protocol_v1;