openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-24 18:18:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

Make realtime extraction layout idiomatic

Browse Source
This commit is contained in:
Ahmed Ibrahim
2026-04-20 19:22:15 -07:00
parent 1896bff9fe
commit a1a5c65059
7 changed files with 246 additions and 238 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/lib.rs
-1
View File
@@ -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;
codex-rs/core/src/realtime_adapter.rs → codex-rs/core/src/realtime.rs
-8
View File
@@ -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;
codex-rs/realtime/src/handoff.rs
+54
View File
@@ -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<String> {
let active_transcript = handoff
.active_transcript
.iter()
.map(|entry| format!("{role}: {text}", role = entry.role, text = entry.text))
.collect::<Vec<_>>()
.join("\n");
(!active_transcript.is_empty()).then_some(active_transcript)
}
pub fn realtime_text_from_handoff_request(handoff: &RealtimeHandoffRequested) -> Option<String> {
(!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<String> {
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!(
"<realtime_delegation>\n <input>{input}</input>\n <transcript_delta>{transcript_delta}</transcript_delta>\n</realtime_delegation>"
);
}
format!("<realtime_delegation>\n <input>{input}</input>\n</realtime_delegation>")
}
fn escape_xml_text(input: &str) -> String {
input
.replace('&', "&amp;")
.replace('<', "&lt;")
.replace('>', "&gt;")
}
codex-rs/realtime/src/lib.rs
+14 -12
View File
@@ -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;
codex-rs/realtime/src/realtime_conversation.rs
+10 -206
View File
@@ -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<String>,
}
pub fn build_realtime_session_config(
config: &RealtimeFeaturesConfig,
prompt: Option<Option<String>>,
session_id: Option<String>,
output_modality: RealtimeOutputModality,
voice: Option<RealtimeVoice>,
startup_context: String,
) -> CodexResult<RealtimeSessionConfig> {
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::<Vec<_>>()
.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<String> {
let active_transcript = handoff
.active_transcript
.iter()
.map(|entry| format!("{role}: {text}", role = entry.role, text = entry.text))
.collect::<Vec<_>>()
.join("\n");
(!active_transcript.is_empty()).then_some(active_transcript)
}
pub fn realtime_text_from_handoff_request(handoff: &RealtimeHandoffRequested) -> Option<String> {
(!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<String> {
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!(
"<realtime_delegation>\n <input>{input}</input>\n <transcript_delta>{transcript_delta}</transcript_delta>\n</realtime_delegation>"
);
}
format!("<realtime_delegation>\n <input>{input}</input>\n</realtime_delegation>")
}
fn escape_xml_text(input: &str) -> String {
input
.replace('&', "&amp;")
.replace('<', "&lt;")
.replace('>', "&gt;")
}
pub fn realtime_api_key(
auth: Option<&CodexAuth>,
provider: &ModelProviderInfo,
) -> CodexResult<String> {
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<Option<HeaderMap>> {
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<u32> {
let samples = bytes.len().checked_div(2)?.checked_div(channels)?;
u32::try_from(samples).ok()
}
#[cfg(test)]
#[path = "realtime_conversation_tests.rs"]
mod tests;
codex-rs/core/src/realtime_conversation_tests.rs → codex-rs/realtime/src/realtime_conversation_tests.rs
+8 -11
View File
@@ -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(
"<realtime_delegation>\n <input>delegate this</input>\n <transcript_delta>user: hello\nassistant: hi there</transcript_delta>\n</realtime_delegation>"
.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),
"<realtime_delegation>\n <input>hello</input>\n</realtime_delegation>"
);
}
@@ -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 <that>")),
crate::wrap_realtime_delegation_input("use a < b && c > d", Some("saw <that>")),
"<realtime_delegation>\n <input>use a &lt; b &amp;&amp; c &gt; d</input>\n <transcript_delta>saw &lt;that&gt;</transcript_delta>\n</realtime_delegation>"
);
}
@@ -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),
"<realtime_delegation>\n <input>use a &lt; b &amp;&amp; c &gt; d</input>\n</realtime_delegation>"
);
}
codex-rs/realtime/src/session_config.rs
+160
View File
@@ -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<Option<String>>,
session_id: Option<String>,
output_modality: RealtimeOutputModality,
voice: Option<RealtimeVoice>,
startup_context: String,
) -> CodexResult<RealtimeSessionConfig> {
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::<Vec<_>>()
.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<String> {
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<Option<HeaderMap>> {
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))
}