diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 69a9f51b0e..f9cf38bc90 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -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" diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index fd203e6259..5decb33ff2 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -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" } diff --git a/codex-rs/core/Cargo.toml b/codex-rs/core/Cargo.toml index 8770aaf487..2da8f3ba6e 100644 --- a/codex-rs/core/Cargo.toml +++ b/codex-rs/core/Cargo.toml @@ -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 } diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 1809789a8e..d2c8cf5b03 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -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; diff --git a/codex-rs/core/src/realtime_context.rs b/codex-rs/core/src/realtime_context.rs index 6dfb56a8a3..d176b3b777 100644 --- a/codex-rs/core/src/realtime_context.rs +++ b/codex-rs/core/src/realtime_context.rs @@ -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 = ""; -const STARTUP_CONTEXT_CLOSE_TAG: &str = ""; -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 { 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 { @@ -149,61 +56,7 @@ async fn load_recent_threads(sess: &Session) -> Vec { } } -async fn build_recent_work_section( - cwd: &AbsolutePathBuf, - recent_threads: &[StoredThread], -) -> Option { - let mut groups: HashMap> = 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::>(); - 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(¤t_group, &group, entries).await { - sections.push(section); - } - } - (!sections.is_empty()).then(|| sections.join("\n\n")) -} - -fn build_current_thread_section(items: &[ResponseItem]) -> Option { +fn current_thread_turns(items: &[ResponseItem]) -> Vec { 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 { 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 { } 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 { + codex_realtime::build_current_thread_section(¤t_thread_turns(items)) } -async fn build_workspace_section_with_user_root( - cwd: &AbsolutePathBuf, - user_root: Option, -) -> Option { - 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> { - 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) { - 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> { - let mut entries = std::fs::read_dir(dir)? - .filter_map(Result::ok) - .filter(|entry| !is_noisy_name(&entry.file_name())) - .collect::>(); - 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, budget_tokens: usize) -> Option { - 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 { - 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::>() - .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::() - ) - } 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"] diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index 0ca7c1c169..cb10541fa3 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -1,33 +1,13 @@ -use crate::client::ModelClient; +use crate::config::Config; use crate::realtime_context::build_realtime_startup_context; -use crate::realtime_prompt::prepare_realtime_backend_prompt; use crate::session::session::Session; -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::Provider as ApiProvider; -use codex_api::RealtimeAudioFrame; use codex_api::RealtimeEvent; -use codex_api::RealtimeEventParser; use codex_api::RealtimeSessionConfig; -use codex_api::RealtimeSessionMode; use codex_api::RealtimeWebsocketClient; -use codex_api::RealtimeWebsocketEvents; -use codex_api::RealtimeWebsocketWriter; use codex_api::map_api_error; use codex_app_server_protocol::AuthMode; -use codex_config::config_toml::RealtimeWsMode; use codex_config::config_toml::RealtimeWsVersion; -use codex_login::CodexAuth; use codex_login::default_client::default_headers; -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::CodexErrorInfo; use codex_protocol::protocol::ConversationAudioParams; @@ -41,38 +21,31 @@ use codex_protocol::protocol::RealtimeConversationClosedEvent; use codex_protocol::protocol::RealtimeConversationRealtimeEvent; use codex_protocol::protocol::RealtimeConversationSdpEvent; use codex_protocol::protocol::RealtimeConversationStartedEvent; -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; use std::sync::atomic::Ordering; -use tokio::sync::Mutex; -use tokio::task::JoinHandle; use tracing::debug; use tracing::error; use tracing::info; 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; +pub(crate) use codex_realtime::REALTIME_USER_TEXT_PREFIX; +pub(crate) use codex_realtime::RealtimeConversationManager; +use codex_realtime::RealtimeFeaturesConfig; +use codex_realtime::RealtimeStart; +use codex_realtime::RealtimeStartOutput; +use codex_realtime::build_realtime_session_config as build_session_config; +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 REALTIME_STARTUP_CONTEXT_TOKEN_BUDGET: usize = 5_300; -const DEFAULT_REALTIME_MODEL: &str = "gpt-realtime-1.5"; -pub(crate) const REALTIME_USER_TEXT_PREFIX: &str = "[USER] "; -pub(crate) 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 = - "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:"; #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum RealtimeConversationEnd { @@ -81,477 +54,6 @@ enum RealtimeConversationEnd { Error, } -enum RealtimeFanoutTaskStop { - Abort, - Detach, -} - -pub(crate) struct RealtimeConversationManager { - state: Mutex>, -} - -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -enum RealtimeSessionKind { - V1, - V2, -} - -#[derive(Clone, Debug)] -struct RealtimeHandoffState { - output_tx: Sender, - active_handoff: Arc>>, - last_output_text: Arc>>, - 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, - 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, - 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, - 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, - handoff_output_rx: Receiver, - audio_rx: Receiver, - events_tx: Sender, - handoff_state: RealtimeHandoffState, - session_kind: RealtimeSessionKind, - event_parser: RealtimeEventParser, -} - -impl RealtimeHandoffState { - fn new(output_tx: Sender, session_kind: RealtimeSessionKind) -> Self { - Self { - output_tx, - active_handoff: Arc::new(Mutex::new(None)), - last_output_text: Arc::new(Mutex::new(None)), - session_kind, - } - } -} - -#[allow(dead_code)] -struct ConversationState { - audio_tx: Sender, - user_text_tx: Sender, - session_kind: RealtimeSessionKind, - writer: RealtimeWebsocketWriter, - handoff: RealtimeHandoffState, - input_task: JoinHandle<()>, - fanout_task: Option>, - realtime_active: Arc, -} - -struct RealtimeStart { - api_provider: ApiProvider, - extra_headers: Option, - session_config: RealtimeSessionConfig, - model_client: ModelClient, - sdp: Option, -} - -struct RealtimeStartOutput { - realtime_active: Arc, - events_rx: Receiver, - sdp: Option, -} - -#[allow(dead_code)] -impl RealtimeConversationManager { - pub(crate) fn new() -> Self { - Self { - state: Mutex::new(None), - } - } - - pub(crate) 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(crate) 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 - ) - } - - async fn start(&self, start: RealtimeStart) -> CodexResult { - 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 { - let RealtimeStart { - api_provider, - extra_headers, - session_config, - model_client, - sdp, - } = start; - let event_parser = session_config.event_parser; - let session_kind = match event_parser { - RealtimeEventParser::V1 => RealtimeSessionKind::V1, - RealtimeEventParser::RealtimeV2 => RealtimeSessionKind::V2, - }; - - let client = RealtimeWebsocketClient::new(api_provider); - let (connection, sdp) = if let Some(sdp) = sdp { - let call = model_client - .create_realtime_call_with_headers( - sdp, - session_config.clone(), - extra_headers.unwrap_or_default(), - ) - .await?; - let connection = client - .connect_webrtc_sideband( - session_config, - &call.call_id, - call.sideband_headers, - default_headers(), - ) - .await - .map_err(map_api_error)?; - (connection, Some(call.sdp)) - } else { - let connection = client - .connect( - session_config, - extra_headers.unwrap_or_default(), - default_headers(), - ) - .await - .map_err(map_api_error)?; - (connection, None) - }; - - let writer = connection.writer(); - let events = connection.events(); - let (audio_tx, audio_rx) = - async_channel::bounded::(AUDIO_IN_QUEUE_CAPACITY); - let (user_text_tx, user_text_rx) = - async_channel::bounded::(USER_TEXT_IN_QUEUE_CAPACITY); - let (handoff_output_tx, handoff_output_rx) = - async_channel::bounded::(HANDOFF_OUT_QUEUE_CAPACITY); - let (events_tx, events_rx) = - async_channel::bounded::(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, - writer, - handoff, - input_task: task, - fanout_task: None, - realtime_active: Arc::clone(&realtime_active), - }); - Ok(RealtimeStartOutput { - realtime_active, - events_rx, - sdp, - }) - } - - pub(crate) async fn register_fanout_task( - &self, - realtime_active: &Arc, - 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(crate) async fn finish_if_active(&self, realtime_active: &Arc) { - 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(crate) 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(crate) 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 = prefix_realtime_text(text, REALTIME_USER_TEXT_PREFIX, session_kind); - sender - .send(text) - .await - .map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))?; - Ok(()) - } - - pub(crate) 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 = prefix_realtime_text( - output_text, - REALTIME_BACKEND_TEXT_PREFIX, - handoff.session_kind, - ); - *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(crate) 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(crate) async fn active_handoff_id(&self) -> Option { - let handoff = { - let guard = self.state.lock().await; - guard.as_ref().map(|state| state.handoff.clone()) - }?; - handoff.active_handoff.lock().await.clone() - } - - pub(crate) 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(crate) 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 => {} - } - } -} - pub(crate) async fn handle_start( sess: &Arc, sub_id: String, @@ -588,12 +90,9 @@ pub(crate) async fn handle_start( } struct PreparedRealtimeConversationStart { - api_provider: ApiProvider, - extra_headers: Option, requested_session_id: Option, version: RealtimeWsVersion, - session_config: RealtimeSessionConfig, - transport: ConversationStartTransport, + start: RealtimeStart, } async fn prepare_realtime_start( @@ -611,13 +110,17 @@ async fn prepare_realtime_start( let transport = params .transport .unwrap_or(ConversationStartTransport::Websocket); + let realtime_config = realtime_features_config(config.as_ref()); + let version = realtime_config.session.version; + let mut api_provider = provider.to_api_provider(Some(AuthMode::ApiKey))?; - if let Some(realtime_ws_base_url) = &config.experimental_realtime_ws_base_url { + if let Some(realtime_ws_base_url) = &realtime_config.websocket_base_url { api_provider.base_url = realtime_ws_base_url.clone(); } - let version = config.realtime.version; + let session_config = build_realtime_session_config( sess, + &realtime_config, params.prompt, params.session_id, params.output_modality, @@ -625,159 +128,109 @@ async fn prepare_realtime_start( ) .await?; let requested_session_id = session_config.session_id.clone(); - let extra_headers = match transport { + let event_parser = session_config.event_parser; + let client = RealtimeWebsocketClient::new(api_provider); + + let start = match transport { ConversationStartTransport::Websocket => { - let realtime_api_key = realtime_api_key(auth.as_ref(), &provider)?; - realtime_request_headers( - requested_session_id.as_deref(), - Some(realtime_api_key.as_str()), - )? + let api_key = realtime_api_key(auth.as_ref(), &provider)?; + let extra_headers = + realtime_request_headers(requested_session_id.as_deref(), Some(api_key.as_str()))? + .unwrap_or_default(); + let connection = client + .connect(session_config, extra_headers, default_headers()) + .await + .map_err(map_api_error)?; + RealtimeStart { + writer: connection.writer(), + events: connection.events(), + event_parser, + sdp: None, + } } - ConversationStartTransport::Webrtc { .. } => { - realtime_request_headers(requested_session_id.as_deref(), /*api_key*/ None)? + ConversationStartTransport::Webrtc { sdp } => { + let extra_headers = + realtime_request_headers(requested_session_id.as_deref(), /*api_key*/ None)? + .unwrap_or_default(); + let call = sess + .services + .model_client + .create_realtime_call_with_headers(sdp, session_config.clone(), extra_headers) + .await?; + let connection = client + .connect_webrtc_sideband( + session_config, + &call.call_id, + call.sideband_headers, + default_headers(), + ) + .await + .map_err(map_api_error)?; + RealtimeStart { + writer: connection.writer(), + events: connection.events(), + event_parser, + sdp: Some(call.sdp), + } } }; + Ok(PreparedRealtimeConversationStart { - api_provider, - extra_headers, requested_session_id, version, - session_config, - transport, + start, }) } pub(crate) async fn build_realtime_session_config( sess: &Arc, + config: &RealtimeFeaturesConfig, prompt: Option>, session_id: Option, output_modality: RealtimeOutputModality, voice: Option, ) -> CodexResult { - let config = sess.get_config().await; - let prompt = prepare_realtime_backend_prompt( - prompt, - config.experimental_realtime_ws_backend_prompt.clone(), - ); - let startup_context = match config.experimental_realtime_ws_startup_context.clone() { - Some(startup_context) => startup_context, + let startup_context = match &config.websocket_startup_context { + Some(startup_context) => startup_context.clone(), None => { build_realtime_startup_context(sess.as_ref(), REALTIME_STARTUP_CONTEXT_TOKEN_BUDGET) .await .unwrap_or_default() } }; - 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 - .experimental_realtime_ws_model - .clone() - .unwrap_or_else(|| DEFAULT_REALTIME_MODEL.to_string()), - ); - let event_parser = match config.realtime.version { - RealtimeWsVersion::V1 => RealtimeEventParser::V1, - RealtimeWsVersion::V2 => RealtimeEventParser::RealtimeV2, - }; - if config.realtime.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.realtime.session_type { - RealtimeWsMode::Conversational => RealtimeSessionMode::Conversational, - RealtimeWsMode::Transcription => RealtimeSessionMode::Transcription, - }; - let voice = voice - .or(config.realtime.voice) - .unwrap_or_else(|| default_realtime_voice(config.realtime.version)); - validate_realtime_voice(config.realtime.version, voice)?; - Ok(RealtimeSessionConfig { - instructions: prompt, - model, - session_id: Some(session_id.unwrap_or_else(|| sess.conversation_id.to_string())), - event_parser, - session_mode, + build_session_config( + config, + prompt, + Some(session_id.unwrap_or_else(|| sess.conversation_id.to_string())), output_modality, voice, - }) + startup_context, + ) } -fn default_realtime_voice(version: RealtimeWsVersion) -> RealtimeVoice { - let voices = RealtimeVoicesList::builtin(); - match version { - RealtimeWsVersion::V1 => voices.default_v1, - RealtimeWsVersion::V2 => voices.default_v2, +fn realtime_features_config(config: &Config) -> RealtimeFeaturesConfig { + RealtimeFeaturesConfig { + audio: config.realtime_audio.clone(), + session: config.realtime.clone(), + websocket_base_url: config.experimental_realtime_ws_base_url.clone(), + websocket_model: config.experimental_realtime_ws_model.clone(), + websocket_backend_prompt: config.experimental_realtime_ws_backend_prompt.clone(), + websocket_startup_context: config.experimental_realtime_ws_startup_context.clone(), + start_instructions: config.experimental_realtime_start_instructions.clone(), } } -fn prefix_realtime_text(text: String, prefix: &str, session_kind: RealtimeSessionKind) -> String { - if session_kind != RealtimeSessionKind::V2 || text.is_empty() || text.starts_with(prefix) { - return text; - } - format!("{prefix}{text}") -} - -pub(crate) fn prefix_realtime_v2_text(text: String, prefix: &str) -> String { - prefix_realtime_text(text, prefix, RealtimeSessionKind::V2) -} - -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() - ))) -} - async fn handle_start_inner( sess: &Arc, sub_id: &str, prepared_start: PreparedRealtimeConversationStart, ) -> CodexResult<()> { let PreparedRealtimeConversationStart { - api_provider, - extra_headers, requested_session_id, version, - session_config, - transport, + start, } = prepared_start; info!("starting realtime conversation"); - let sdp = match transport { - ConversationStartTransport::Websocket => None, - ConversationStartTransport::Webrtc { sdp } => Some(sdp), - }; - let start = RealtimeStart { - api_provider, - extra_headers, - session_config, - model_client: sess.services.model_client.clone(), - sdp, - }; let start_output = sess.conversation.start(start).await?; info!("realtime conversation started"); @@ -820,22 +273,18 @@ async fn handle_start_inner( match &event { RealtimeEvent::AudioOut(_) => {} _ => { - info!( - event = ?event, - "received realtime conversation event" - ); + info!(event = ?event, "received realtime conversation event"); } } if let RealtimeEvent::Error(_) = &event { end = RealtimeConversationEnd::Error; } - let maybe_routed_text = match &event { + if let Some(text) = match &event { RealtimeEvent::HandoffRequested(handoff) => { realtime_delegation_from_handoff(handoff) } _ => None, - }; - if let Some(text) = maybe_routed_text { + } { debug!(text = %text, "[realtime-text] realtime conversation text output"); let sess_for_routed_text = Arc::clone(&sess_clone); sess_for_routed_text.route_realtime_text_input(text).await; @@ -888,97 +337,6 @@ pub(crate) async fn handle_audio( } } -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) -} - -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)) -} - -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(), - )) -} - -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('>', ">") -} - -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()); - } - - // TODO(aibrahim): Remove this temporary fallback once realtime auth no longer - // requires API key auth for ChatGPT/SIWC sessions. - 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(), - )) -} - -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)) -} - pub(crate) async fn handle_text( sess: &Arc, sub_id: String, @@ -1000,403 +358,6 @@ pub(crate) async fn handle_close(sess: &Arc, sub_id: String) { end_realtime_conversation(sess, sub_id, RealtimeConversationEnd::Requested).await; } -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 = None; - let mut response_create_queue = RealtimeResponseCreateQueue::default(); - - loop { - let result = tokio::select! { - // Text typed by the user that should be sent into realtime. - user_text = user_text_rx.recv() => { - handle_user_text_input( - user_text, - &writer, - &events_tx, - ) - .await - } - // Background agent progress or final output that should be sent back to realtime. - 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 - } - // Events received from the realtime server. - 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 - } - // Audio frames captured from the user microphone. - 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, - writer: &RealtimeWebsocketWriter, - events_tx: &Sender, -) -> 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, - writer: &RealtimeWebsocketWriter, - events_tx: &Sender, - 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, ApiError>, - writer: &RealtimeWebsocketWriter, - events_tx: &Sender, - handoff_state: &RealtimeHandoffState, - session_kind: RealtimeSessionKind, - output_audio_state: &mut Option, - 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) => { - match session_kind { - RealtimeSessionKind::V1 => {} - RealtimeSessionKind::V2 => { - update_output_audio_state(output_audio_state, frame); - } - } - false - } - RealtimeEvent::InputAudioSpeechStarted(event) => { - match session_kind { - RealtimeSessionKind::V1 => {} - RealtimeSessionKind::V2 => { - if 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(_) => { - match session_kind { - RealtimeSessionKind::V1 => {} - RealtimeSessionKind::V2 => response_create_queue.mark_started(), - } - false - } - RealtimeEvent::ResponseCancelled(_) => { - *output_audio_state = None; - match session_kind { - RealtimeSessionKind::V1 => {} - RealtimeSessionKind::V2 => { - response_create_queue - .mark_finished(writer, events_tx, "deferred") - .await?; - } - } - false - } - RealtimeEvent::ResponseDone(_) => { - *output_audio_state = None; - match session_kind { - RealtimeSessionKind::V1 => {} - 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; - - match session_kind { - RealtimeSessionKind::V1 => {} - RealtimeSessionKind::V2 => { - if 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 { session_id, .. } => { - info!(realtime_session_id = %session_id, "realtime session updated"); - false - } - 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, - writer: &RealtimeWebsocketWriter, - events_tx: &Sender, -) -> 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, - 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 { - 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() -} - async fn send_conversation_error( sess: &Arc, sub_id: String, diff --git a/codex-rs/realtime/BUILD.bazel b/codex-rs/realtime/BUILD.bazel new file mode 100644 index 0000000000..a0e4114ae1 --- /dev/null +++ b/codex-rs/realtime/BUILD.bazel @@ -0,0 +1,7 @@ +load("//:defs.bzl", "codex_rust_crate") + +codex_rust_crate( + name = "realtime", + crate_name = "codex_realtime", + compile_data = glob(["templates/**/*.md"]), +) diff --git a/codex-rs/realtime/Cargo.toml b/codex-rs/realtime/Cargo.toml new file mode 100644 index 0000000000..35fff0201d --- /dev/null +++ b/codex-rs/realtime/Cargo.toml @@ -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 diff --git a/codex-rs/realtime/src/config.rs b/codex-rs/realtime/src/config.rs new file mode 100644 index 0000000000..80327a7aff --- /dev/null +++ b/codex-rs/realtime/src/config.rs @@ -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, + pub websocket_model: Option, + pub websocket_backend_prompt: Option, + pub websocket_startup_context: Option, + pub start_instructions: Option, +} diff --git a/codex-rs/realtime/src/conversation_manager.rs b/codex-rs/realtime/src/conversation_manager.rs new file mode 100644 index 0000000000..f6ddf67067 --- /dev/null +++ b/codex-rs/realtime/src/conversation_manager.rs @@ -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>, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum RealtimeSessionKind { + V1, + V2, +} + +#[derive(Clone, Debug)] +struct RealtimeHandoffState { + output_tx: Sender, + active_handoff: Arc>>, + last_output_text: Arc>>, + 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, + 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, + 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, + 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, + handoff_output_rx: Receiver, + audio_rx: Receiver, + events_tx: Sender, + handoff_state: RealtimeHandoffState, + session_kind: RealtimeSessionKind, + event_parser: RealtimeEventParser, +} + +impl RealtimeHandoffState { + fn new(output_tx: Sender, 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, + user_text_tx: Sender, + session_kind: RealtimeSessionKind, + handoff: RealtimeHandoffState, + input_task: JoinHandle<()>, + fanout_task: Option>, + realtime_active: Arc, +} + +pub struct RealtimeStart { + pub writer: RealtimeWebsocketWriter, + pub events: RealtimeWebsocketEvents, + pub event_parser: RealtimeEventParser, + pub sdp: Option, +} + +pub struct RealtimeStartOutput { + pub realtime_active: Arc, + pub events_rx: Receiver, + pub sdp: Option, +} + +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 { + 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 { + 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::(AUDIO_IN_QUEUE_CAPACITY); + let (user_text_tx, user_text_rx) = + async_channel::bounded::(USER_TEXT_IN_QUEUE_CAPACITY); + let (handoff_output_tx, handoff_output_rx) = + async_channel::bounded::(HANDOFF_OUT_QUEUE_CAPACITY); + let (events_tx, events_rx) = + async_channel::bounded::(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, + 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) { + 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 { + 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 = 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, + writer: &RealtimeWebsocketWriter, + events_tx: &Sender, +) -> 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, + writer: &RealtimeWebsocketWriter, + events_tx: &Sender, + 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, ApiError>, + writer: &RealtimeWebsocketWriter, + events_tx: &Sender, + handoff_state: &RealtimeHandoffState, + session_kind: RealtimeSessionKind, + output_audio_state: &mut Option, + 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, + writer: &RealtimeWebsocketWriter, + events_tx: &Sender, +) -> 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, + 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 { + 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() +} 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 new file mode 100644 index 0000000000..4c55d928bc --- /dev/null +++ b/codex-rs/realtime/src/lib.rs @@ -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; diff --git a/codex-rs/realtime/src/prompt.rs b/codex-rs/realtime/src/prompt.rs new file mode 100644 index 0000000000..e99e6ebad3 --- /dev/null +++ b/codex-rs/realtime/src/prompt.rs @@ -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>, + config_prompt: Option, +) -> 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, ¤t_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 }}")); + } +} 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)) +} diff --git a/codex-rs/realtime/src/startup_context.rs b/codex-rs/realtime/src/startup_context.rs new file mode 100644 index 0000000000..1a6b3bfddf --- /dev/null +++ b/codex-rs/realtime/src/startup_context.rs @@ -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 = ""; +const STARTUP_CONTEXT_CLOSE_TAG: &str = ""; +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, + pub assistant_messages: Vec, +} + +#[derive(Clone, Debug)] +pub struct RealtimeStartupContext { + pub cwd: AbsolutePathBuf, + pub current_thread_turns: Vec, + pub recent_threads: Vec, + pub user_root: Option, +} + +pub async fn build_realtime_startup_context( + context: &RealtimeStartupContext, + budget_tokens: usize, +) -> Option { + 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 { + 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 { + let mut groups: HashMap> = 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::>(); + 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(¤t_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, +) -> Option { + 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, budget_tokens: usize) -> Option { + 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> { + 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) { + 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> { + let mut entries = std::fs::read_dir(dir)? + .filter_map(Result::ok) + .filter(|entry| !is_noisy_name(&entry.file_name())) + .collect::>(); + 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 { + 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::>() + .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::() + ) + } 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) +} diff --git a/codex-rs/realtime/templates/realtime/backend_prompt.md b/codex-rs/realtime/templates/realtime/backend_prompt.md new file mode 100644 index 0000000000..e8a9d4451c --- /dev/null +++ b/codex-rs/realtime/templates/realtime/backend_prompt.md @@ -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 }}.