From a1a5c65059396bb383d28ccfd3819349e3b2a03b Mon Sep 17 00:00:00 2001 From: Ahmed Ibrahim Date: Mon, 20 Apr 2026 19:22:15 -0700 Subject: [PATCH] Make realtime extraction layout idiomatic --- codex-rs/core/src/lib.rs | 1 - .../src/{realtime_adapter.rs => realtime.rs} | 8 - codex-rs/realtime/src/handoff.rs | 54 +++++ codex-rs/realtime/src/lib.rs | 26 ++- .../realtime/src/realtime_conversation.rs | 216 +----------------- .../src/realtime_conversation_tests.rs | 19 +- codex-rs/realtime/src/session_config.rs | 160 +++++++++++++ 7 files changed, 246 insertions(+), 238 deletions(-) rename codex-rs/core/src/{realtime_adapter.rs => realtime.rs} (98%) create mode 100644 codex-rs/realtime/src/handoff.rs rename codex-rs/{core => realtime}/src/realtime_conversation_tests.rs (85%) create mode 100644 codex-rs/realtime/src/session_config.rs diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 3d61f0cf50..ef0ee4479b 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -11,7 +11,6 @@ mod apps; mod arc_monitor; mod client; mod client_common; -#[path = "realtime_adapter.rs"] mod realtime; pub(crate) mod session; pub use session::SteerInputError; diff --git a/codex-rs/core/src/realtime_adapter.rs b/codex-rs/core/src/realtime.rs similarity index 98% rename from codex-rs/core/src/realtime_adapter.rs rename to codex-rs/core/src/realtime.rs index da0fccca71..3ed6414600 100644 --- a/codex-rs/core/src/realtime_adapter.rs +++ b/codex-rs/core/src/realtime.rs @@ -48,10 +48,6 @@ pub(crate) use codex_realtime::prefix_realtime_v2_text; use codex_realtime::realtime_api_key; use codex_realtime::realtime_delegation_from_handoff; use codex_realtime::realtime_request_headers; -#[cfg(test)] -use codex_realtime::realtime_text_from_handoff_request; -#[cfg(test)] -use codex_realtime::wrap_realtime_delegation_input; const MAX_RECENT_THREADS: usize = 40; const REALTIME_STARTUP_CONTEXT_TOKEN_BUDGET: usize = 5_300; @@ -528,10 +524,6 @@ async fn send_realtime_conversation_closed( .await; } -#[cfg(test)] -#[path = "realtime_conversation_tests.rs"] -mod conversation_tests; - #[cfg(test)] #[path = "realtime_context_tests.rs"] mod context_tests; diff --git a/codex-rs/realtime/src/handoff.rs b/codex-rs/realtime/src/handoff.rs new file mode 100644 index 0000000000..2f9417a38c --- /dev/null +++ b/codex-rs/realtime/src/handoff.rs @@ -0,0 +1,54 @@ +use codex_protocol::protocol::RealtimeHandoffRequested; + +pub const REALTIME_USER_TEXT_PREFIX: &str = "[USER] "; +pub const REALTIME_BACKEND_TEXT_PREFIX: &str = "[BACKEND] "; + +pub fn prefix_realtime_v2_text(text: String, prefix: &str) -> String { + if text.is_empty() || text.starts_with(prefix) { + return text; + } + format!("{prefix}{text}") +} + +fn realtime_transcript_delta_from_handoff(handoff: &RealtimeHandoffRequested) -> Option { + let active_transcript = handoff + .active_transcript + .iter() + .map(|entry| format!("{role}: {text}", role = entry.role, text = entry.text)) + .collect::>() + .join("\n"); + (!active_transcript.is_empty()).then_some(active_transcript) +} + +pub fn realtime_text_from_handoff_request(handoff: &RealtimeHandoffRequested) -> Option { + (!handoff.input_transcript.is_empty()) + .then_some(handoff.input_transcript.clone()) + .or_else(|| realtime_transcript_delta_from_handoff(handoff)) +} + +pub fn realtime_delegation_from_handoff(handoff: &RealtimeHandoffRequested) -> Option { + let input = realtime_text_from_handoff_request(handoff)?; + Some(wrap_realtime_delegation_input( + &input, + realtime_transcript_delta_from_handoff(handoff).as_deref(), + )) +} + +pub fn wrap_realtime_delegation_input(input: &str, transcript_delta: Option<&str>) -> String { + let input = escape_xml_text(input); + if let Some(transcript_delta) = transcript_delta.filter(|text| !text.is_empty()) { + let transcript_delta = escape_xml_text(transcript_delta); + return format!( + "\n {input}\n {transcript_delta}\n" + ); + } + + format!("\n {input}\n") +} + +fn escape_xml_text(input: &str) -> String { + input + .replace('&', "&") + .replace('<', "<") + .replace('>', ">") +} diff --git a/codex-rs/realtime/src/lib.rs b/codex-rs/realtime/src/lib.rs index 8f63725b58..e791dc532b 100644 --- a/codex-rs/realtime/src/lib.rs +++ b/codex-rs/realtime/src/lib.rs @@ -1,25 +1,27 @@ mod config; +mod handoff; mod prompt; mod realtime_conversation; +mod session_config; mod startup_context; pub use config::RealtimeFeaturesConfig; +pub use handoff::REALTIME_BACKEND_TEXT_PREFIX; +pub use handoff::REALTIME_USER_TEXT_PREFIX; +pub use handoff::prefix_realtime_v2_text; +pub use handoff::realtime_delegation_from_handoff; +pub use handoff::realtime_text_from_handoff_request; +pub use handoff::wrap_realtime_delegation_input; pub use prompt::prepare_realtime_backend_prompt; -pub use realtime_conversation::DEFAULT_REALTIME_MODEL; -pub use realtime_conversation::REALTIME_BACKEND_TEXT_PREFIX; -pub use realtime_conversation::REALTIME_USER_TEXT_PREFIX; pub use realtime_conversation::RealtimeConversationManager; pub use realtime_conversation::RealtimeStart; pub use realtime_conversation::RealtimeStartOutput; -pub use realtime_conversation::build_realtime_session_config; -pub use realtime_conversation::default_realtime_voice; -pub use realtime_conversation::prefix_realtime_v2_text; -pub use realtime_conversation::realtime_api_key; -pub use realtime_conversation::realtime_delegation_from_handoff; -pub use realtime_conversation::realtime_request_headers; -pub use realtime_conversation::realtime_text_from_handoff_request; -pub use realtime_conversation::validate_realtime_voice; -pub use realtime_conversation::wrap_realtime_delegation_input; +pub use session_config::DEFAULT_REALTIME_MODEL; +pub use session_config::build_realtime_session_config; +pub use session_config::default_realtime_voice; +pub use session_config::realtime_api_key; +pub use session_config::realtime_request_headers; +pub use session_config::validate_realtime_voice; pub use startup_context::CURRENT_THREAD_SECTION_TOKEN_BUDGET; pub use startup_context::NOTES_SECTION_TOKEN_BUDGET; pub use startup_context::REALTIME_TURN_TOKEN_BUDGET; diff --git a/codex-rs/realtime/src/realtime_conversation.rs b/codex-rs/realtime/src/realtime_conversation.rs index e6c4b56132..999018c43f 100644 --- a/codex-rs/realtime/src/realtime_conversation.rs +++ b/codex-rs/realtime/src/realtime_conversation.rs @@ -1,5 +1,6 @@ -use crate::config::RealtimeFeaturesConfig; -use crate::prompt::prepare_realtime_backend_prompt; +use crate::handoff::REALTIME_BACKEND_TEXT_PREFIX; +use crate::handoff::REALTIME_USER_TEXT_PREFIX; +use crate::handoff::prefix_realtime_v2_text; use anyhow::Context; use async_channel::Receiver; use async_channel::RecvError; @@ -11,25 +12,11 @@ use codex_api::ApiError; use codex_api::RealtimeAudioFrame; use codex_api::RealtimeEvent; use codex_api::RealtimeEventParser; -use codex_api::RealtimeSessionConfig; -use codex_api::RealtimeSessionMode; use codex_api::RealtimeWebsocketEvents; use codex_api::RealtimeWebsocketWriter; use codex_api::map_api_error; -use codex_config::config_toml::RealtimeWsMode; -use codex_config::config_toml::RealtimeWsVersion; -use codex_login::CodexAuth; -use codex_login::read_openai_api_key_from_env; -use codex_model_provider_info::ModelProviderInfo; use codex_protocol::error::CodexErr; use codex_protocol::error::Result as CodexResult; -use codex_protocol::protocol::RealtimeHandoffRequested; -use codex_protocol::protocol::RealtimeOutputModality; -use codex_protocol::protocol::RealtimeVoice; -use codex_protocol::protocol::RealtimeVoicesList; -use http::HeaderMap; -use http::HeaderValue; -use http::header::AUTHORIZATION; use serde_json::json; use std::sync::Arc; use std::sync::atomic::AtomicBool; @@ -44,9 +31,6 @@ const AUDIO_IN_QUEUE_CAPACITY: usize = 256; const USER_TEXT_IN_QUEUE_CAPACITY: usize = 64; const HANDOFF_OUT_QUEUE_CAPACITY: usize = 64; const OUTPUT_EVENTS_QUEUE_CAPACITY: usize = 256; -pub const DEFAULT_REALTIME_MODEL: &str = "gpt-realtime-1.5"; -pub const REALTIME_USER_TEXT_PREFIX: &str = "[USER] "; -pub const REALTIME_BACKEND_TEXT_PREFIX: &str = "[BACKEND] "; const REALTIME_V2_HANDOFF_COMPLETE_ACKNOWLEDGEMENT: &str = "Background agent finished. Use the preceding [BACKEND] messages as the result."; const REALTIME_V2_STEER_ACKNOWLEDGEMENT: &str = @@ -203,194 +187,10 @@ pub struct RealtimeStartOutput { pub sdp: Option, } -pub fn build_realtime_session_config( - config: &RealtimeFeaturesConfig, - prompt: Option>, - session_id: Option, - output_modality: RealtimeOutputModality, - voice: Option, - startup_context: String, -) -> CodexResult { - let prompt = prepare_realtime_backend_prompt(prompt, config.websocket_backend_prompt.clone()); - let startup_context = config - .websocket_startup_context - .clone() - .unwrap_or(startup_context); - let prompt = match (prompt.is_empty(), startup_context.is_empty()) { - (true, true) => String::new(), - (true, false) => startup_context, - (false, true) => prompt, - (false, false) => format!("{prompt}\n\n{startup_context}"), - }; - let model = Some( - config - .websocket_model - .clone() - .unwrap_or_else(|| DEFAULT_REALTIME_MODEL.to_string()), - ); - let event_parser = match config.session.version { - RealtimeWsVersion::V1 => RealtimeEventParser::V1, - RealtimeWsVersion::V2 => RealtimeEventParser::RealtimeV2, - }; - if config.session.version == RealtimeWsVersion::V1 - && matches!(output_modality, RealtimeOutputModality::Text) - { - return Err(CodexErr::InvalidRequest( - "text realtime output modality requires realtime v2".to_string(), - )); +impl Default for RealtimeConversationManager { + fn default() -> Self { + Self::new() } - let session_mode = match config.session.session_type { - RealtimeWsMode::Conversational => RealtimeSessionMode::Conversational, - RealtimeWsMode::Transcription => RealtimeSessionMode::Transcription, - }; - let voice = voice - .or(config.session.voice) - .unwrap_or_else(|| default_realtime_voice(config.session.version)); - validate_realtime_voice(config.session.version, voice)?; - Ok(RealtimeSessionConfig { - instructions: prompt, - model, - session_id: Some(session_id.unwrap_or_default()), - event_parser, - session_mode, - output_modality, - voice, - }) -} - -pub fn default_realtime_voice(version: RealtimeWsVersion) -> RealtimeVoice { - let voices = RealtimeVoicesList::builtin(); - match version { - RealtimeWsVersion::V1 => voices.default_v1, - RealtimeWsVersion::V2 => voices.default_v2, - } -} - -pub fn prefix_realtime_v2_text(text: String, prefix: &str) -> String { - if text.is_empty() || text.starts_with(prefix) { - return text; - } - format!("{prefix}{text}") -} - -pub fn validate_realtime_voice( - version: RealtimeWsVersion, - voice: RealtimeVoice, -) -> CodexResult<()> { - let voices = RealtimeVoicesList::builtin(); - let allowed = match version { - RealtimeWsVersion::V1 => &voices.v1, - RealtimeWsVersion::V2 => &voices.v2, - }; - if allowed.contains(&voice) { - return Ok(()); - } - - let version = match version { - RealtimeWsVersion::V1 => "v1", - RealtimeWsVersion::V2 => "v2", - }; - let allowed = allowed - .iter() - .map(|voice| voice.wire_name()) - .collect::>() - .join(", "); - Err(CodexErr::InvalidRequest(format!( - "realtime voice `{}` is not supported for {version}; supported voices: {allowed}", - voice.wire_name() - ))) -} - -fn realtime_transcript_delta_from_handoff(handoff: &RealtimeHandoffRequested) -> Option { - let active_transcript = handoff - .active_transcript - .iter() - .map(|entry| format!("{role}: {text}", role = entry.role, text = entry.text)) - .collect::>() - .join("\n"); - (!active_transcript.is_empty()).then_some(active_transcript) -} - -pub fn realtime_text_from_handoff_request(handoff: &RealtimeHandoffRequested) -> Option { - (!handoff.input_transcript.is_empty()) - .then_some(handoff.input_transcript.clone()) - .or_else(|| realtime_transcript_delta_from_handoff(handoff)) -} - -pub fn realtime_delegation_from_handoff(handoff: &RealtimeHandoffRequested) -> Option { - let input = realtime_text_from_handoff_request(handoff)?; - Some(wrap_realtime_delegation_input( - &input, - realtime_transcript_delta_from_handoff(handoff).as_deref(), - )) -} - -pub fn wrap_realtime_delegation_input(input: &str, transcript_delta: Option<&str>) -> String { - let input = escape_xml_text(input); - if let Some(transcript_delta) = transcript_delta.filter(|text| !text.is_empty()) { - let transcript_delta = escape_xml_text(transcript_delta); - return format!( - "\n {input}\n {transcript_delta}\n" - ); - } - - format!("\n {input}\n") -} - -fn escape_xml_text(input: &str) -> String { - input - .replace('&', "&") - .replace('<', "<") - .replace('>', ">") -} - -pub fn realtime_api_key( - auth: Option<&CodexAuth>, - provider: &ModelProviderInfo, -) -> CodexResult { - if let Some(api_key) = provider.api_key()? { - return Ok(api_key); - } - - if let Some(token) = provider.experimental_bearer_token.clone() { - return Ok(token); - } - - if let Some(api_key) = auth.and_then(CodexAuth::api_key) { - return Ok(api_key.to_string()); - } - - if provider.is_openai() - && let Some(api_key) = read_openai_api_key_from_env() - { - return Ok(api_key); - } - - Err(CodexErr::InvalidRequest( - "realtime conversation requires API key auth".to_string(), - )) -} - -pub fn realtime_request_headers( - session_id: Option<&str>, - api_key: Option<&str>, -) -> CodexResult> { - let mut headers = HeaderMap::new(); - - if let Some(session_id) = session_id - && let Ok(session_id) = HeaderValue::from_str(session_id) - { - headers.insert("x-session-id", session_id); - } - - if let Some(api_key) = api_key { - let auth_value = HeaderValue::from_str(&format!("Bearer {api_key}")).map_err(|err| { - CodexErr::InvalidRequest(format!("invalid realtime api key header: {err}")) - })?; - headers.insert(AUTHORIZATION, auth_value); - } - - Ok(Some(headers)) } impl RealtimeConversationManager { @@ -1034,3 +834,7 @@ fn decoded_samples_per_channel(frame: &RealtimeAudioFrame) -> Option { let samples = bytes.len().checked_div(2)?.checked_div(channels)?; u32::try_from(samples).ok() } + +#[cfg(test)] +#[path = "realtime_conversation_tests.rs"] +mod tests; diff --git a/codex-rs/core/src/realtime_conversation_tests.rs b/codex-rs/realtime/src/realtime_conversation_tests.rs similarity index 85% rename from codex-rs/core/src/realtime_conversation_tests.rs rename to codex-rs/realtime/src/realtime_conversation_tests.rs index a146c43869..2fe036004d 100644 --- a/codex-rs/core/src/realtime_conversation_tests.rs +++ b/codex-rs/realtime/src/realtime_conversation_tests.rs @@ -1,8 +1,5 @@ use super::RealtimeHandoffState; use super::RealtimeSessionKind; -use super::realtime_delegation_from_handoff; -use super::realtime_text_from_handoff_request; -use super::wrap_realtime_delegation_input; use async_channel::bounded; use codex_protocol::protocol::RealtimeHandoffRequested; use codex_protocol::protocol::RealtimeTranscriptEntry; @@ -26,7 +23,7 @@ fn prefers_handoff_input_transcript_over_active_transcript() { ], }; assert_eq!( - realtime_text_from_handoff_request(&handoff), + crate::realtime_text_from_handoff_request(&handoff), Some("ignored".to_string()) ); } @@ -43,7 +40,7 @@ fn extracts_text_from_handoff_request_active_transcript_if_input_missing() { }], }; assert_eq!( - realtime_text_from_handoff_request(&handoff), + crate::realtime_text_from_handoff_request(&handoff), Some("user: hello".to_string()) ); } @@ -66,7 +63,7 @@ fn wraps_handoff_with_transcript_delta() { ], }; assert_eq!( - realtime_delegation_from_handoff(&handoff), + crate::realtime_delegation_from_handoff(&handoff), Some( "\n delegate this\n user: hello\nassistant: hi there\n" .to_string() @@ -83,7 +80,7 @@ fn extracts_text_from_handoff_request_input_transcript_if_messages_missing() { active_transcript: vec![], }; assert_eq!( - realtime_text_from_handoff_request(&handoff), + crate::realtime_text_from_handoff_request(&handoff), Some("ignored".to_string()) ); } @@ -96,13 +93,13 @@ fn ignores_empty_handoff_request_input_transcript() { input_transcript: String::new(), active_transcript: vec![], }; - assert_eq!(realtime_text_from_handoff_request(&handoff), None); + assert_eq!(crate::realtime_text_from_handoff_request(&handoff), None); } #[test] fn wraps_realtime_delegation_input() { assert_eq!( - wrap_realtime_delegation_input("hello", /*transcript_delta*/ None), + crate::wrap_realtime_delegation_input("hello", /*transcript_delta*/ None), "\n hello\n" ); } @@ -110,7 +107,7 @@ fn wraps_realtime_delegation_input() { #[test] fn wraps_realtime_delegation_input_with_xml_escaping() { assert_eq!( - wrap_realtime_delegation_input("use a < b && c > d", Some("saw ")), + crate::wrap_realtime_delegation_input("use a < b && c > d", Some("saw ")), "\n use a < b && c > d\n saw <that>\n" ); } @@ -118,7 +115,7 @@ fn wraps_realtime_delegation_input_with_xml_escaping() { #[test] fn wraps_realtime_delegation_input_with_xml_escaping_without_transcript() { assert_eq!( - wrap_realtime_delegation_input("use a < b && c > d", /*transcript_delta*/ None), + crate::wrap_realtime_delegation_input("use a < b && c > d", /*transcript_delta*/ None), "\n use a < b && c > d\n" ); } diff --git a/codex-rs/realtime/src/session_config.rs b/codex-rs/realtime/src/session_config.rs new file mode 100644 index 0000000000..cccdbd4fd3 --- /dev/null +++ b/codex-rs/realtime/src/session_config.rs @@ -0,0 +1,160 @@ +use crate::config::RealtimeFeaturesConfig; +use crate::prompt::prepare_realtime_backend_prompt; +use codex_api::RealtimeEventParser; +use codex_api::RealtimeSessionConfig; +use codex_api::RealtimeSessionMode; +use codex_config::config_toml::RealtimeWsMode; +use codex_config::config_toml::RealtimeWsVersion; +use codex_login::CodexAuth; +use codex_login::read_openai_api_key_from_env; +use codex_model_provider_info::ModelProviderInfo; +use codex_protocol::error::CodexErr; +use codex_protocol::error::Result as CodexResult; +use codex_protocol::protocol::RealtimeOutputModality; +use codex_protocol::protocol::RealtimeVoice; +use codex_protocol::protocol::RealtimeVoicesList; +use http::HeaderMap; +use http::HeaderValue; +use http::header::AUTHORIZATION; + +pub const DEFAULT_REALTIME_MODEL: &str = "gpt-realtime-1.5"; + +pub fn build_realtime_session_config( + config: &RealtimeFeaturesConfig, + prompt: Option>, + session_id: Option, + output_modality: RealtimeOutputModality, + voice: Option, + startup_context: String, +) -> CodexResult { + let prompt = prepare_realtime_backend_prompt(prompt, config.websocket_backend_prompt.clone()); + let startup_context = config + .websocket_startup_context + .clone() + .unwrap_or(startup_context); + let prompt = match (prompt.is_empty(), startup_context.is_empty()) { + (true, true) => String::new(), + (true, false) => startup_context, + (false, true) => prompt, + (false, false) => format!("{prompt}\n\n{startup_context}"), + }; + let model = Some( + config + .websocket_model + .clone() + .unwrap_or_else(|| DEFAULT_REALTIME_MODEL.to_string()), + ); + let event_parser = match config.session.version { + RealtimeWsVersion::V1 => RealtimeEventParser::V1, + RealtimeWsVersion::V2 => RealtimeEventParser::RealtimeV2, + }; + if config.session.version == RealtimeWsVersion::V1 + && matches!(output_modality, RealtimeOutputModality::Text) + { + return Err(CodexErr::InvalidRequest( + "text realtime output modality requires realtime v2".to_string(), + )); + } + let session_mode = match config.session.session_type { + RealtimeWsMode::Conversational => RealtimeSessionMode::Conversational, + RealtimeWsMode::Transcription => RealtimeSessionMode::Transcription, + }; + let voice = voice + .or(config.session.voice) + .unwrap_or_else(|| default_realtime_voice(config.session.version)); + validate_realtime_voice(config.session.version, voice)?; + Ok(RealtimeSessionConfig { + instructions: prompt, + model, + session_id: Some(session_id.unwrap_or_default()), + event_parser, + session_mode, + output_modality, + voice, + }) +} + +pub fn default_realtime_voice(version: RealtimeWsVersion) -> RealtimeVoice { + let voices = RealtimeVoicesList::builtin(); + match version { + RealtimeWsVersion::V1 => voices.default_v1, + RealtimeWsVersion::V2 => voices.default_v2, + } +} + +pub fn validate_realtime_voice( + version: RealtimeWsVersion, + voice: RealtimeVoice, +) -> CodexResult<()> { + let voices = RealtimeVoicesList::builtin(); + let allowed = match version { + RealtimeWsVersion::V1 => &voices.v1, + RealtimeWsVersion::V2 => &voices.v2, + }; + if allowed.contains(&voice) { + return Ok(()); + } + + let version = match version { + RealtimeWsVersion::V1 => "v1", + RealtimeWsVersion::V2 => "v2", + }; + let allowed = allowed + .iter() + .map(|voice| voice.wire_name()) + .collect::>() + .join(", "); + Err(CodexErr::InvalidRequest(format!( + "realtime voice `{}` is not supported for {version}; supported voices: {allowed}", + voice.wire_name() + ))) +} + +pub fn realtime_api_key( + auth: Option<&CodexAuth>, + provider: &ModelProviderInfo, +) -> CodexResult { + if let Some(api_key) = provider.api_key()? { + return Ok(api_key); + } + + if let Some(token) = provider.experimental_bearer_token.clone() { + return Ok(token); + } + + if let Some(api_key) = auth.and_then(CodexAuth::api_key) { + return Ok(api_key.to_string()); + } + + if provider.is_openai() + && let Some(api_key) = read_openai_api_key_from_env() + { + return Ok(api_key); + } + + Err(CodexErr::InvalidRequest( + "realtime conversation requires API key auth".to_string(), + )) +} + +pub fn realtime_request_headers( + session_id: Option<&str>, + api_key: Option<&str>, +) -> CodexResult> { + let mut headers = HeaderMap::new(); + + if let Some(session_id) = session_id + && let Ok(session_id) = HeaderValue::from_str(session_id) + { + headers.insert("x-session-id", session_id); + } + + if let Some(api_key) = api_key { + let auth_value = HeaderValue::from_str(&format!("Bearer {api_key}")).map_err(|err| { + CodexErr::InvalidRequest(format!("invalid realtime api key header: {err}")) + })?; + headers.insert(AUTHORIZATION, auth_value); + } + + Ok(Some(headers)) +}