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

refactor: move realtime out of codex-core

Browse Source
This commit is contained in:
Ahmed Ibrahim
2026-04-20 18:58:40 -07:00
parent cefcfe43b9
commit b2b94fe755
16 changed files with 1878 additions and 1597 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+28
View File
@@ -1947,6 +1947,7 @@ dependencies = [
"codex-otel",
"codex-plugin",
"codex-protocol",
"codex-realtime",
"codex-response-debug-context",
"codex-rmcp-client",
"codex-rollout",
@@ -2677,6 +2678,33 @@ dependencies = [
"uuid",
]
[[package]]
name = "codex-realtime"
version = "0.0.0"
dependencies = [
"anyhow",
"async-channel",
"base64 0.22.1",
"chrono",
"codex-api",
"codex-config",
"codex-exec-server",
"codex-git-utils",
"codex-login",
"codex-model-provider-info",
"codex-protocol",
"codex-thread-store",
"codex-utils-absolute-path",
"codex-utils-output-truncation",
"http 1.4.0",
"pretty_assertions",
"serde_json",
"tempfile",
"tokio",
"tracing",
"whoami",
]
[[package]]
name = "codex-realtime-webrtc"
version = "0.0.0"
codex-rs/Cargo.toml
+2
View File
@@ -50,6 +50,7 @@ members = [
"ollama",
"process-hardening",
"protocol",
"realtime",
"realtime-webrtc",
"rollout",
"rmcp-client",
@@ -159,6 +160,7 @@ codex-plugin = { path = "plugin" }
codex-model-provider = { path = "model-provider" }
codex-process-hardening = { path = "process-hardening" }
codex-protocol = { path = "protocol" }
codex-realtime = { path = "realtime" }
codex-realtime-webrtc = { path = "realtime-webrtc" }
codex-responses-api-proxy = { path = "responses-api-proxy" }
codex-response-debug-context = { path = "response-debug-context" }
codex-rs/core/Cargo.toml
+1
View File
@@ -55,6 +55,7 @@ codex-plugin = { workspace = true }
codex-model-provider = { workspace = true }
codex-protocol = { workspace = true }
codex-response-debug-context = { workspace = true }
codex-realtime = { workspace = true }
codex-rollout = { workspace = true }
codex-rmcp-client = { workspace = true }
codex-sandboxing = { workspace = true }
codex-rs/core/src/lib.rs
-1
View File
@@ -13,7 +13,6 @@ mod client;
mod client_common;
mod realtime_context;
mod realtime_conversation;
mod realtime_prompt;
pub(crate) mod session;
pub use session::SteerInputError;
mod codex_thread;
codex-rs/core/src/realtime_context.rs
+43 -468
View File
@@ -1,128 +1,35 @@
use crate::compact::content_items_to_text;
use crate::event_mapping::is_contextual_user_message_content;
use crate::session::session::Session;
use chrono::Utc;
use codex_exec_server::LOCAL_FS;
use codex_git_utils::resolve_root_git_project_for_trust;
use codex_protocol::models::ResponseItem;
use codex_realtime::RealtimeStartupContext;
use codex_realtime::RealtimeStartupContextTurn;
use codex_thread_store::ListThreadsParams;
use codex_thread_store::SortDirection;
use codex_thread_store::StoredThread;
use codex_thread_store::ThreadSortKey;
use codex_thread_store::ThreadStore;
use codex_utils_absolute_path::AbsolutePathBuf;
use codex_utils_output_truncation::TruncationPolicy;
use codex_utils_output_truncation::truncate_text;
use dirs::home_dir;
use std::cmp::Reverse;
use std::collections::HashMap;
use std::collections::HashSet;
use std::ffi::OsStr;
use std::fs::DirEntry;
use std::io;
use std::mem::take;
use std::path::Path;
use std::path::PathBuf;
use tracing::debug;
use tracing::info;
use tracing::warn;
const STARTUP_CONTEXT_HEADER: &str = "Startup context from Codex.\nThis is background context about recent work and machine/workspace layout. It may be incomplete or stale. Use it to inform responses, and do not repeat it back unless relevant.";
const STARTUP_CONTEXT_OPEN_TAG: &str = "<startup_context>";
const STARTUP_CONTEXT_CLOSE_TAG: &str = "</startup_context>";
const CURRENT_THREAD_SECTION_TOKEN_BUDGET: usize = 1_200;
const RECENT_WORK_SECTION_TOKEN_BUDGET: usize = 2_200;
const WORKSPACE_SECTION_TOKEN_BUDGET: usize = 1_600;
const NOTES_SECTION_TOKEN_BUDGET: usize = 300;
pub(crate) const REALTIME_TURN_TOKEN_BUDGET: usize = 300;
const MAX_RECENT_THREADS: usize = 40;
const MAX_RECENT_WORK_GROUPS: usize = 8;
const MAX_CURRENT_CWD_ASKS: usize = 8;
const MAX_OTHER_CWD_ASKS: usize = 5;
const MAX_ASK_CHARS: usize = 240;
const TREE_MAX_DEPTH: usize = 2;
const DIR_ENTRY_LIMIT: usize = 20;
const APPROX_BYTES_PER_TOKEN: usize = 4;
const NOISY_DIR_NAMES: &[&str] = &[
".git",
".next",
".pytest_cache",
".ruff_cache",
"__pycache__",
"build",
"dist",
"node_modules",
"out",
"target",
];
pub(crate) async fn build_realtime_startup_context(
sess: &Session,
budget_tokens: usize,
) -> Option<String> {
let config = sess.get_config().await;
let cwd = config.cwd.clone();
let history = sess.clone_history().await;
let current_thread_section = build_current_thread_section(history.raw_items());
let current_thread_turns = current_thread_turns(history.raw_items());
let recent_threads = load_recent_threads(sess).await;
let recent_work_section = build_recent_work_section(&cwd, &recent_threads).await;
let workspace_section = build_workspace_section_with_user_root(&cwd, home_dir()).await;
if current_thread_section.is_none()
&& recent_work_section.is_none()
&& workspace_section.is_none()
{
debug!("realtime startup context unavailable; skipping injection");
return None;
}
let mut parts = vec![STARTUP_CONTEXT_HEADER.to_string()];
let has_current_thread_section = current_thread_section.is_some();
let has_recent_work_section = recent_work_section.is_some();
let has_workspace_section = workspace_section.is_some();
if let Some(section) = format_section(
"Current Thread",
current_thread_section,
CURRENT_THREAD_SECTION_TOKEN_BUDGET,
) {
parts.push(section);
}
if let Some(section) = format_section(
"Recent Work",
recent_work_section,
RECENT_WORK_SECTION_TOKEN_BUDGET,
) {
parts.push(section);
}
if let Some(section) = format_section(
"Machine / Workspace Map",
workspace_section,
WORKSPACE_SECTION_TOKEN_BUDGET,
) {
parts.push(section);
}
if let Some(section) = format_section(
"Notes",
Some("Built at realtime startup from the current thread history, local thread metadata, and a bounded local workspace scan. This excludes repo memory instructions, AGENTS files, project-doc prompt blends, and memory summaries.".to_string()),
NOTES_SECTION_TOKEN_BUDGET,
) {
parts.push(section);
}
let context = format_startup_context_blob(&parts.join("\n\n"));
debug!(
approx_tokens = approx_token_count(&context),
requested_budget_tokens = budget_tokens,
bytes = context.len(),
has_current_thread_section,
has_recent_work_section,
has_workspace_section,
"built realtime startup context"
);
info!("realtime startup context: {context}");
Some(context)
let context = RealtimeStartupContext {
cwd: config.cwd.clone(),
current_thread_turns,
recent_threads,
user_root: home_dir(),
};
codex_realtime::build_realtime_startup_context(&context, budget_tokens).await
}
async fn load_recent_threads(sess: &Session) -> Vec<StoredThread> {
@@ -149,61 +56,7 @@ async fn load_recent_threads(sess: &Session) -> Vec<StoredThread> {
}
}
async fn build_recent_work_section(
cwd: &AbsolutePathBuf,
recent_threads: &[StoredThread],
) -> Option<String> {
let mut groups: HashMap<PathBuf, Vec<&StoredThread>> = HashMap::new();
for entry in recent_threads {
let group = match AbsolutePathBuf::from_absolute_path(entry.cwd.as_path()) {
Ok(entry_cwd) => resolve_root_git_project_for_trust(LOCAL_FS.as_ref(), &entry_cwd)
.await
.map(AbsolutePathBuf::into_path_buf)
.unwrap_or_else(|| entry.cwd.clone()),
Err(_) => entry.cwd.clone(),
};
groups.entry(group).or_default().push(entry);
}
let current_group = resolve_root_git_project_for_trust(LOCAL_FS.as_ref(), cwd)
.await
.map(AbsolutePathBuf::into_path_buf)
.unwrap_or_else(|| cwd.clone().into_path_buf());
let mut groups = groups.into_iter().collect::<Vec<_>>();
groups.sort_by(|(left_group, left_entries), (right_group, right_entries)| {
let left_latest = left_entries
.iter()
.map(|entry| entry.updated_at)
.max()
.unwrap_or_else(Utc::now);
let right_latest = right_entries
.iter()
.map(|entry| entry.updated_at)
.max()
.unwrap_or_else(Utc::now);
(
*left_group != current_group,
Reverse(left_latest),
left_group.as_os_str(),
)
.cmp(&(
*right_group != current_group,
Reverse(right_latest),
right_group.as_os_str(),
))
});
let mut sections = Vec::new();
for (group, mut entries) in groups.into_iter().take(MAX_RECENT_WORK_GROUPS) {
entries.sort_by_key(|entry| Reverse(entry.updated_at));
if let Some(section) = format_thread_group(&current_group, &group, entries).await {
sections.push(section);
}
}
(!sections.is_empty()).then(|| sections.join("\n\n"))
}
fn build_current_thread_section(items: &[ResponseItem]) -> Option<String> {
fn current_thread_turns(items: &[ResponseItem]) -> Vec<RealtimeStartupContextTurn> {
let mut turns = Vec::new();
let mut current_user = Vec::new();
let mut current_assistant = Vec::new();
@@ -221,7 +74,10 @@ fn build_current_thread_section(items: &[ResponseItem]) -> Option<String> {
continue;
};
if !current_user.is_empty() || !current_assistant.is_empty() {
turns.push((take(&mut current_user), take(&mut current_assistant)));
turns.push(RealtimeStartupContextTurn {
user_messages: take(&mut current_user),
assistant_messages: take(&mut current_assistant),
});
}
current_user.push(text);
}
@@ -242,321 +98,40 @@ fn build_current_thread_section(items: &[ResponseItem]) -> Option<String> {
}
if !current_user.is_empty() || !current_assistant.is_empty() {
turns.push((current_user, current_assistant));
turns.push(RealtimeStartupContextTurn {
user_messages: current_user,
assistant_messages: current_assistant,
});
}
if turns.is_empty() {
return None;
}
let mut lines = vec![
"Most recent user/assistant turns from this exact thread. Use them for continuity when responding.".to_string(),
];
let mut remaining_budget =
CURRENT_THREAD_SECTION_TOKEN_BUDGET.saturating_sub(approx_token_count(&lines.join("\n")));
let mut retained_turn_count = 0;
for (index, (user_messages, assistant_messages)) in turns.into_iter().rev().enumerate() {
if remaining_budget == 0 {
break;
}
let mut turn_lines = Vec::new();
if index == 0 {
turn_lines.push("### Latest turn".to_string());
} else {
turn_lines.push(format!("### Previous turn {index}"));
}
if !user_messages.is_empty() {
turn_lines.push("User:".to_string());
turn_lines.push(user_messages.join("\n\n"));
}
if !assistant_messages.is_empty() {
turn_lines.push(String::new());
turn_lines.push("Assistant:".to_string());
turn_lines.push(assistant_messages.join("\n\n"));
}
let turn_budget = REALTIME_TURN_TOKEN_BUDGET.min(remaining_budget);
let turn_text = turn_lines.join("\n");
let turn_text = truncate_realtime_text_to_token_budget(&turn_text, turn_budget);
let turn_tokens = approx_token_count(&turn_text);
if turn_tokens == 0 {
continue;
}
lines.push(String::new());
lines.push(turn_text);
remaining_budget = remaining_budget.saturating_sub(turn_tokens);
retained_turn_count += 1;
}
(retained_turn_count > 0).then(|| lines.join("\n"))
turns
}
pub(crate) fn truncate_realtime_text_to_token_budget(text: &str, budget_tokens: usize) -> String {
let mut truncation_budget = budget_tokens;
loop {
let candidate = truncate_text(text, TruncationPolicy::Tokens(truncation_budget));
let candidate_tokens = approx_token_count(&candidate);
if candidate_tokens <= budget_tokens {
break candidate;
}
// The shared truncator adds its marker after choosing preserved
// content, so tighten the content budget until the rendered turn
// itself fits the per-turn cap.
let excess_tokens = candidate_tokens.saturating_sub(budget_tokens);
let next_budget = truncation_budget.saturating_sub(excess_tokens.max(1));
if next_budget == 0 {
let candidate = truncate_text(text, TruncationPolicy::Tokens(0));
if approx_token_count(&candidate) <= budget_tokens {
break candidate;
}
break String::new();
}
truncation_budget = next_budget;
}
#[cfg(test)]
fn build_current_thread_section(items: &[ResponseItem]) -> Option<String> {
codex_realtime::build_current_thread_section(&current_thread_turns(items))
}
async fn build_workspace_section_with_user_root(
cwd: &AbsolutePathBuf,
user_root: Option<PathBuf>,
) -> Option<String> {
let cwd_path = cwd.as_path();
let git_root = resolve_root_git_project_for_trust(LOCAL_FS.as_ref(), cwd).await;
let cwd_tree = render_tree(cwd_path);
let git_root_tree = git_root
.as_ref()
.filter(|git_root| git_root.as_path() != cwd_path)
.and_then(|git_root| render_tree(git_root.as_path()));
let user_root_tree = user_root
.as_ref()
.filter(|user_root| user_root.as_path() != cwd_path)
.filter(|user_root| {
git_root
.as_ref()
.is_none_or(|git_root| git_root.as_path() != user_root.as_path())
})
.and_then(|user_root| render_tree(user_root));
if cwd_tree.is_none() && git_root.is_none() && user_root_tree.is_none() {
return None;
}
let mut lines = vec![
format!("Current working directory: {}", cwd_path.display()),
format!("Working directory name: {}", file_name_string(cwd_path)),
];
if let Some(git_root) = &git_root {
lines.push(format!("Git root: {}", git_root.display()));
lines.push(format!("Git project: {}", file_name_string(git_root)));
}
if let Some(user_root) = &user_root {
lines.push(format!("User root: {}", user_root.display()));
}
if let Some(tree) = cwd_tree {
lines.push(String::new());
lines.push("Working directory tree:".to_string());
lines.extend(tree);
}
if let Some(tree) = git_root_tree {
lines.push(String::new());
lines.push("Git root tree:".to_string());
lines.extend(tree);
}
if let Some(tree) = user_root_tree {
lines.push(String::new());
lines.push("User root tree:".to_string());
lines.extend(tree);
}
Some(lines.join("\n"))
}
fn render_tree(root: &Path) -> Option<Vec<String>> {
if !root.is_dir() {
return None;
}
let mut lines = Vec::new();
collect_tree_lines(root, /*depth*/ 0, &mut lines);
(!lines.is_empty()).then_some(lines)
}
fn collect_tree_lines(dir: &Path, depth: usize, lines: &mut Vec<String>) {
if depth >= TREE_MAX_DEPTH {
return;
}
let entries = match read_sorted_entries(dir) {
Ok(entries) => entries,
Err(_) => return,
};
let total_entries = entries.len();
for entry in entries.into_iter().take(DIR_ENTRY_LIMIT) {
let Ok(file_type) = entry.file_type() else {
continue;
};
let name = file_name_string(&entry.path());
let indent = " ".repeat(depth);
let suffix = if file_type.is_dir() { "/" } else { "" };
lines.push(format!("{indent}- {name}{suffix}"));
if file_type.is_dir() {
collect_tree_lines(&entry.path(), depth + 1, lines);
}
}
if total_entries > DIR_ENTRY_LIMIT {
lines.push(format!(
"{}- ... {} more entries",
" ".repeat(depth),
total_entries - DIR_ENTRY_LIMIT
));
}
}
fn read_sorted_entries(dir: &Path) -> io::Result<Vec<DirEntry>> {
let mut entries = std::fs::read_dir(dir)?
.filter_map(Result::ok)
.filter(|entry| !is_noisy_name(&entry.file_name()))
.collect::<Vec<_>>();
entries.sort_by(|left, right| {
let left_is_dir = left
.file_type()
.map(|file_type| file_type.is_dir())
.unwrap_or(false);
let right_is_dir = right
.file_type()
.map(|file_type| file_type.is_dir())
.unwrap_or(false);
(!left_is_dir, file_name_string(&left.path()))
.cmp(&(!right_is_dir, file_name_string(&right.path())))
});
Ok(entries)
}
fn is_noisy_name(name: &OsStr) -> bool {
let name = name.to_string_lossy();
name.starts_with('.') || NOISY_DIR_NAMES.iter().any(|noisy| *noisy == name)
}
fn format_section(title: &str, body: Option<String>, budget_tokens: usize) -> Option<String> {
let body = body?;
let body = body.trim();
if body.is_empty() {
return None;
}
let heading = format!("## {title}\n");
let body_budget = budget_tokens.saturating_sub(approx_token_count(&heading));
if body_budget == 0 {
return None;
}
let body = truncate_realtime_text_to_token_budget(body, body_budget);
if body.is_empty() {
return None;
}
Some(format!("{heading}{body}"))
}
fn format_startup_context_blob(body: &str) -> String {
format!("{STARTUP_CONTEXT_OPEN_TAG}\n{body}\n{STARTUP_CONTEXT_CLOSE_TAG}")
}
async fn format_thread_group(
current_group: &Path,
group: &Path,
entries: Vec<&StoredThread>,
) -> Option<String> {
let latest = entries.first()?;
let group_label =
if let Ok(latest_cwd) = AbsolutePathBuf::from_absolute_path(latest.cwd.as_path()) {
if resolve_root_git_project_for_trust(LOCAL_FS.as_ref(), &latest_cwd)
.await
.is_some()
{
format!("### Git repo: {}", group.display())
} else {
format!("### Directory: {}", group.display())
}
} else {
format!("### Directory: {}", group.display())
};
let mut lines = vec![
group_label,
format!("Recent sessions: {}", entries.len()),
format!("Latest activity: {}", latest.updated_at.to_rfc3339()),
];
if let Some(git_branch) = latest
.git_info
.as_ref()
.and_then(|git| git.branch.as_deref())
.filter(|git_branch| !git_branch.is_empty())
{
lines.push(format!("Latest branch: {git_branch}"));
}
lines.push(String::new());
lines.push("User asks:".to_string());
let mut seen = HashSet::new();
let max_asks = if group == current_group {
MAX_CURRENT_CWD_ASKS
} else {
MAX_OTHER_CWD_ASKS
};
for entry in entries {
let Some(first_user_message) = entry.first_user_message.as_deref() else {
continue;
};
let ask = first_user_message
.split_whitespace()
.collect::<Vec<_>>()
.join(" ");
let dedupe_key = format!("{}:{ask}", entry.cwd.display());
if ask.is_empty() || !seen.insert(dedupe_key) {
continue;
}
let ask = if ask.chars().count() > MAX_ASK_CHARS {
format!(
"{}...",
ask.chars()
.take(MAX_ASK_CHARS.saturating_sub(3))
.collect::<String>()
)
} else {
ask
};
lines.push(format!("- {}: {ask}", entry.cwd.display()));
if seen.len() == max_asks {
break;
}
}
(lines.len() > 5).then(|| lines.join("\n"))
}
fn file_name_string(path: &Path) -> String {
path.file_name()
.and_then(OsStr::to_str)
.map(str::to_owned)
.unwrap_or_else(|| path.display().to_string())
}
fn approx_token_count(text: &str) -> usize {
text.len().div_ceil(APPROX_BYTES_PER_TOKEN)
}
#[cfg(test)]
pub(crate) use codex_realtime::CURRENT_THREAD_SECTION_TOKEN_BUDGET;
#[cfg(test)]
pub(crate) use codex_realtime::NOTES_SECTION_TOKEN_BUDGET;
pub(crate) use codex_realtime::REALTIME_TURN_TOKEN_BUDGET;
#[cfg(test)]
pub(crate) use codex_realtime::RECENT_WORK_SECTION_TOKEN_BUDGET;
#[cfg(test)]
pub(crate) use codex_realtime::STARTUP_CONTEXT_HEADER;
#[cfg(test)]
pub(crate) use codex_realtime::WORKSPACE_SECTION_TOKEN_BUDGET;
#[cfg(test)]
use codex_realtime::build_recent_work_section;
#[cfg(test)]
use codex_realtime::build_workspace_section_with_user_root;
#[cfg(test)]
use codex_realtime::format_section;
#[cfg(test)]
use codex_realtime::format_startup_context_blob;
pub(crate) use codex_realtime::truncate_realtime_text_to_token_budget;
#[cfg(test)]
#[path = "realtime_context_tests.rs"]
codex-rs/core/src/realtime_conversation.rs
+89 -1128
View File
File diff suppressed because it is too large Load Diff
codex-rs/realtime/BUILD.bazel
+7
View File
@@ -0,0 +1,7 @@
load("//:defs.bzl", "codex_rust_crate")
codex_rust_crate(
name = "realtime",
crate_name = "codex_realtime",
compile_data = glob(["templates/**/*.md"]),
)
codex-rs/realtime/Cargo.toml
+37
View File
@@ -0,0 +1,37 @@
[package]
name = "codex-realtime"
version.workspace = true
edition.workspace = true
license.workspace = true
[lib]
name = "codex_realtime"
path = "src/lib.rs"
[dependencies]
anyhow = { workspace = true }
async-channel = { workspace = true }
base64 = { workspace = true }
chrono = { workspace = true }
codex-api = { workspace = true }
codex-config = { workspace = true }
codex-exec-server = { workspace = true }
codex-git-utils = { workspace = true }
codex-login = { workspace = true }
codex-model-provider-info = { workspace = true }
codex-protocol = { workspace = true }
codex-thread-store = { workspace = true }
codex-utils-absolute-path = { workspace = true }
codex-utils-output-truncation = { workspace = true }
http = { workspace = true }
serde_json = { workspace = true }
tokio = { workspace = true, features = ["rt-multi-thread"] }
tracing = { workspace = true }
whoami = { workspace = true }
[dev-dependencies]
pretty_assertions = { workspace = true }
tempfile = { workspace = true }
[lints]
workspace = true
codex-rs/realtime/src/config.rs
+13
View File
@@ -0,0 +1,13 @@
use codex_config::config_toml::RealtimeAudioConfig;
use codex_config::config_toml::RealtimeConfig;
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct RealtimeFeaturesConfig {
pub audio: RealtimeAudioConfig,
pub session: RealtimeConfig,
pub websocket_base_url: Option<String>,
pub websocket_model: Option<String>,
pub websocket_backend_prompt: Option<String>,
pub websocket_startup_context: Option<String>,
pub start_instructions: Option<String>,
}
codex-rs/realtime/src/conversation_manager.rs
+830
View File
@@ -0,0 +1,830 @@
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;
use async_channel::Sender;
use async_channel::TrySendError;
use base64::Engine;
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
use codex_api::ApiError;
use codex_api::RealtimeAudioFrame;
use codex_api::RealtimeEvent;
use codex_api::RealtimeEventParser;
use codex_api::RealtimeWebsocketEvents;
use codex_api::RealtimeWebsocketWriter;
use codex_api::map_api_error;
use codex_protocol::error::CodexErr;
use codex_protocol::error::Result as CodexResult;
use serde_json::json;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use tokio::sync::Mutex;
use tokio::task::JoinHandle;
use tracing::debug;
use tracing::error;
use tracing::warn;
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;
const REALTIME_V2_HANDOFF_COMPLETE_ACKNOWLEDGEMENT: &str =
"Background agent finished. Use the preceding [BACKEND] messages as the result.";
const REALTIME_V2_STEER_ACKNOWLEDGEMENT: &str =
"This was sent to steer the previous background agent task.";
const REALTIME_ACTIVE_RESPONSE_ERROR_PREFIX: &str =
"Conversation already has an active response in progress:";
enum RealtimeFanoutTaskStop {
Abort,
Detach,
}
pub struct RealtimeConversationManager {
state: Mutex<Option<ConversationState>>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum RealtimeSessionKind {
V1,
V2,
}
#[derive(Clone, Debug)]
struct RealtimeHandoffState {
output_tx: Sender<HandoffOutput>,
active_handoff: Arc<Mutex<Option<String>>>,
last_output_text: Arc<Mutex<Option<String>>>,
session_kind: RealtimeSessionKind,
}
#[derive(Debug, PartialEq, Eq)]
enum HandoffOutput {
ProgressUpdate {
handoff_id: String,
output_text: String,
},
FinalUpdate {
handoff_id: String,
output_text: String,
},
}
#[derive(Debug, PartialEq, Eq)]
struct OutputAudioState {
item_id: String,
audio_end_ms: u32,
}
#[derive(Default)]
struct RealtimeResponseCreateQueue {
active_default_response: bool,
pending_create: bool,
}
impl RealtimeResponseCreateQueue {
async fn request_create(
&mut self,
writer: &RealtimeWebsocketWriter,
events_tx: &Sender<RealtimeEvent>,
reason: &str,
) -> anyhow::Result<()> {
if self.active_default_response {
self.pending_create = true;
return Ok(());
}
self.send_create_now(writer, events_tx, reason).await
}
fn mark_started(&mut self) {
self.active_default_response = true;
}
async fn mark_finished(
&mut self,
writer: &RealtimeWebsocketWriter,
events_tx: &Sender<RealtimeEvent>,
reason: &str,
) -> anyhow::Result<()> {
self.active_default_response = false;
if !self.pending_create {
return Ok(());
}
self.pending_create = false;
self.send_create_now(writer, events_tx, reason).await
}
async fn send_create_now(
&mut self,
writer: &RealtimeWebsocketWriter,
events_tx: &Sender<RealtimeEvent>,
reason: &str,
) -> anyhow::Result<()> {
if let Err(err) = writer.send_response_create().await {
let mapped_error = map_api_error(err);
let error_message = mapped_error.to_string();
if error_message.starts_with(REALTIME_ACTIVE_RESPONSE_ERROR_PREFIX) {
warn!("realtime response.create raced an active response; deferring");
self.active_default_response = true;
self.pending_create = true;
return Ok(());
}
warn!("failed to send {reason} response.create: {mapped_error}");
let _ = events_tx.send(RealtimeEvent::Error(error_message)).await;
return Err(mapped_error.into());
}
self.active_default_response = true;
Ok(())
}
}
struct RealtimeInputTask {
writer: RealtimeWebsocketWriter,
events: RealtimeWebsocketEvents,
user_text_rx: Receiver<String>,
handoff_output_rx: Receiver<HandoffOutput>,
audio_rx: Receiver<RealtimeAudioFrame>,
events_tx: Sender<RealtimeEvent>,
handoff_state: RealtimeHandoffState,
session_kind: RealtimeSessionKind,
event_parser: RealtimeEventParser,
}
impl RealtimeHandoffState {
fn new(output_tx: Sender<HandoffOutput>, session_kind: RealtimeSessionKind) -> Self {
Self {
output_tx,
active_handoff: Arc::new(Mutex::new(None)),
last_output_text: Arc::new(Mutex::new(None)),
session_kind,
}
}
}
struct ConversationState {
audio_tx: Sender<RealtimeAudioFrame>,
user_text_tx: Sender<String>,
session_kind: RealtimeSessionKind,
handoff: RealtimeHandoffState,
input_task: JoinHandle<()>,
fanout_task: Option<JoinHandle<()>>,
realtime_active: Arc<AtomicBool>,
}
pub struct RealtimeStart {
pub writer: RealtimeWebsocketWriter,
pub events: RealtimeWebsocketEvents,
pub event_parser: RealtimeEventParser,
pub sdp: Option<String>,
}
pub struct RealtimeStartOutput {
pub realtime_active: Arc<AtomicBool>,
pub events_rx: Receiver<RealtimeEvent>,
pub sdp: Option<String>,
}
impl RealtimeConversationManager {
pub fn new() -> Self {
Self {
state: Mutex::new(None),
}
}
pub async fn running_state(&self) -> Option<()> {
let state = self.state.lock().await;
state
.as_ref()
.and_then(|state| state.realtime_active.load(Ordering::Relaxed).then_some(()))
}
pub async fn is_running_v2(&self) -> bool {
let state = self.state.lock().await;
matches!(
state.as_ref(),
Some(state)
if state.realtime_active.load(Ordering::Relaxed)
&& state.session_kind == RealtimeSessionKind::V2
)
}
pub async fn start(&self, start: RealtimeStart) -> CodexResult<RealtimeStartOutput> {
let previous_state = {
let mut guard = self.state.lock().await;
guard.take()
};
if let Some(state) = previous_state {
stop_conversation_state(state, RealtimeFanoutTaskStop::Abort).await;
}
self.start_inner(start).await
}
async fn start_inner(&self, start: RealtimeStart) -> CodexResult<RealtimeStartOutput> {
let RealtimeStart {
writer,
events,
event_parser,
sdp,
} = start;
let session_kind = match event_parser {
RealtimeEventParser::V1 => RealtimeSessionKind::V1,
RealtimeEventParser::RealtimeV2 => RealtimeSessionKind::V2,
};
let (audio_tx, audio_rx) =
async_channel::bounded::<RealtimeAudioFrame>(AUDIO_IN_QUEUE_CAPACITY);
let (user_text_tx, user_text_rx) =
async_channel::bounded::<String>(USER_TEXT_IN_QUEUE_CAPACITY);
let (handoff_output_tx, handoff_output_rx) =
async_channel::bounded::<HandoffOutput>(HANDOFF_OUT_QUEUE_CAPACITY);
let (events_tx, events_rx) =
async_channel::bounded::<RealtimeEvent>(OUTPUT_EVENTS_QUEUE_CAPACITY);
let realtime_active = Arc::new(AtomicBool::new(true));
let handoff = RealtimeHandoffState::new(handoff_output_tx, session_kind);
let task = spawn_realtime_input_task(RealtimeInputTask {
writer: writer.clone(),
events,
user_text_rx,
handoff_output_rx,
audio_rx,
events_tx,
handoff_state: handoff.clone(),
session_kind,
event_parser,
});
let mut guard = self.state.lock().await;
*guard = Some(ConversationState {
audio_tx,
user_text_tx,
session_kind,
handoff,
input_task: task,
fanout_task: None,
realtime_active: Arc::clone(&realtime_active),
});
Ok(RealtimeStartOutput {
realtime_active,
events_rx,
sdp,
})
}
pub async fn register_fanout_task(
&self,
realtime_active: &Arc<AtomicBool>,
fanout_task: JoinHandle<()>,
) {
let mut fanout_task = Some(fanout_task);
{
let mut guard = self.state.lock().await;
if let Some(state) = guard.as_mut()
&& Arc::ptr_eq(&state.realtime_active, realtime_active)
{
state.fanout_task = fanout_task.take();
}
}
if let Some(fanout_task) = fanout_task {
fanout_task.abort();
let _ = fanout_task.await;
}
}
pub async fn finish_if_active(&self, realtime_active: &Arc<AtomicBool>) {
let state = {
let mut guard = self.state.lock().await;
match guard.as_ref() {
Some(state) if Arc::ptr_eq(&state.realtime_active, realtime_active) => guard.take(),
_ => None,
}
};
if let Some(state) = state {
stop_conversation_state(state, RealtimeFanoutTaskStop::Detach).await;
}
}
pub async fn audio_in(&self, frame: RealtimeAudioFrame) -> CodexResult<()> {
let sender = {
let guard = self.state.lock().await;
guard.as_ref().map(|state| state.audio_tx.clone())
};
let Some(sender) = sender else {
return Err(CodexErr::InvalidRequest(
"conversation is not running".to_string(),
));
};
match sender.try_send(frame) {
Ok(()) => Ok(()),
Err(TrySendError::Full(_)) => {
warn!("dropping input audio frame due to full queue");
Ok(())
}
Err(TrySendError::Closed(_)) => Err(CodexErr::InvalidRequest(
"conversation is not running".to_string(),
)),
}
}
pub async fn text_in(&self, text: String) -> CodexResult<()> {
let sender = {
let guard = self.state.lock().await;
guard
.as_ref()
.map(|state| (state.user_text_tx.clone(), state.session_kind))
};
let Some((sender, session_kind)) = sender else {
return Err(CodexErr::InvalidRequest(
"conversation is not running".to_string(),
));
};
let text = if session_kind == RealtimeSessionKind::V2 {
prefix_realtime_v2_text(text, REALTIME_USER_TEXT_PREFIX)
} else {
text
};
sender
.send(text)
.await
.map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))?;
Ok(())
}
pub async fn handoff_out(&self, output_text: String) -> CodexResult<()> {
let handoff = {
let guard = self.state.lock().await;
let Some(state) = guard.as_ref() else {
return Err(CodexErr::InvalidRequest(
"conversation is not running".to_string(),
));
};
state.handoff.clone()
};
let Some(handoff_id) = handoff.active_handoff.lock().await.clone() else {
return Ok(());
};
let output_text = if handoff.session_kind == RealtimeSessionKind::V2 {
prefix_realtime_v2_text(output_text, REALTIME_BACKEND_TEXT_PREFIX)
} else {
output_text
};
*handoff.last_output_text.lock().await = Some(output_text.clone());
handoff
.output_tx
.send(HandoffOutput::ProgressUpdate {
handoff_id,
output_text,
})
.await
.map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))?;
Ok(())
}
pub async fn handoff_complete(&self) -> CodexResult<()> {
let handoff = {
let guard = self.state.lock().await;
guard.as_ref().map(|state| state.handoff.clone())
};
let Some(handoff) = handoff else {
return Ok(());
};
match handoff.session_kind {
RealtimeSessionKind::V1 => return Ok(()),
RealtimeSessionKind::V2 => {}
}
let Some(handoff_id) = handoff.active_handoff.lock().await.clone() else {
return Ok(());
};
let Some(output_text) = handoff.last_output_text.lock().await.clone() else {
return Ok(());
};
handoff
.output_tx
.send(HandoffOutput::FinalUpdate {
handoff_id,
output_text,
})
.await
.map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))
}
pub async fn active_handoff_id(&self) -> Option<String> {
let handoff = {
let guard = self.state.lock().await;
guard.as_ref().map(|state| state.handoff.clone())
}?;
handoff.active_handoff.lock().await.clone()
}
pub async fn clear_active_handoff(&self) {
let handoff = {
let guard = self.state.lock().await;
guard.as_ref().map(|state| state.handoff.clone())
};
if let Some(handoff) = handoff {
*handoff.active_handoff.lock().await = None;
*handoff.last_output_text.lock().await = None;
}
}
pub async fn shutdown(&self) -> CodexResult<()> {
let state = {
let mut guard = self.state.lock().await;
guard.take()
};
if let Some(state) = state {
stop_conversation_state(state, RealtimeFanoutTaskStop::Abort).await;
}
Ok(())
}
}
async fn stop_conversation_state(
mut state: ConversationState,
fanout_task_stop: RealtimeFanoutTaskStop,
) {
state.realtime_active.store(false, Ordering::Relaxed);
state.input_task.abort();
let _ = state.input_task.await;
if let Some(fanout_task) = state.fanout_task.take() {
match fanout_task_stop {
RealtimeFanoutTaskStop::Abort => {
fanout_task.abort();
let _ = fanout_task.await;
}
RealtimeFanoutTaskStop::Detach => {}
}
}
}
fn spawn_realtime_input_task(input: RealtimeInputTask) -> JoinHandle<()> {
let RealtimeInputTask {
writer,
events,
user_text_rx,
handoff_output_rx,
audio_rx,
events_tx,
handoff_state,
session_kind,
event_parser,
} = input;
tokio::spawn(async move {
let mut output_audio_state: Option<OutputAudioState> = None;
let mut response_create_queue = RealtimeResponseCreateQueue::default();
loop {
let result = tokio::select! {
user_text = user_text_rx.recv() => {
handle_user_text_input(user_text, &writer, &events_tx).await
}
background_agent_output = handoff_output_rx.recv() => {
handle_handoff_output(
background_agent_output,
&writer,
&events_tx,
&handoff_state,
event_parser,
&mut response_create_queue,
)
.await
}
realtime_event = events.next_event() => {
handle_realtime_server_event(
realtime_event,
&writer,
&events_tx,
&handoff_state,
session_kind,
&mut output_audio_state,
&mut response_create_queue,
)
.await
}
user_audio_frame = audio_rx.recv() => {
handle_user_audio_input(user_audio_frame, &writer, &events_tx).await
}
};
if result.is_err() {
break;
}
}
})
}
async fn handle_user_text_input(
text: Result<String, RecvError>,
writer: &RealtimeWebsocketWriter,
events_tx: &Sender<RealtimeEvent>,
) -> anyhow::Result<()> {
let text = text.context("user text input channel closed")?;
if let Err(err) = writer.send_conversation_item_create(text).await {
let mapped_error = map_api_error(err);
warn!("failed to send input text: {mapped_error}");
let _ = events_tx
.send(RealtimeEvent::Error(mapped_error.to_string()))
.await;
return Err(mapped_error.into());
}
Ok(())
}
async fn handle_handoff_output(
handoff_output: Result<HandoffOutput, RecvError>,
writer: &RealtimeWebsocketWriter,
events_tx: &Sender<RealtimeEvent>,
handoff_state: &RealtimeHandoffState,
event_parser: RealtimeEventParser,
response_create_queue: &mut RealtimeResponseCreateQueue,
) -> anyhow::Result<()> {
let handoff_output = handoff_output.context("handoff output channel closed")?;
let result = match event_parser {
RealtimeEventParser::V1 => match handoff_output {
HandoffOutput::ProgressUpdate {
handoff_id,
output_text,
}
| HandoffOutput::FinalUpdate {
handoff_id,
output_text,
} => {
writer
.send_conversation_function_call_output(handoff_id, output_text)
.await
}
},
RealtimeEventParser::RealtimeV2 => match handoff_output {
HandoffOutput::ProgressUpdate {
handoff_id,
output_text,
} => {
let active_handoff = handoff_state.active_handoff.lock().await.clone();
match active_handoff {
Some(active_handoff) if active_handoff == handoff_id => {}
Some(_) | None => {
debug!("dropping stale realtime handoff progress update");
return Ok(());
}
}
writer.send_conversation_item_create(output_text).await
}
HandoffOutput::FinalUpdate {
handoff_id,
output_text: _,
} => {
if let Err(err) = writer
.send_conversation_function_call_output(
handoff_id,
REALTIME_V2_HANDOFF_COMPLETE_ACKNOWLEDGEMENT.to_string(),
)
.await
{
Err(err)
} else {
return response_create_queue
.request_create(writer, events_tx, "handoff")
.await;
}
}
},
};
if let Err(err) = result {
let mapped_error = map_api_error(err);
warn!("failed to send handoff output: {mapped_error}");
let _ = events_tx
.send(RealtimeEvent::Error(mapped_error.to_string()))
.await;
return Err(mapped_error.into());
}
Ok(())
}
async fn handle_realtime_server_event(
event: Result<Option<RealtimeEvent>, ApiError>,
writer: &RealtimeWebsocketWriter,
events_tx: &Sender<RealtimeEvent>,
handoff_state: &RealtimeHandoffState,
session_kind: RealtimeSessionKind,
output_audio_state: &mut Option<OutputAudioState>,
response_create_queue: &mut RealtimeResponseCreateQueue,
) -> anyhow::Result<()> {
let event = match event {
Ok(Some(event)) => event,
Ok(None) => anyhow::bail!("realtime event stream ended"),
Err(err) => {
let mapped_error = map_api_error(err);
if events_tx
.send(RealtimeEvent::Error(mapped_error.to_string()))
.await
.is_err()
{
return Err(mapped_error.into());
}
error!("realtime stream closed: {mapped_error}");
return Err(mapped_error.into());
}
};
let should_stop = match &event {
RealtimeEvent::AudioOut(frame) => {
if session_kind == RealtimeSessionKind::V2 {
update_output_audio_state(output_audio_state, frame);
}
false
}
RealtimeEvent::InputAudioSpeechStarted(event) => {
if session_kind == RealtimeSessionKind::V2
&& let Some(output_audio_state) = output_audio_state.take()
&& event
.item_id
.as_deref()
.is_none_or(|item_id| item_id == output_audio_state.item_id)
&& let Err(err) = writer
.send_payload(
json!({
"type": "conversation.item.truncate",
"item_id": output_audio_state.item_id,
"content_index": 0,
"audio_end_ms": output_audio_state.audio_end_ms,
})
.to_string(),
)
.await
{
let mapped_error = map_api_error(err);
warn!("failed to truncate realtime audio: {mapped_error}");
}
false
}
RealtimeEvent::ResponseCreated(_) => {
if session_kind == RealtimeSessionKind::V2 {
response_create_queue.mark_started();
}
false
}
RealtimeEvent::ResponseCancelled(_) | RealtimeEvent::ResponseDone(_) => {
*output_audio_state = None;
if session_kind == RealtimeSessionKind::V2 {
response_create_queue
.mark_finished(writer, events_tx, "deferred")
.await?;
}
false
}
RealtimeEvent::HandoffRequested(handoff) => {
*output_audio_state = None;
match session_kind {
RealtimeSessionKind::V1 => {
*handoff_state.last_output_text.lock().await = None;
*handoff_state.active_handoff.lock().await = Some(handoff.handoff_id.clone());
}
RealtimeSessionKind::V2 => {
let active_handoff = handoff_state.active_handoff.lock().await.clone();
match active_handoff {
Some(_) => {
if let Err(err) = writer
.send_conversation_function_call_output(
handoff.handoff_id.clone(),
REALTIME_V2_STEER_ACKNOWLEDGEMENT.to_string(),
)
.await
{
let mapped_error = map_api_error(err);
warn!(
"failed to send handoff steering acknowledgement: {mapped_error}"
);
let _ = events_tx
.send(RealtimeEvent::Error(mapped_error.to_string()))
.await;
return Err(mapped_error.into());
}
response_create_queue
.request_create(writer, events_tx, "handoff steering")
.await?;
}
None => {
*handoff_state.last_output_text.lock().await = None;
*handoff_state.active_handoff.lock().await =
Some(handoff.handoff_id.clone());
}
}
}
}
false
}
RealtimeEvent::NoopRequested(noop) => {
*output_audio_state = None;
if session_kind == RealtimeSessionKind::V2
&& let Err(err) = writer
.send_conversation_function_call_output(noop.call_id.clone(), String::new())
.await
{
let mapped_error = map_api_error(err);
warn!("failed to send realtime noop function output: {mapped_error}");
let _ = events_tx
.send(RealtimeEvent::Error(mapped_error.to_string()))
.await;
return Err(mapped_error.into());
}
false
}
RealtimeEvent::Error(_) => true,
RealtimeEvent::SessionUpdated { .. }
| RealtimeEvent::InputTranscriptDelta(_)
| RealtimeEvent::InputTranscriptDone(_)
| RealtimeEvent::OutputTranscriptDelta(_)
| RealtimeEvent::OutputTranscriptDone(_)
| RealtimeEvent::ConversationItemAdded(_)
| RealtimeEvent::ConversationItemDone { .. } => false,
};
if events_tx.send(event).await.is_err() {
anyhow::bail!("realtime output event channel closed");
}
if should_stop {
error!("realtime stream error event received");
anyhow::bail!("realtime stream error event received");
}
Ok(())
}
async fn handle_user_audio_input(
frame: Result<RealtimeAudioFrame, RecvError>,
writer: &RealtimeWebsocketWriter,
events_tx: &Sender<RealtimeEvent>,
) -> anyhow::Result<()> {
let frame = frame.context("user audio input channel closed")?;
if let Err(err) = writer.send_audio_frame(frame).await {
let mapped_error = map_api_error(err);
error!("failed to send input audio: {mapped_error}");
let _ = events_tx
.send(RealtimeEvent::Error(mapped_error.to_string()))
.await;
return Err(mapped_error.into());
}
Ok(())
}
fn update_output_audio_state(
output_audio_state: &mut Option<OutputAudioState>,
frame: &RealtimeAudioFrame,
) {
let Some(item_id) = frame.item_id.clone() else {
return;
};
let audio_end_ms = audio_duration_ms(frame);
if audio_end_ms == 0 {
return;
}
if let Some(current) = output_audio_state.as_mut()
&& current.item_id == item_id
{
current.audio_end_ms = current.audio_end_ms.saturating_add(audio_end_ms);
return;
}
*output_audio_state = Some(OutputAudioState {
item_id,
audio_end_ms,
});
}
fn audio_duration_ms(frame: &RealtimeAudioFrame) -> u32 {
let Some(samples_per_channel) = frame
.samples_per_channel
.or(decoded_samples_per_channel(frame))
else {
return 0;
};
let sample_rate = u64::from(frame.sample_rate.max(1));
((u64::from(samples_per_channel) * 1_000) / sample_rate) as u32
}
fn decoded_samples_per_channel(frame: &RealtimeAudioFrame) -> Option<u32> {
let bytes = BASE64_STANDARD.decode(&frame.data).ok()?;
let channels = usize::from(frame.num_channels.max(1));
let samples = bytes.len().checked_div(2)?.checked_div(channels)?;
u32::try_from(samples).ok()
}
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
+38
View File
@@ -0,0 +1,38 @@
mod config;
mod conversation_manager;
mod handoff;
mod prompt;
mod session_config;
mod startup_context;
pub use config::RealtimeFeaturesConfig;
pub use conversation_manager::RealtimeConversationManager;
pub use conversation_manager::RealtimeStart;
pub use conversation_manager::RealtimeStartOutput;
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 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;
pub use startup_context::RECENT_WORK_SECTION_TOKEN_BUDGET;
pub use startup_context::RealtimeStartupContext;
pub use startup_context::RealtimeStartupContextTurn;
pub use startup_context::STARTUP_CONTEXT_HEADER;
pub use startup_context::WORKSPACE_SECTION_TOKEN_BUDGET;
pub use startup_context::build_current_thread_section;
pub use startup_context::build_realtime_startup_context;
pub use startup_context::build_recent_work_section;
pub use startup_context::build_workspace_section_with_user_root;
pub use startup_context::format_section;
pub use startup_context::format_startup_context_blob;
pub use startup_context::truncate_realtime_text_to_token_budget;
codex-rs/realtime/src/prompt.rs
+82
View File
@@ -0,0 +1,82 @@
const BACKEND_PROMPT: &str = include_str!("../templates/realtime/backend_prompt.md");
const DEFAULT_USER_FIRST_NAME: &str = "there";
const USER_FIRST_NAME_PLACEHOLDER: &str = "{{ user_first_name }}";
pub fn prepare_realtime_backend_prompt(
prompt: Option<Option<String>>,
config_prompt: Option<String>,
) -> String {
if let Some(config_prompt) = config_prompt
&& !config_prompt.trim().is_empty()
{
return config_prompt;
}
match prompt {
Some(Some(prompt)) => return prompt,
Some(None) => return String::new(),
None => {}
}
BACKEND_PROMPT
.trim_end()
.replace(USER_FIRST_NAME_PLACEHOLDER, &current_user_first_name())
}
fn current_user_first_name() -> String {
[whoami::realname(), whoami::username()]
.into_iter()
.filter_map(|name| name.split_whitespace().next().map(str::to_string))
.find(|name| !name.is_empty())
.unwrap_or_else(|| DEFAULT_USER_FIRST_NAME.to_string())
}
#[cfg(test)]
mod tests {
use super::prepare_realtime_backend_prompt;
#[test]
fn prepare_realtime_backend_prompt_prefers_config_override() {
assert_eq!(
prepare_realtime_backend_prompt(
Some(Some("prompt from request".to_string())),
Some("prompt from config".to_string()),
),
"prompt from config"
);
}
#[test]
fn prepare_realtime_backend_prompt_uses_request_prompt() {
assert_eq!(
prepare_realtime_backend_prompt(
Some(Some("prompt from request".to_string())),
/*config_prompt*/ None,
),
"prompt from request"
);
}
#[test]
fn prepare_realtime_backend_prompt_preserves_empty_request_prompt() {
assert_eq!(
prepare_realtime_backend_prompt(Some(Some(String::new())), /*config_prompt*/ None),
""
);
assert_eq!(
prepare_realtime_backend_prompt(Some(None), /*config_prompt*/ None),
""
);
}
#[test]
fn prepare_realtime_backend_prompt_renders_default() {
let prompt =
prepare_realtime_backend_prompt(/*prompt*/ None, /*config_prompt*/ None);
assert!(prompt.starts_with("## Identity, tone, and role"));
assert!(prompt.contains("You are Codex, an OpenAI general-purpose agentic assistant"));
assert!(prompt.contains("The user's name is "));
assert!(!prompt.contains("{{ user_first_name }}"));
}
}
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))
}
codex-rs/realtime/src/startup_context.rs
+491
View File
@@ -0,0 +1,491 @@
use chrono::Utc;
use codex_exec_server::LOCAL_FS;
use codex_git_utils::resolve_root_git_project_for_trust;
use codex_thread_store::StoredThread;
use codex_utils_absolute_path::AbsolutePathBuf;
use codex_utils_output_truncation::TruncationPolicy;
use codex_utils_output_truncation::truncate_text;
use std::cmp::Reverse;
use std::collections::HashMap;
use std::collections::HashSet;
use std::ffi::OsStr;
use std::fs::DirEntry;
use std::io;
use std::path::Path;
use std::path::PathBuf;
use tracing::debug;
use tracing::info;
pub const STARTUP_CONTEXT_HEADER: &str = "Startup context from Codex.\nThis is background context about recent work and machine/workspace layout. It may be incomplete or stale. Use it to inform responses, and do not repeat it back unless relevant.";
const STARTUP_CONTEXT_OPEN_TAG: &str = "<startup_context>";
const STARTUP_CONTEXT_CLOSE_TAG: &str = "</startup_context>";
pub const CURRENT_THREAD_SECTION_TOKEN_BUDGET: usize = 1_200;
pub const RECENT_WORK_SECTION_TOKEN_BUDGET: usize = 2_200;
pub const WORKSPACE_SECTION_TOKEN_BUDGET: usize = 1_600;
pub const NOTES_SECTION_TOKEN_BUDGET: usize = 300;
pub const REALTIME_TURN_TOKEN_BUDGET: usize = 300;
const MAX_RECENT_WORK_GROUPS: usize = 8;
const MAX_CURRENT_CWD_ASKS: usize = 8;
const MAX_OTHER_CWD_ASKS: usize = 5;
const MAX_ASK_CHARS: usize = 240;
const TREE_MAX_DEPTH: usize = 2;
const DIR_ENTRY_LIMIT: usize = 20;
const APPROX_BYTES_PER_TOKEN: usize = 4;
const NOISY_DIR_NAMES: &[&str] = &[
".git",
".next",
".pytest_cache",
".ruff_cache",
"__pycache__",
"build",
"dist",
"node_modules",
"out",
"target",
];
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RealtimeStartupContextTurn {
pub user_messages: Vec<String>,
pub assistant_messages: Vec<String>,
}
#[derive(Clone, Debug)]
pub struct RealtimeStartupContext {
pub cwd: AbsolutePathBuf,
pub current_thread_turns: Vec<RealtimeStartupContextTurn>,
pub recent_threads: Vec<StoredThread>,
pub user_root: Option<PathBuf>,
}
pub async fn build_realtime_startup_context(
context: &RealtimeStartupContext,
budget_tokens: usize,
) -> Option<String> {
let current_thread_section = build_current_thread_section(&context.current_thread_turns);
let recent_work_section =
build_recent_work_section(&context.cwd, &context.recent_threads).await;
let workspace_section =
build_workspace_section_with_user_root(&context.cwd, context.user_root.clone()).await;
if current_thread_section.is_none()
&& recent_work_section.is_none()
&& workspace_section.is_none()
{
debug!("realtime startup context unavailable; skipping injection");
return None;
}
let mut parts = vec![STARTUP_CONTEXT_HEADER.to_string()];
let has_current_thread_section = current_thread_section.is_some();
let has_recent_work_section = recent_work_section.is_some();
let has_workspace_section = workspace_section.is_some();
if let Some(section) = format_section(
"Current Thread",
current_thread_section,
CURRENT_THREAD_SECTION_TOKEN_BUDGET,
) {
parts.push(section);
}
if let Some(section) = format_section(
"Recent Work",
recent_work_section,
RECENT_WORK_SECTION_TOKEN_BUDGET,
) {
parts.push(section);
}
if let Some(section) = format_section(
"Machine / Workspace Map",
workspace_section,
WORKSPACE_SECTION_TOKEN_BUDGET,
) {
parts.push(section);
}
if let Some(section) = format_section(
"Notes",
Some("Built at realtime startup from the current thread history, local thread metadata, and a bounded local workspace scan. This excludes repo memory instructions, AGENTS files, project-doc prompt blends, and memory summaries.".to_string()),
NOTES_SECTION_TOKEN_BUDGET,
) {
parts.push(section);
}
let context = format_startup_context_blob(&parts.join("\n\n"));
debug!(
approx_tokens = approx_token_count(&context),
requested_budget_tokens = budget_tokens,
bytes = context.len(),
has_current_thread_section,
has_recent_work_section,
has_workspace_section,
"built realtime startup context"
);
info!("realtime startup context: {context}");
Some(context)
}
pub fn build_current_thread_section(turns: &[RealtimeStartupContextTurn]) -> Option<String> {
if turns.is_empty() {
return None;
}
let mut lines = vec![
"Most recent user/assistant turns from this exact thread. Use them for continuity when responding.".to_string(),
];
let mut remaining_budget =
CURRENT_THREAD_SECTION_TOKEN_BUDGET.saturating_sub(approx_token_count(&lines.join("\n")));
let mut retained_turn_count = 0;
for (index, turn) in turns.iter().rev().enumerate() {
if remaining_budget == 0 {
break;
}
let mut turn_lines = Vec::new();
if index == 0 {
turn_lines.push("### Latest turn".to_string());
} else {
turn_lines.push(format!("### Previous turn {index}"));
}
if !turn.user_messages.is_empty() {
turn_lines.push("User:".to_string());
turn_lines.push(turn.user_messages.join("\n\n"));
}
if !turn.assistant_messages.is_empty() {
turn_lines.push(String::new());
turn_lines.push("Assistant:".to_string());
turn_lines.push(turn.assistant_messages.join("\n\n"));
}
let turn_budget = REALTIME_TURN_TOKEN_BUDGET.min(remaining_budget);
let turn_text = turn_lines.join("\n");
let turn_text = truncate_realtime_text_to_token_budget(&turn_text, turn_budget);
let turn_tokens = approx_token_count(&turn_text);
if turn_tokens == 0 {
continue;
}
lines.push(String::new());
lines.push(turn_text);
remaining_budget = remaining_budget.saturating_sub(turn_tokens);
retained_turn_count += 1;
}
(retained_turn_count > 0).then(|| lines.join("\n"))
}
pub fn truncate_realtime_text_to_token_budget(text: &str, budget_tokens: usize) -> String {
let mut truncation_budget = budget_tokens;
loop {
let candidate = truncate_text(text, TruncationPolicy::Tokens(truncation_budget));
let candidate_tokens = approx_token_count(&candidate);
if candidate_tokens <= budget_tokens {
break candidate;
}
let excess_tokens = candidate_tokens.saturating_sub(budget_tokens);
let next_budget = truncation_budget.saturating_sub(excess_tokens.max(1));
if next_budget == 0 {
let candidate = truncate_text(text, TruncationPolicy::Tokens(0));
if approx_token_count(&candidate) <= budget_tokens {
break candidate;
}
break String::new();
}
truncation_budget = next_budget;
}
}
pub async fn build_recent_work_section(
cwd: &AbsolutePathBuf,
recent_threads: &[StoredThread],
) -> Option<String> {
let mut groups: HashMap<PathBuf, Vec<&StoredThread>> = HashMap::new();
for entry in recent_threads {
let group = match AbsolutePathBuf::from_absolute_path(entry.cwd.as_path()) {
Ok(entry_cwd) => resolve_root_git_project_for_trust(LOCAL_FS.as_ref(), &entry_cwd)
.await
.map(AbsolutePathBuf::into_path_buf)
.unwrap_or_else(|| entry.cwd.clone()),
Err(_) => entry.cwd.clone(),
};
groups.entry(group).or_default().push(entry);
}
let current_group = resolve_root_git_project_for_trust(LOCAL_FS.as_ref(), cwd)
.await
.map(AbsolutePathBuf::into_path_buf)
.unwrap_or_else(|| cwd.clone().into_path_buf());
let mut groups = groups.into_iter().collect::<Vec<_>>();
groups.sort_by(|(left_group, left_entries), (right_group, right_entries)| {
let left_latest = left_entries
.iter()
.map(|entry| entry.updated_at)
.max()
.unwrap_or_else(Utc::now);
let right_latest = right_entries
.iter()
.map(|entry| entry.updated_at)
.max()
.unwrap_or_else(Utc::now);
(
*left_group != current_group,
Reverse(left_latest),
left_group.as_os_str(),
)
.cmp(&(
*right_group != current_group,
Reverse(right_latest),
right_group.as_os_str(),
))
});
let mut sections = Vec::new();
for (group, mut entries) in groups.into_iter().take(MAX_RECENT_WORK_GROUPS) {
entries.sort_by_key(|entry| Reverse(entry.updated_at));
if let Some(section) = format_thread_group(&current_group, &group, entries).await {
sections.push(section);
}
}
(!sections.is_empty()).then(|| sections.join("\n\n"))
}
pub async fn build_workspace_section_with_user_root(
cwd: &AbsolutePathBuf,
user_root: Option<PathBuf>,
) -> Option<String> {
let cwd_path = cwd.as_path();
let git_root = resolve_root_git_project_for_trust(LOCAL_FS.as_ref(), cwd).await;
let cwd_tree = render_tree(cwd_path);
let git_root_tree = git_root
.as_ref()
.filter(|git_root| git_root.as_path() != cwd_path)
.and_then(|git_root| render_tree(git_root.as_path()));
let user_root_tree = user_root
.as_ref()
.filter(|user_root| user_root.as_path() != cwd_path)
.filter(|user_root| {
git_root
.as_ref()
.is_none_or(|git_root| git_root.as_path() != user_root.as_path())
})
.and_then(|user_root| render_tree(user_root));
if cwd_tree.is_none() && git_root.is_none() && user_root_tree.is_none() {
return None;
}
let mut lines = vec![
format!("Current working directory: {}", cwd_path.display()),
format!("Working directory name: {}", file_name_string(cwd_path)),
];
if let Some(git_root) = &git_root {
lines.push(format!("Git root: {}", git_root.display()));
lines.push(format!("Git project: {}", file_name_string(git_root)));
}
if let Some(user_root) = &user_root {
lines.push(format!("User root: {}", user_root.display()));
}
if let Some(tree) = cwd_tree {
lines.push(String::new());
lines.push("Working directory tree:".to_string());
lines.extend(tree);
}
if let Some(tree) = git_root_tree {
lines.push(String::new());
lines.push("Git root tree:".to_string());
lines.extend(tree);
}
if let Some(tree) = user_root_tree {
lines.push(String::new());
lines.push("User root tree:".to_string());
lines.extend(tree);
}
Some(lines.join("\n"))
}
pub fn format_section(title: &str, body: Option<String>, budget_tokens: usize) -> Option<String> {
let body = body?;
let body = body.trim();
if body.is_empty() {
return None;
}
let heading = format!("## {title}\n");
let body_budget = budget_tokens.saturating_sub(approx_token_count(&heading));
if body_budget == 0 {
return None;
}
let body = truncate_realtime_text_to_token_budget(body, body_budget);
if body.is_empty() {
return None;
}
Some(format!("{heading}{body}"))
}
pub fn format_startup_context_blob(body: &str) -> String {
format!("{STARTUP_CONTEXT_OPEN_TAG}\n{body}\n{STARTUP_CONTEXT_CLOSE_TAG}")
}
fn render_tree(root: &Path) -> Option<Vec<String>> {
if !root.is_dir() {
return None;
}
let mut lines = Vec::new();
collect_tree_lines(root, /*depth*/ 0, &mut lines);
(!lines.is_empty()).then_some(lines)
}
fn collect_tree_lines(dir: &Path, depth: usize, lines: &mut Vec<String>) {
if depth >= TREE_MAX_DEPTH {
return;
}
let entries = match read_sorted_entries(dir) {
Ok(entries) => entries,
Err(_) => return,
};
let total_entries = entries.len();
for entry in entries.into_iter().take(DIR_ENTRY_LIMIT) {
let Ok(file_type) = entry.file_type() else {
continue;
};
let name = file_name_string(&entry.path());
let indent = " ".repeat(depth);
let suffix = if file_type.is_dir() { "/" } else { "" };
lines.push(format!("{indent}- {name}{suffix}"));
if file_type.is_dir() {
collect_tree_lines(&entry.path(), depth + 1, lines);
}
}
if total_entries > DIR_ENTRY_LIMIT {
lines.push(format!(
"{}- ... {} more entries",
" ".repeat(depth),
total_entries - DIR_ENTRY_LIMIT
));
}
}
fn read_sorted_entries(dir: &Path) -> io::Result<Vec<DirEntry>> {
let mut entries = std::fs::read_dir(dir)?
.filter_map(Result::ok)
.filter(|entry| !is_noisy_name(&entry.file_name()))
.collect::<Vec<_>>();
entries.sort_by(|left, right| {
let left_is_dir = left
.file_type()
.map(|file_type| file_type.is_dir())
.unwrap_or(false);
let right_is_dir = right
.file_type()
.map(|file_type| file_type.is_dir())
.unwrap_or(false);
(!left_is_dir, file_name_string(&left.path()))
.cmp(&(!right_is_dir, file_name_string(&right.path())))
});
Ok(entries)
}
fn is_noisy_name(name: &OsStr) -> bool {
let name = name.to_string_lossy();
name.starts_with('.') || NOISY_DIR_NAMES.iter().any(|noisy| *noisy == name)
}
async fn format_thread_group(
current_group: &Path,
group: &Path,
entries: Vec<&StoredThread>,
) -> Option<String> {
let latest = entries.first()?;
let group_label =
if let Ok(latest_cwd) = AbsolutePathBuf::from_absolute_path(latest.cwd.as_path()) {
if resolve_root_git_project_for_trust(LOCAL_FS.as_ref(), &latest_cwd)
.await
.is_some()
{
format!("### Git repo: {}", group.display())
} else {
format!("### Directory: {}", group.display())
}
} else {
format!("### Directory: {}", group.display())
};
let mut lines = vec![
group_label,
format!("Recent sessions: {}", entries.len()),
format!("Latest activity: {}", latest.updated_at.to_rfc3339()),
];
if let Some(git_branch) = latest
.git_info
.as_ref()
.and_then(|git| git.branch.as_deref())
.filter(|git_branch| !git_branch.is_empty())
{
lines.push(format!("Latest branch: {git_branch}"));
}
lines.push(String::new());
lines.push("User asks:".to_string());
let mut seen = HashSet::new();
let max_asks = if group == current_group {
MAX_CURRENT_CWD_ASKS
} else {
MAX_OTHER_CWD_ASKS
};
for entry in entries {
let Some(first_user_message) = entry.first_user_message.as_deref() else {
continue;
};
let ask = first_user_message
.split_whitespace()
.collect::<Vec<_>>()
.join(" ");
let dedupe_key = format!("{}:{ask}", entry.cwd.display());
if ask.is_empty() || !seen.insert(dedupe_key) {
continue;
}
let ask = if ask.chars().count() > MAX_ASK_CHARS {
format!(
"{}...",
ask.chars()
.take(MAX_ASK_CHARS.saturating_sub(3))
.collect::<String>()
)
} else {
ask
};
lines.push(format!("- {}: {ask}", entry.cwd.display()));
if seen.len() == max_asks {
break;
}
}
(lines.len() > 5).then(|| lines.join("\n"))
}
fn file_name_string(path: &Path) -> String {
path.file_name()
.and_then(OsStr::to_str)
.map(str::to_owned)
.unwrap_or_else(|| path.display().to_string())
}
fn approx_token_count(text: &str) -> usize {
text.len().div_ceil(APPROX_BYTES_PER_TOKEN)
}
codex-rs/realtime/templates/realtime/backend_prompt.md
+3
View File
@@ -0,0 +1,3 @@
## Identity, tone, and role
You are Codex, an OpenAI general-purpose agentic assistant helping the user in a live voice conversation. The user's name is {{ user_first_name }}.