diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 836156e7d8..9ff507f98f 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2058,6 +2058,7 @@ dependencies = [ "codex-http-client", "codex-protocol", "codex-utils-rustls-provider", + "codex-utils-string", "codex-websocket-client", "eventsource-stream", "futures", @@ -4664,6 +4665,7 @@ dependencies = [ "codex-utils-template", "pretty_assertions", "tracing", + "whoami 1.6.1", ] [[package]] @@ -5219,6 +5221,7 @@ dependencies = [ "chrono", "clap", "codex-ansi-escape", + "codex-api", "codex-app-server-client", "codex-app-server-daemon", "codex-app-server-protocol", @@ -5247,6 +5250,7 @@ dependencies = [ "codex-models-manager", "codex-otel", "codex-plugin", + "codex-prompts", "codex-protocol", "codex-protocol-test-support", "codex-realtime-webrtc", diff --git a/codex-rs/codex-api/Cargo.toml b/codex-rs/codex-api/Cargo.toml index 90db05b840..bc82dd0b92 100644 --- a/codex-rs/codex-api/Cargo.toml +++ b/codex-rs/codex-api/Cargo.toml @@ -13,6 +13,7 @@ codex-client = { workspace = true } codex-http-client = { workspace = true } codex-protocol = { workspace = true } codex-utils-rustls-provider = { workspace = true } +codex-utils-string = { workspace = true } codex-websocket-client = { workspace = true } futures = { workspace = true } http = { workspace = true } diff --git a/codex-rs/codex-api/src/lib.rs b/codex-rs/codex-api/src/lib.rs index c2e302f9f1..23bf38193c 100644 --- a/codex-rs/codex-api/src/lib.rs +++ b/codex-rs/codex-api/src/lib.rs @@ -18,6 +18,7 @@ pub(crate) mod telemetry; /// Path of the Responses endpoint, relative to a provider's base URL. pub const RESPONSES_PATH: &str = "/responses"; +pub use crate::requests::headers::build_realtime_thread_headers; pub use crate::requests::headers::build_session_headers; pub use codex_client::RequestTelemetry; pub use codex_client::ReqwestTransport; diff --git a/codex-rs/codex-api/src/requests/headers.rs b/codex-rs/codex-api/src/requests/headers.rs index c5cede8ac6..9c02f9a5e5 100644 --- a/codex-rs/codex-api/src/requests/headers.rs +++ b/codex-rs/codex-api/src/requests/headers.rs @@ -1,7 +1,27 @@ use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::ThreadSource; +use codex_utils_string::to_ascii_json_string; use http::HeaderMap; use http::HeaderValue; +/// Correlate both the media call and its observer with the same thread and session tree. +pub fn build_realtime_thread_headers( + session_id: String, + thread_id: String, + thread_source: Option, +) -> HeaderMap { + let mut headers = build_session_headers(Some(session_id), Some(thread_id)); + // Voice calls can span zero or many backing turns. Only the saved thread source applies. + if let Some(source) = thread_source + && source.as_str().len() <= 256 + && let Ok(metadata) = to_ascii_json_string(&serde_json::json!({ "thread_source": source })) + && let Ok(value) = HeaderValue::from_str(&metadata) + { + headers.insert("x-codex-turn-metadata", value); + } + headers +} + pub fn build_session_headers(session_id: Option, thread_id: Option) -> HeaderMap { let mut headers = HeaderMap::new(); if let Some(id) = session_id { diff --git a/codex-rs/core/Cargo.toml b/codex-rs/core/Cargo.toml index 5c26e95f6c..c485813d02 100644 --- a/codex-rs/core/Cargo.toml +++ b/codex-rs/core/Cargo.toml @@ -131,7 +131,6 @@ tracing = { workspace = true, features = ["log"] } url = { workspace = true } uuid = { workspace = true, features = ["serde", "v4", "v5", "v7"] } which = { workspace = true } -whoami = { workspace = true } # Build OpenSSL from source for musl builds. [target.x86_64-unknown-linux-musl.dependencies] @@ -182,6 +181,7 @@ tracing-opentelemetry = { workspace = true } tracing-subscriber = { workspace = true } tracing-test = { workspace = true, features = ["no-env-filter"] } walkdir = { workspace = true } +whoami = { workspace = true } wiremock = { workspace = true } zstd = { workspace = true } diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 90d089405b..12d65a0b78 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -13,7 +13,6 @@ mod model_request; mod realtime_context; mod realtime_conversation; mod realtime_history; -mod realtime_prompt; mod responses_headers; pub use responses_headers::CodexResponsesHeaders; mod responses_metadata; diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index cf4c4f5a13..91958831d0 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -1,12 +1,9 @@ use crate::client::ModelClient; -use crate::client::X_CODEX_TURN_METADATA_HEADER; use crate::context::ContextualUserFragment; use crate::context::RealtimeDelegation; use crate::context::RealtimeDelegationSource; use crate::realtime_context::build_realtime_startup_context; use crate::realtime_context::truncate_realtime_text_to_token_budget; -use crate::realtime_prompt::prepare_realtime_backend_prompt; -use crate::responses_metadata::THREAD_SOURCE_KEY; use crate::session::session::Session; use anyhow::Context; use async_channel::Receiver; @@ -27,7 +24,7 @@ use codex_api::RealtimeWebsocketClient; use codex_api::RealtimeWebsocketConnection; use codex_api::RealtimeWebsocketEvents; use codex_api::RealtimeWebsocketWriter; -use codex_api::build_session_headers; +use codex_api::build_realtime_thread_headers; use codex_api::map_api_error; use codex_async_utils::OrCancelExt; use codex_config::config_toml::RealtimeWsMode; @@ -41,6 +38,7 @@ use codex_login::default_client::default_headers; use codex_login::read_openai_api_key_from_env; use codex_model_provider::resolve_api_provider; use codex_model_provider_info::ModelProviderInfo; +use codex_prompts::prepare_realtime_backend_prompt; use codex_protocol::auth::AuthMode; use codex_protocol::error::CodexErr; use codex_protocol::error::Result as CodexResult; @@ -69,7 +67,6 @@ use codex_protocol::protocol::RealtimeVoicesList; use codex_utils_output_truncation::approx_bytes_for_tokens; use codex_utils_string::approx_token_count; use codex_utils_string::take_bytes_at_char_boundary; -use codex_utils_string::to_ascii_json_string; use http::HeaderMap; use http::HeaderValue; use http::header::AUTHORIZATION; @@ -106,7 +103,6 @@ const AUDIO_IN_QUEUE_CAPACITY: usize = 256; const TEXT_IN_QUEUE_CAPACITY: usize = 64; const HANDOFF_OUT_QUEUE_CAPACITY: usize = 64; const OUTPUT_EVENTS_QUEUE_CAPACITY: usize = 256; -const REALTIME_THREAD_SOURCE_MAX_BYTES: usize = 256; const REALTIME_STARTUP_CONTEXT_TOKEN_BUDGET: usize = 5_300; const REALTIME_ASSISTANT_OUTPUT_TOKEN_BUDGET: usize = 1_000; const REALTIME_REASONING_STATUS_MAX_BYTES: usize = 256; @@ -1648,18 +1644,11 @@ async fn prepare_realtime_start( } } .unwrap_or_default(); - extra_headers.extend(build_session_headers( - Some(sess.session_id().to_string()), - Some(sess.thread_id().to_string()), + extra_headers.extend(build_realtime_thread_headers( + sess.session_id().to_string(), + sess.thread_id().to_string(), + sess.thread_config_snapshot().await.thread_source, )); - // Voice calls can span zero or many backing turns; send only the saved thread source. - if let Some(thread_source) = sess.thread_config_snapshot().await.thread_source - && thread_source.as_str().len() <= REALTIME_THREAD_SOURCE_MAX_BYTES - && let Ok(metadata) = to_ascii_json_string(&json!({ THREAD_SOURCE_KEY: thread_source })) - && let Ok(metadata) = HeaderValue::from_str(&metadata) - { - extra_headers.insert(X_CODEX_TURN_METADATA_HEADER, metadata); - } Ok(PreparedRealtimeConversationStart { attachment_control: RealtimeAttachmentControl::default(), sideband: None, diff --git a/codex-rs/core/src/realtime_prompt.rs b/codex-rs/core/src/realtime_prompt.rs deleted file mode 100644 index 689c412387..0000000000 --- a/codex-rs/core/src/realtime_prompt.rs +++ /dev/null @@ -1,82 +0,0 @@ -use codex_prompts::BACKEND_PROMPT; -const DEFAULT_USER_FIRST_NAME: &str = "there"; -const USER_FIRST_NAME_PLACEHOLDER: &str = "{{ user_first_name }}"; - -pub(crate) 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/model-provider/src/lib.rs b/codex-rs/model-provider/src/lib.rs index 5b10f51263..6338f9f4a2 100644 --- a/codex-rs/model-provider/src/lib.rs +++ b/codex-rs/model-provider/src/lib.rs @@ -57,6 +57,7 @@ pub use provider::ProviderAuthRecoveryMessages; pub use provider::ProviderUnauthorizedRecovery; pub use provider::SharedModelProvider; pub use provider::create_model_provider; +pub use provider::create_model_provider_from_local_config; pub use provider::provider_uses_first_party_auth_path; pub use provider::resolve_api_provider; diff --git a/codex-rs/model-provider/src/provider.rs b/codex-rs/model-provider/src/provider.rs index e0c73083b7..897d9aaf73 100644 --- a/codex-rs/model-provider/src/provider.rs +++ b/codex-rs/model-provider/src/provider.rs @@ -11,6 +11,8 @@ use codex_api::SharedAuthProvider; use codex_api::TransportError; use codex_features::RuntimeAuthMode; use codex_login::AuthManager; +use codex_login::AuthManagerConfig; +use codex_login::AuthManagerInitializationError; use codex_login::CodexAuth; use codex_login::WorkspaceRoutingRequest; use codex_login::default_client::ClientRedirectPolicy; @@ -444,6 +446,16 @@ pub fn create_model_provider( create_model_provider_with_runtime(provider_info, auth_manager, /*runtime_auth*/ None) } +/// Creates a model provider using the local login settings. +pub async fn create_model_provider_from_local_config( + provider_info: ModelProviderInfo, + config: &impl AuthManagerConfig, +) -> Result { + let credentials = + AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ false).await?; + Ok(create_model_provider(provider_info, Some(credentials))) +} + pub(crate) fn create_model_provider_with_runtime( provider_info: ModelProviderInfo, auth_manager: Option>, diff --git a/codex-rs/prompts/Cargo.toml b/codex-rs/prompts/Cargo.toml index faf6be294f..5f0df856ab 100644 --- a/codex-rs/prompts/Cargo.toml +++ b/codex-rs/prompts/Cargo.toml @@ -23,6 +23,7 @@ codex-protocol = { workspace = true } codex-utils-absolute-path = { workspace = true } codex-utils-template = { workspace = true } tracing = { workspace = true } +whoami = { workspace = true } [dev-dependencies] codex-protocol-test-support = { workspace = true } diff --git a/codex-rs/prompts/src/lib.rs b/codex-rs/prompts/src/lib.rs index 7690ceac6b..a83c268817 100644 --- a/codex-rs/prompts/src/lib.rs +++ b/codex-rs/prompts/src/lib.rs @@ -30,6 +30,7 @@ pub use permissions_instructions::PermissionsInstructions; pub use realtime::BACKEND_PROMPT; pub use realtime::END_INSTRUCTIONS; pub use realtime::START_INSTRUCTIONS; +pub use realtime::prepare_realtime_backend_prompt; pub use review_exit::render_review_exit_interrupted; pub use review_exit::render_review_exit_success; pub use review_request::REVIEW_PROMPT; diff --git a/codex-rs/prompts/src/realtime.rs b/codex-rs/prompts/src/realtime.rs index b992e2b111..006e60ad10 100644 --- a/codex-rs/prompts/src/realtime.rs +++ b/codex-rs/prompts/src/realtime.rs @@ -1,3 +1,90 @@ pub const BACKEND_PROMPT: &str = include_str!("../templates/realtime/backend_prompt.md"); pub const END_INSTRUCTIONS: &str = include_str!("../templates/realtime/realtime_end.md"); pub const START_INSTRUCTIONS: &str = include_str!("../templates/realtime/realtime_start.md"); + +const DEFAULT_USER_FIRST_NAME: &str = "there"; +const USER_FIRST_NAME_PLACEHOLDER: &str = "{{ user_first_name }}"; + +/// Resolve the configured, requested, or default realtime instructions. +/// +/// A nonblank config prompt takes precedence. Otherwise, an omitted request +/// prompt (`None`) uses the default, explicit JSON `null` (`Some(None)`) clears +/// the backend prompt, and a supplied string (`Some(Some(text))`) is used as-is. +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/tui/Cargo.toml b/codex-rs/tui/Cargo.toml index 6e5c58f6a0..8a5a0ecbc4 100644 --- a/codex-rs/tui/Cargo.toml +++ b/codex-rs/tui/Cargo.toml @@ -29,6 +29,8 @@ base64 = { workspace = true } chrono = { workspace = true, features = ["serde"] } clap = { workspace = true, features = ["derive"] } codex-ansi-escape = { workspace = true } +codex-api = { workspace = true } +codex-prompts = { workspace = true } codex-app-server-client = { workspace = true } codex-app-server-daemon = { workspace = true } codex-app-server-protocol = { workspace = true } @@ -84,6 +86,7 @@ derive_more = { workspace = true, features = ["debug", "is_variant"] } diffy = { workspace = true } dirs = { workspace = true } dunce = { workspace = true } +http = { workspace = true } image = { workspace = true, features = ["jpeg", "png", "gif", "webp"] } itertools = { workspace = true } lazy_static = { workspace = true } @@ -185,7 +188,6 @@ codex-utils-cargo-bin = { workspace = true } assert_matches = { workspace = true } chrono = { workspace = true, features = ["serde"] } futures = { workspace = true } -http = { workspace = true } insta = { workspace = true } opentelemetry_sdk = { workspace = true } pretty_assertions = { workspace = true } diff --git a/codex-rs/tui/src/app/event_dispatch.rs b/codex-rs/tui/src/app/event_dispatch.rs index d98f762733..eeebcbac23 100644 --- a/codex-rs/tui/src/app/event_dispatch.rs +++ b/codex-rs/tui/src/app/event_dispatch.rs @@ -2466,6 +2466,28 @@ impl App { offer.handle.close(); } } + AppEvent::RealtimeCallCreated { thread_id, attempt_id, result } => { + if let Some(owner) = self.voice_widget_for_thread(thread_id) + && owner.is_starting_realtime_attempt(thread_id, attempt_id) + { + let result = match result { + Ok(call) => { + owner.accept_realtime_call(call.answer_sdp); + app_server.thread_realtime_start(thread_id, call.call_id).await + } + Err(error) => Err(error), + }; + if let Err(error) = result + && !self.recover_transport_error(&error) + && let Some(owner) = self.voice_widget_for_thread(thread_id) + { + owner.on_realtime_error( + format!("Voice conversation failed: {error:#}"), + crate::chatwidget::RealtimeFailureCause::AppServerRequest, + ); + } + } + } AppEvent::RealtimeWebrtcConnected { thread_id, attempt_id, diff --git a/codex-rs/tui/src/app/realtime_settings.rs b/codex-rs/tui/src/app/realtime_settings.rs index 356f508b54..426b52f879 100644 --- a/codex-rs/tui/src/app/realtime_settings.rs +++ b/codex-rs/tui/src/app/realtime_settings.rs @@ -180,7 +180,6 @@ impl App { } pub(super) async fn realtime_voices( - &self, app_server: &AppServerSession, ) -> Result { let response = app_server @@ -196,14 +195,14 @@ impl App { } pub(super) async fn effective_realtime_voice( - &self, app_server: &AppServerSession, + cwd: &std::path::Path, voices: &RealtimeVoicesList, ) -> Result> { // Session setup installs the active conversation cwd for both server modes. let config = crate::config_update::read_effective_config_if_supported( app_server.request_handle(), - &self.chat_widget.config_ref().cwd, + cwd, ) .await?; config @@ -214,7 +213,7 @@ impl App { } pub(super) async fn open_realtime_voices(&mut self, app_server: &AppServerSession) { - let voices = match self.realtime_voices(app_server).await { + let voices = match Self::realtime_voices(app_server).await { Ok(voices) => voices, Err(error) => { self.chat_widget @@ -222,7 +221,13 @@ impl App { return; } }; - match self.effective_realtime_voice(app_server, &voices).await { + match Self::effective_realtime_voice( + app_server, + &self.chat_widget.config_ref().cwd, + &voices, + ) + .await + { Ok(voice) => self.chat_widget.open_realtime_voices(voice, voices), Err(error) => self .chat_widget @@ -245,7 +250,7 @@ impl App { .await { Ok(response) => { - let voices = match self.realtime_voices(app_server).await { + let voices = match Self::realtime_voices(app_server).await { Ok(voices) => voices, Err(error) => { self.chat_widget.add_error_message(format!( diff --git a/codex-rs/tui/src/app/tests/realtime_handoff_e2e.rs b/codex-rs/tui/src/app/tests/realtime_handoff_e2e.rs index 38a395e89f..dbc2770284 100644 --- a/codex-rs/tui/src/app/tests/realtime_handoff_e2e.rs +++ b/codex-rs/tui/src/app/tests/realtime_handoff_e2e.rs @@ -127,13 +127,19 @@ async fn delegated_core_events_keep_private_output_hidden_and_deliver_final_spee crate::chatwidget::activate_voice_for_thread(&mut app.chat_widget, thread_id); while ops.try_recv().is_ok() {} let mut tui = crate::tui::test_support::make_test_tui()?; + let thread = app_server + .thread_read(thread_id, /*include_turns*/ false) + .await?; + let call = crate::realtime_call::create( + &app.config, + &thread, + "v=offer\r\n".to_string(), + codex_protocol::protocol::RealtimeVoicesList::builtin().default_v1, + ) + .await?; + assert_eq!(String::from(call.answer_sdp), "v=answer\r\n"); app_server - .thread_realtime_start( - thread_id, - "v=offer\r\n".to_string(), - /*model*/ None, - /*voice*/ None, - ) + .thread_realtime_start(thread_id, call.call_id) .await?; // The sideband fixture sends its delegation after the first outbound request. app_server diff --git a/codex-rs/tui/src/app/tests/realtime_requests.rs b/codex-rs/tui/src/app/tests/realtime_requests.rs index d4cb7cc43b..67c97e4a97 100644 --- a/codex-rs/tui/src/app/tests/realtime_requests.rs +++ b/codex-rs/tui/src/app/tests/realtime_requests.rs @@ -85,6 +85,22 @@ enum ItemEventKind { Completed, } +async fn mock_voice_calls(config: &mut crate::legacy_core::config::Config) -> wiremock::MockServer { + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("POST")) + .and(wiremock::matchers::path("/backend-api/realtime/calls")) + .respond_with( + wiremock::ResponseTemplate::new(201) + .insert_header("Location", "/realtime/calls/rtc_tui") + .set_body_string("v=answer\r\n"), + ) + .mount(&server) + .await; + config.experimental_realtime_webrtc_call_base_url = + Some(format!("{}/backend-api", server.uri())); + server +} + #[tokio::test] async fn remote_voice_start_routes_v3_offer_and_effective_preference() -> Result<()> { use super::session_lifecycle_requests::HistoryCapabilities; @@ -100,11 +116,26 @@ async fn remote_voice_start_routes_v3_offer_and_effective_preference() -> Result Ok(()) } +async fn next_call_result( + events: &mut tokio::sync::mpsc::UnboundedReceiver, +) -> Result { + tokio::time::timeout(std::time::Duration::from_secs(10), async { + while let Some(event) = events.recv().await { + if matches!(event, AppEvent::RealtimeCallCreated { .. }) { + return Ok(event); + } + } + color_eyre::eyre::bail!("app event channel closed during call startup") + }) + .await? +} + async fn check_remote_voice_start( capabilities: super::session_lifecycle_requests::HistoryCapabilities, expected_voice: Option<&str>, ) -> Result<()> { - let (mut app, _events, _ops) = make_test_app_with_channels().await; + let (mut app, mut events, _ops) = make_test_app_with_channels().await; + let calls = mock_voice_calls(&mut app.config).await; let (mut app_server, requests, proxy) = super::session_lifecycle_requests::start_recording_app_server_with_realtime_speech( &app.config, @@ -116,10 +147,10 @@ async fn check_remote_voice_start( codex_config::LoaderOverrides::default(), ) .await?; - let thread_id = ThreadId::new(); + let started = app_server.start_thread(&app.config).await?; + let thread_id = started.session.thread_id; app.active_thread_id = Some(thread_id); - app.chat_widget - .handle_thread_session_quiet(test_thread_session(thread_id, app.config.cwd.to_path_buf())); + app.chat_widget.handle_thread_session_quiet(started.session); let mut tui = crate::tui::test_support::make_test_tui()?; Box::pin(app.handle_event( &mut tui, @@ -155,19 +186,43 @@ async fn check_remote_voice_start( app.config.realtime.voice = Some(codex_protocol::protocol::RealtimeVoice::Maple); app.chat_widget .on_realtime_voice_saved(codex_protocol::protocol::RealtimeVoice::Maple); + crate::chatwidget::start_voice_attempt_for_test( + &mut app.chat_widget, + thread_id, + /*attempt_id*/ 1, + ); Box::pin(app.handle_event( &mut tui, &mut app_server, AppEvent::CodexOp(Op::RealtimeConversationStart { thread_id, + attempt_id: 1, offer_sdp: String::from("v=0\r\n").into(), }), )) .await?; + Box::pin(app.handle_event( + &mut tui, + &mut app_server, + next_call_result(&mut events).await?, + )) + .await?; + let calls = calls.received_requests().await.unwrap(); + assert_eq!(calls.len(), 1); + assert_eq!(calls[0].headers["thread-id"], thread_id.to_string()); + assert_eq!(calls[0].headers["session-id"], thread_id.to_string()); + let body = calls[0].body_json::()?; + assert_eq!(body["sdp"], "v=0\r\n"); + let default_voice = codex_protocol::protocol::RealtimeVoicesList::builtin().default_v1; + assert_eq!( + body["session"]["audio"]["output"]["voice"], + expected_voice.unwrap_or(default_voice.wire_name()) + ); let starts = recorded_params(&requests, "thread/realtime/start"); assert_eq!(starts.len(), 1); let start = &starts[0]; + assert_eq!(start["realtimeSessionId"], thread_id.to_string()); assert_eq!( serde_json::json!({ "threadId": start["threadId"], @@ -184,8 +239,8 @@ async fn check_remote_voice_start( "includeStartupContext": false, "outputModality": "audio", "version": "v3", - "voice": expected_voice, - "transport": {"type": "webrtc", "sdp": "v=0\r\n"}, + "voice": null, + "transport": {"type": "existingCall", "callId": "rtc_tui"}, }) ); app_server.shutdown().await?; @@ -193,6 +248,109 @@ async fn check_remote_voice_start( Ok(()) } +#[tokio::test] +async fn voice_cancel_restart_only_attaches_the_new_attempt() -> Result<()> { + // Hold the provider response or the resulting UI event across a same-thread restart. + for cancel_before_response in [true, false] { + let (mut app, mut events, mut ops) = make_test_app_with_channels().await; + let (request_tx, mut requests) = tokio::sync::mpsc::unbounded_channel(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + app.config.experimental_realtime_webrtc_call_base_url = + Some(format!("http://{}/backend-api", listener.local_addr()?)); + let provider = axum::Router::new().route( + "/backend-api/realtime/calls", + axum::routing::post(move || { + let request_tx = request_tx.clone(); + async move { + let (tx, rx) = tokio::sync::oneshot::channel::<&'static str>(); + request_tx.send(tx).unwrap(); + ( + axum::http::StatusCode::CREATED, + [("Location", rx.await.unwrap())], + "v=answer", + ) + } + }), + ); + let provider = tokio::spawn(async move { axum::serve(listener, provider).await }); + let (mut server, rpc, proxy) = start_recording_realtime_speech_app_server( + &app.config, + RealtimeRequestBehavior::AcceptStart, + ) + .await?; + let started = server.start_thread(&app.config).await?; + let thread_id = started.session.thread_id; + app.active_thread_id = Some(thread_id); + app.chat_widget.handle_thread_session_quiet(started.session); + let mut tui = crate::tui::test_support::make_test_tui()?; + let mut start = async |app: &mut App, + tui: &mut crate::tui::Tui, + server: &mut AppServerSession, + attempt_id| { + crate::chatwidget::start_voice_attempt_for_test( + &mut app.chat_widget, + thread_id, + attempt_id, + ); + let command = AppEvent::CodexOp(Op::RealtimeConversationStart { + thread_id, + attempt_id, + offer_sdp: String::from("v=0").into(), + }); + tokio::time::timeout( + std::time::Duration::from_secs(10), + Box::pin(app.handle_event(tui, server, command)), + ) + .await??; + Ok::<_, color_eyre::Report>( + tokio::time::timeout(std::time::Duration::from_secs(10), requests.recv()) + .await? + .unwrap(), + ) + }; + let first = start(&mut app, &mut tui, &mut server, /*attempt_id*/ 1).await?; + let (pending, queued) = if cancel_before_response { + (Some(first), None) + } else { + first.send("/realtime/calls/rtc_old").unwrap(); + (None, Some(next_call_result(&mut events).await?)) + }; + Box::pin(app.handle_event( + &mut tui, + &mut server, + AppEvent::VoiceControl { + thread_id: Some(thread_id), + control: crate::app_event::VoiceControl::Stop, + }, + )) + .await?; + assert_eq!(app.voice_owner_thread_id(), None); + assert!( + !std::iter::from_fn(|| ops.try_recv().ok()) + .any(|op| matches!(op, Op::RealtimeConversationStop { .. })), + "Core has no call to stop yet", + ); + + let second = start(&mut app, &mut tui, &mut server, /*attempt_id*/ 2).await?; + if let Some(response) = pending { + let _ = response.send("/realtime/calls/rtc_old"); + } + if let Some(stale) = queued { + Box::pin(app.handle_event(&mut tui, &mut server, stale)).await?; + } + second.send("/realtime/calls/rtc_new").unwrap(); + Box::pin(app.handle_event(&mut tui, &mut server, next_call_result(&mut events).await?)) + .await?; + let starts = recorded_params(&rpc, "thread/realtime/start"); + assert_eq!(starts.len(), 1); + assert_eq!(starts[0]["transport"]["callId"], "rtc_new"); + server.shutdown().await?; + proxy.await??; + provider.abort(); + } + Ok(()) +} + #[tokio::test] async fn shutdown_stops_backend_voice_once_through_app_server() -> Result<()> { let (mut app, _events, _ops) = make_test_app_with_channels().await; @@ -1788,21 +1946,31 @@ async fn rejected_voice_setting_preserves_current_selection() -> Result<()> { #[tokio::test] async fn remote_voice_config_read_failure_does_not_start_with_local_voice() -> Result<()> { - let (mut app, _events, _ops) = make_test_app_with_channels().await; + let (mut app, mut events, _ops) = make_test_app_with_channels().await; let (mut app_server, requests, proxy) = start_recording_remote_app_server(&app.config).await?; std::fs::write(app.config.codex_home.join("config.toml"), "realtime = [")?; let thread_id = ThreadId::new(); - let error = app - .try_submit_active_thread_op_via_app_server( - &mut app_server, + crate::chatwidget::start_voice_attempt_for_test( + &mut app.chat_widget, + thread_id, + /*attempt_id*/ 1, + ); + app.try_submit_active_thread_op_via_app_server( + &mut app_server, + thread_id, + &AppCommand::RealtimeConversationStart { thread_id, - &AppCommand::RealtimeConversationStart { - thread_id, - offer_sdp: String::from("v=0\r\n").into(), - }, - ) - .await - .expect_err("invalid server config must not fall back to local voice"); + attempt_id: 1, + offer_sdp: String::from("v=0\r\n").into(), + }, + ) + .await?; + let AppEvent::RealtimeCallCreated { + result: Err(error), .. + } = next_call_result(&mut events).await? + else { + panic!("invalid server config must not fall back to local voice"); + }; assert!(error.to_string().contains("config/read")); assert!(recorded_params(&requests, "thread/realtime/start").is_empty()); app_server.shutdown().await?; @@ -1813,7 +1981,8 @@ async fn remote_voice_config_read_failure_does_not_start_with_local_voice() -> R #[tokio::test] async fn embedded_voice_settings_follow_project_after_thread_switch() -> Result<()> { use codex_protocol::protocol::RealtimeVoice; - let (mut app, _events, _ops) = make_test_app_with_channels().await; + let (mut app, mut events, _ops) = make_test_app_with_channels().await; + let calls = mock_voice_calls(&mut app.config).await; let projects = tempfile::tempdir()?; for voice in [RealtimeVoice::Maple, RealtimeVoice::Sol] { let cwd = projects.path().join(voice.wire_name()); @@ -1836,21 +2005,23 @@ async fn embedded_voice_settings_follow_project_after_thread_switch() -> Result< ) .await?; let mut tui = crate::tui::test_support::make_test_tui()?; - for voice in [ - RealtimeVoice::Maple, - RealtimeVoice::Sol, - RealtimeVoice::Maple, - ] { - let thread_id = ThreadId::new(); + let mut threads = Vec::new(); + for voice in [RealtimeVoice::Maple, RealtimeVoice::Sol] { let cwd = projects.path().join(voice.wire_name()); + let config = app.rebuild_config_for_cwd(cwd).await?; + let started = server.start_thread(&config).await?; + let thread_id = started.session.thread_id; app.thread_event_channels.insert( thread_id, ThreadEventChannel::new_with_session( THREAD_EVENT_CHANNEL_CAPACITY, - test_thread_session(thread_id, cwd), + started.session, Vec::new(), ), ); + threads.push((thread_id, voice)); + } + for (thread_id, voice) in [threads[0], threads[1], threads[0]] { Box::pin(app.select_agent_thread(&mut tui, &mut server, thread_id)).await?; app.open_realtime_voices(&server).await; let popup = crate::chatwidget::tests::helpers::render_bottom_popup( @@ -1861,20 +2032,35 @@ async fn embedded_voice_settings_follow_project_after_thread_switch() -> Result< popup.contains(&format!("{} (current)", voice.wire_name())), "{popup}" ); + crate::chatwidget::start_voice_attempt_for_test( + &mut app.chat_widget, + thread_id, + /*attempt_id*/ 1, + ); app.try_submit_active_thread_op_via_app_server( &mut server, thread_id, &AppCommand::RealtimeConversationStart { thread_id, + attempt_id: 1, offer_sdp: String::from("v=0\r\n").into(), }, ) .await?; + Box::pin(app.handle_event(&mut tui, &mut server, next_call_result(&mut events).await?)) + .await?; + app.chat_widget.reset_realtime_conversation(); + let calls = calls.received_requests().await.unwrap(); + let body = calls.last().unwrap().body_json::()?; + assert_eq!( + body["session"]["audio"]["output"]["voice"], + serde_json::json!(voice) + ); assert_eq!( recorded_params(&requests, "thread/realtime/start") .last() - .unwrap()["voice"], - serde_json::json!(voice), + .unwrap()["transport"]["callId"], + "rtc_tui" ); } server.shutdown().await?; @@ -1974,9 +2160,10 @@ async fn remote_voice_catalog_success_and_failure() -> Result<()> { app.config.cwd.to_path_buf(), )); app.active_thread_id = Some(thread_id); - crate::chatwidget::activate_voice_for_thread( + crate::chatwidget::start_voice_attempt_for_test( &mut app.chat_widget, thread_id, + /*attempt_id*/ 1, ); assert!(app.chat_widget.may_receive_realtime_transcripts()); let mut tui = crate::tui::test_support::make_test_tui()?; @@ -1985,10 +2172,17 @@ async fn remote_voice_catalog_success_and_failure() -> Result<()> { &mut server, AppEvent::CodexOp(Op::RealtimeConversationStart { thread_id, + attempt_id: 1, offer_sdp: String::from("v=0\r\n").into(), }), )) .await?; + Box::pin(app.handle_event( + &mut tui, + &mut server, + next_call_result(&mut events).await?, + )) + .await?; assert!(!app.chat_widget.may_receive_realtime_transcripts()); assert!(recorded_params(&requests, "thread/realtime/start").is_empty()); } diff --git a/codex-rs/tui/src/app/tests/realtime_start.rs b/codex-rs/tui/src/app/tests/realtime_start.rs index 8c9e6262d3..df9178b683 100644 --- a/codex-rs/tui/src/app/tests/realtime_start.rs +++ b/codex-rs/tui/src/app/tests/realtime_start.rs @@ -33,6 +33,7 @@ async fn rejected_voice_start_does_not_leave_local_capture_active() -> Result<() &mut app_server, AppEvent::CodexOp(Op::RealtimeConversationStart { thread_id, + attempt_id: 1, offer_sdp: String::from("v=0\r\n").into(), }), )) diff --git a/codex-rs/tui/src/app/thread_routing.rs b/codex-rs/tui/src/app/thread_routing.rs index dec69188a8..abe4441d4b 100644 --- a/codex-rs/tui/src/app/thread_routing.rs +++ b/codex-rs/tui/src/app/thread_routing.rs @@ -897,26 +897,43 @@ impl App { } AppCommand::RealtimeConversationStart { thread_id: realtime_thread_id, + attempt_id, offer_sdp, } => { - if *realtime_thread_id != thread_id { + let attempt_id = *attempt_id; + if *realtime_thread_id != thread_id + || !self + .chat_widget + .is_starting_realtime_attempt(thread_id, attempt_id) + { return Ok(true); } - let model = self - .chat_widget - .config_ref() - .experimental_realtime_ws_model - .clone(); - let voices = self.realtime_voices(app_server).await?; - let voice = self.effective_realtime_voice(app_server, &voices).await?; - app_server - .thread_realtime_start( - *realtime_thread_id, - String::from(offer_sdp.clone()), - model, - voice, - ) - .await?; + let registration = self.chat_widget.begin_call_negotiation(); + let mut server = app_server.request_view(); + let config = self.config.clone(); + let cwd = self.chat_widget.config_ref().cwd.clone(); + let offer = String::from(offer_sdp.clone()); + let tx = self.app_event_tx.clone(); + tokio::spawn(async move { + let create = async { + let voices = Self::realtime_voices(&server).await?; + let voice = Self::effective_realtime_voice(&server, &cwd, &voices) + .await? + .unwrap_or(voices.default_v1); + let thread = server + .thread_read(thread_id, /*include_turns*/ false) + .await?; + crate::realtime_call::create(&config, &thread, offer, voice).await + }; + if let Ok(result) = futures::future::Abortable::new(create, registration).await + { + tx.send(AppEvent::RealtimeCallCreated { + thread_id, + attempt_id, + result, + }); + } + }); Ok(true) } AppCommand::RealtimeConversationStop { diff --git a/codex-rs/tui/src/app_command.rs b/codex-rs/tui/src/app_command.rs index 82034eea3b..fe69981dd0 100644 --- a/codex-rs/tui/src/app_command.rs +++ b/codex-rs/tui/src/app_command.rs @@ -25,29 +25,29 @@ use serde::Serializer; use serde_json::Value; /// Keeps ICE credentials out of command diagnostics and session recordings. -/// Convert back to a string only when sending the offer to app-server signaling. +/// Convert back to a string only at the media signaling boundary. #[derive(Clone, PartialEq, Eq)] -pub(crate) struct RealtimeOfferSdp(String); +pub(crate) struct RealtimeSdp(String); -impl From for RealtimeOfferSdp { +impl From for RealtimeSdp { fn from(value: String) -> Self { Self(value) } } -impl From for String { - fn from(value: RealtimeOfferSdp) -> Self { +impl From for String { + fn from(value: RealtimeSdp) -> Self { value.0 } } -impl fmt::Debug for RealtimeOfferSdp { +impl fmt::Debug for RealtimeSdp { fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { formatter.write_str("") } } -impl Serialize for RealtimeOfferSdp { +impl Serialize for RealtimeSdp { fn serialize(&self, serializer: S) -> Result where S: Serializer, @@ -113,7 +113,8 @@ pub(crate) enum AppCommand { CleanBackgroundTerminals, RealtimeConversationStart { thread_id: ThreadId, - offer_sdp: RealtimeOfferSdp, + attempt_id: u64, + offer_sdp: RealtimeSdp, }, RealtimeConversationStop { thread_id: ThreadId, diff --git a/codex-rs/tui/src/app_command_tests.rs b/codex-rs/tui/src/app_command_tests.rs index c765205c62..f584be7b65 100644 --- a/codex-rs/tui/src/app_command_tests.rs +++ b/codex-rs/tui/src/app_command_tests.rs @@ -13,18 +13,22 @@ fn realtime_offer_is_redacted_in_diagnostics_and_preserved_for_signaling() { "v=0\r\na=ice-ufrag:private-user\r\na=ice-pwd:private-credential\r\n".to_string(); let command = AppCommand::RealtimeConversationStart { thread_id, + attempt_id: 1, offer_sdp: original_offer.clone().into(), }; assert_eq!( format!("{command:?}"), - format!("RealtimeConversationStart {{ thread_id: {thread_id:?}, offer_sdp: }}") + format!( + "RealtimeConversationStart {{ thread_id: {thread_id:?}, attempt_id: 1, offer_sdp: }}" + ) ); assert_eq!( serde_json::to_value(&command).expect("command diagnostics serialize"), json!({ "RealtimeConversationStart": { "thread_id": thread_id, + "attempt_id": 1, "offer_sdp": "" } }) diff --git a/codex-rs/tui/src/app_event.rs b/codex-rs/tui/src/app_event.rs index c5a017d7f2..f900fa57fb 100644 --- a/codex-rs/tui/src/app_event.rs +++ b/codex-rs/tui/src/app_event.rs @@ -1435,6 +1435,13 @@ pub(crate) enum AppEvent { result: Result, }, + /// Provider negotiation completed for this local voice startup attempt. + RealtimeCallCreated { + thread_id: ThreadId, + attempt_id: u64, + result: color_eyre::Result, + }, + /// Result of establishing the WebRTC connection for an active voice attempt. RealtimeWebrtcConnected { thread_id: ThreadId, diff --git a/codex-rs/tui/src/app_server_session/realtime.rs b/codex-rs/tui/src/app_server_session/realtime.rs index 14f62f7a13..721dcdc3b4 100644 --- a/codex-rs/tui/src/app_server_session/realtime.rs +++ b/codex-rs/tui/src/app_server_session/realtime.rs @@ -12,7 +12,6 @@ use codex_app_server_protocol::ThreadRealtimeStopResponse; use codex_protocol::ThreadId; use codex_protocol::protocol::RealtimeConversationVersion; use codex_protocol::protocol::RealtimeOutputModality; -use codex_protocol::protocol::RealtimeVoice; use color_eyre::eyre::Result; use color_eyre::eyre::WrapErr; @@ -20,9 +19,7 @@ impl AppServerSession { pub(crate) async fn thread_realtime_start( &mut self, thread_id: ThreadId, - offer_sdp: String, - model: Option, - voice: Option, + call_id: String, ) -> Result<()> { let request_id = self.next_request_id(); let _: ThreadRealtimeStartResponse = self @@ -39,18 +36,21 @@ impl AppServerSession { codex_response_handoff_mode: None, backend_reasoning_status: false, codex_response_handoff_channel_prefixes: None, - model, + model: None, output_modality: RealtimeOutputModality::Audio, include_startup_context: Some(false), initial_items: None, realtime_start_instructions: None, realtime_end_instructions: None, prompt: None, - realtime_session_id: None, + realtime_session_id: Some(thread_id.to_string()), voice_session_id: None, - transport: Some(ThreadRealtimeStartTransport::Webrtc { sdp: offer_sdp }), + transport: Some(ThreadRealtimeStartTransport::ExistingCall { + // The TUI owns the peer connection and has already negotiated this call. + call_id, + }), version: Some(RealtimeConversationVersion::V3), - voice, + voice: None, }, }) .await diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index e2243343b9..9b2da49216 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -376,6 +376,8 @@ pub(crate) use realtime::realtime_delegation_input; pub(crate) use realtime::tests::activate_voice_for_thread; #[cfg(test)] pub(crate) use realtime::tests::commit_realtime_history_events; +#[cfg(test)] +pub(crate) use realtime::tests::start_voice_attempt_for_test; mod reasoning_shortcuts; use self::realtime::RealtimeConversationUiState; mod rendering; @@ -2144,13 +2146,13 @@ fn has_websocket_timing_metrics(summary: RuntimeMetricsSummary) -> bool { impl Drop for ChatWidget { fn drop(&mut self) { - if self.realtime_conversation.handle.is_some() - && let Some(thread_id) = self.thread_id + let had_live_peer = self.realtime_conversation.handle.is_some(); + if let Some(thread_id) = self.reset_realtime_conversation() + && had_live_peer { self.app_event_tx .send(AppEvent::StopRealtimeConversation { thread_id }); } - self.reset_realtime_conversation(); self.stop_rate_limit_poller(); } } diff --git a/codex-rs/tui/src/chatwidget/realtime.rs b/codex-rs/tui/src/chatwidget/realtime.rs index e5dd41e650..cdffddc77e 100644 --- a/codex-rs/tui/src/chatwidget/realtime.rs +++ b/codex-rs/tui/src/chatwidget/realtime.rs @@ -377,7 +377,10 @@ impl ChatWidget { // Stopping may prevent final transcript events from arriving. self.finish_realtime_partial_transcripts(); - if let Some(handle) = self.realtime_conversation.handle.take() { + if self.realtime_conversation.startup_abort.is_some() { + // Core has no attachment to stop until the provider answer is accepted. + self.reset_realtime_conversation(); + } else if let Some(handle) = self.realtime_conversation.handle.take() { handle.close(); self.realtime_conversation.phase = RealtimeConversationPhase::Stopping; self.refresh_terminal_title(); @@ -424,7 +427,6 @@ impl ChatWidget { return; } }; - self.realtime_conversation.startup_abort = None; if self.realtime_conversation.microphone_muted && let Err(error) = offer.handle.set_microphone_muted(/*muted*/ true) { @@ -449,6 +451,7 @@ impl ChatWidget { .schedule_frame_in(MICROPHONE_METER_INTERVAL); if !self.submit_op(AppCommand::RealtimeConversationStart { thread_id, + attempt_id, offer_sdp: offer.offer_sdp.into(), }) { self.record_realtime_failure(RealtimeFailureCause::LocalRequest); @@ -456,7 +459,28 @@ impl ChatWidget { } } - pub(super) fn on_realtime_conversation_sdp(&mut self, answer_sdp: String) { + pub(crate) fn is_starting_realtime_attempt( + &self, + thread_id: ThreadId, + attempt_id: u64, + ) -> bool { + self.realtime_conversation.thread_id == Some(thread_id) + && self.realtime_conversation.attempt_id == attempt_id + && self.realtime_conversation.phase == RealtimeConversationPhase::Starting + } + + pub(crate) fn begin_call_negotiation(&mut self) -> futures::future::AbortRegistration { + let (abort, registration) = AbortHandle::new_pair(); + self.realtime_conversation.startup_abort = Some(abort); + registration + } + + pub(crate) fn accept_realtime_call(&mut self, answer: crate::app_command::RealtimeSdp) { + self.realtime_conversation.startup_abort = None; + self.on_realtime_conversation_sdp(answer.into()); + } + + pub(crate) fn on_realtime_conversation_sdp(&mut self, answer_sdp: String) { if self.realtime_conversation.phase != RealtimeConversationPhase::Starting { return; } @@ -1866,7 +1890,8 @@ impl ChatWidget { } // The helper may already have exited while app-server still owns the // voice session. An acknowledged backend start still needs a stop RPC. - let backend_thread_id = if self.realtime_conversation.handle.is_some() + let backend_thread_id = if (self.realtime_conversation.handle.is_some() + && self.realtime_conversation.startup_abort.is_none()) || self.realtime_conversation.backend_started { self.realtime_conversation.thread_id diff --git a/codex-rs/tui/src/chatwidget/realtime_tests.rs b/codex-rs/tui/src/chatwidget/realtime_tests.rs index ec9ceb181b..e75da4ff47 100644 --- a/codex-rs/tui/src/chatwidget/realtime_tests.rs +++ b/codex-rs/tui/src/chatwidget/realtime_tests.rs @@ -62,6 +62,16 @@ fn activate_voice(chat: &mut ChatWidget) -> ThreadId { thread_id } +pub(crate) fn start_voice_attempt_for_test( + chat: &mut ChatWidget, + thread_id: ThreadId, + attempt_id: u64, +) { + chat.realtime_conversation.thread_id = Some(thread_id); + chat.realtime_conversation.attempt_id = attempt_id; + chat.realtime_conversation.phase = RealtimeConversationPhase::Starting; +} + pub(crate) fn activate_voice_for_thread(chat: &mut ChatWidget, thread_id: ThreadId) { chat.thread_id = Some(thread_id); chat.realtime_conversation.phase = RealtimeConversationPhase::Active; diff --git a/codex-rs/tui/src/lib.rs b/codex-rs/tui/src/lib.rs index 8b3608cf9c..c3abb00935 100644 --- a/codex-rs/tui/src/lib.rs +++ b/codex-rs/tui/src/lib.rs @@ -141,6 +141,7 @@ mod image_decode; mod markdown_copy; mod permission_discovery; mod pets; +mod realtime_call; mod security_setup; mod worktree_browser; pub use custom_terminal::Terminal; diff --git a/codex-rs/tui/src/realtime_call.rs b/codex-rs/tui/src/realtime_call.rs new file mode 100644 index 0000000000..37bfbee368 --- /dev/null +++ b/codex-rs/tui/src/realtime_call.rs @@ -0,0 +1,98 @@ +//! Negotiate the TUI's media connection using the local user's provider and credentials. + +use crate::app_command::RealtimeSdp; +use crate::legacy_core::config::Config; +use codex_api::RealtimeCallClient; +use codex_api::RealtimeEventParser; +use codex_api::RealtimeSessionConfig; +use codex_api::RealtimeSessionMode; +use codex_api::ReqwestTransport; +use codex_api::build_realtime_thread_headers; +use codex_app_server_protocol::Thread; +use codex_config::config_toml::RealtimeWsMode; +use codex_http_client::ClientRouteClass; +use codex_login::default_client::ClientRedirectPolicy; +use codex_login::default_client::add_originator_header; +use codex_login::default_client::create_client_for_route; +use codex_login::default_client::originator; +use codex_model_provider::create_model_provider_from_local_config; +use codex_model_provider::prepare_provider_request; +use codex_prompts::prepare_realtime_backend_prompt; +use codex_protocol::protocol::RealtimeOutputModality; +use codex_protocol::protocol::RealtimeVoice; +use color_eyre::eyre::Result; +use color_eyre::eyre::ensure; +use http::HeaderValue; + +#[derive(Debug)] +pub(crate) struct CreatedCall { + pub(crate) call_id: String, + pub(crate) answer_sdp: RealtimeSdp, +} + +pub(crate) async fn create( + config: &Config, + thread: &Thread, + offer: String, + voice: RealtimeVoice, +) -> Result { + ensure!( + config.realtime.session_type == RealtimeWsMode::Conversational, + "AVAS realtime calls require conversational realtime" + ); + let provider = + create_model_provider_from_local_config(config.model_provider.clone(), config).await?; + let (mut api_provider, resolved_auth) = + prepare_provider_request(provider.as_ref(), provider.info().auth.is_none(), |_| { + provider.request_provider_and_auth(/*scope*/ None) + }) + .await?; + if let Some(url) = &config.experimental_realtime_webrtc_call_base_url { + api_provider.base_url.clone_from(url); + } + let client = create_client_for_route( + &config.http_client_factory(), + &api_provider.url_for_path("realtime/calls"), + ClientRouteClass::Api, + ClientRedirectPolicy::Reject, + )?; + let id = thread.id.clone(); + let mut headers = build_realtime_thread_headers( + thread.session_id.clone(), + id.clone(), + thread.thread_source.clone().map(Into::into), + ); + headers.insert("openai-alpha", HeaderValue::from_static("quicksilver=v2")); + headers.insert("x-session-id", HeaderValue::from_str(&id)?); + add_originator_header(&mut headers, &originator().value); + let session = RealtimeSessionConfig { + instructions: prepare_realtime_backend_prompt( + /*prompt*/ None, + config.experimental_realtime_ws_backend_prompt.clone(), + ), + initial_items: Vec::new(), + delegation_ack_filler: None, + model: Some( + config + .experimental_realtime_ws_model + .clone() + .unwrap_or_else(|| "gpt-live-1-codex".to_string()), + ), + session_id: Some(id), + event_parser: RealtimeEventParser::FramelessBidi, + session_mode: RealtimeSessionMode::Conversational, + output_modality: RealtimeOutputModality::Audio, + voice, + }; + let call = RealtimeCallClient::new( + ReqwestTransport::from_http_client(client), + api_provider, + resolved_auth.auth, + ) + .create_with_session_and_headers(offer, session, headers) + .await?; + Ok(CreatedCall { + call_id: call.call_id, + answer_sdp: call.sdp.into(), + }) +}