From e1d0ef995f07530d2063e0ce51259b8fce1a82bf Mon Sep 17 00:00:00 2001 From: guinness-oai Date: Wed, 2 Sep 2026 21:04:25 +0000 Subject: [PATCH] Make app-server realtime sessions always available (#42377) ## What changed - Remove the per-thread `RealtimeConversation` feature check when starting a realtime session. - Retain `features.realtime_conversation` as a removed compatibility setting that has no effect, even when set to `false`. - Document the always-available behavior for `thread/realtime/start`. ## Testing - Cover realtime startup on forked and cold-resumed threads with the legacy setting absent, disabled, or enabled. GitOrigin-RevId: d9cd9aa3898d1c2efb9414a8f40c86be8951e210 --- codex-rs/app-server/README.md | 2 + .../src/request_processors/turn_processor.rs | 6 - .../tests/suite/v2/realtime_conversation.rs | 165 ++++++++++-------- .../app-server/tests/suite/v2/turn_start.rs | 1 - codex-rs/features/src/lib.rs | 4 +- 5 files changed, 95 insertions(+), 83 deletions(-) diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 7c46d1b327..c33ccbcc95 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -1189,6 +1189,8 @@ Use `thread/inject_items` to append prebuilt Responses API items to a loaded thr ### Example: Start realtime with WebRTC +Realtime sessions do not require a per-thread feature opt-in. The legacy `features.realtime_conversation` setting is accepted but has no effect, including when set to `false`. + Use `thread/realtime/start` with `transport.type: "webrtc"` when a browser or webview owns the `RTCPeerConnection` and app-server should create the server-side realtime session. The transport `sdp` must be the offer SDP produced by `RTCPeerConnection.createOffer()`, not a hand-written or minimal SDP string. The offer should include the media sections the client wants to negotiate. For the standard realtime UI flow, create the audio track/transceiver and the `oai-events` data channel before calling `createOffer()`: diff --git a/codex-rs/app-server/src/request_processors/turn_processor.rs b/codex-rs/app-server/src/request_processors/turn_processor.rs index 16aec22986..21403d1c11 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/codex-rs/app-server/src/request_processors/turn_processor.rs @@ -1148,12 +1148,6 @@ impl TurnRequestProcessor { Err(error) => return Err(error), } - if !thread.enabled(Feature::RealtimeConversation) { - return Err(invalid_request(format!( - "thread {thread_id} does not support realtime conversation" - ))); - } - Ok(Some((thread_id, thread))) } diff --git a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs index c9e0d181c0..a5d187e5df 100644 --- a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs +++ b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs @@ -11,6 +11,8 @@ use codex_app_server_protocol::ItemStartedNotification; use codex_app_server_protocol::JSONRPCError; use codex_app_server_protocol::LoginAccountResponse; use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ThreadForkParams; +use codex_app_server_protocol::ThreadForkResponse; use codex_app_server_protocol::ThreadHistoryMode; use codex_app_server_protocol::ThreadItem; use codex_app_server_protocol::ThreadRealtimeAppendAudioParams; @@ -41,6 +43,8 @@ use codex_app_server_protocol::ThreadRealtimeStopParams; use codex_app_server_protocol::ThreadRealtimeStopResponse; use codex_app_server_protocol::ThreadRealtimeTranscriptDeltaNotification; use codex_app_server_protocol::ThreadRealtimeTranscriptDoneNotification; +use codex_app_server_protocol::ThreadResumeParams; +use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadTimelineEntry; @@ -53,7 +57,6 @@ use codex_app_server_protocol::TurnStartedNotification; use codex_app_server_protocol::TurnSteerParams; use codex_app_server_protocol::TurnSteerResponse; use codex_app_server_protocol::UserInput as V2UserInput; -use codex_features::Feature; use codex_protocol::protocol::CodexResponseHandoffMode; use codex_protocol::protocol::ConversationTextRole; use codex_protocol::protocol::RealtimeConversationVersion; @@ -73,6 +76,7 @@ use serde::de::DeserializeOwned; use serde_json::Value; use serde_json::json; use std::collections::BTreeMap; +use std::collections::HashMap; use std::path::Path; use std::sync::Arc; use std::sync::Mutex; @@ -308,7 +312,6 @@ impl RealtimeE2eHarness { codex_home.path(), &main_loop_responses_server.uri(), realtime_server.uri(), - /*realtime_enabled*/ true, StartupContextConfig::Override("startup context"), realtime_version, sandbox, @@ -674,7 +677,6 @@ async fn realtime_conversation_streams_timeline_items() -> Result<()> { codex_home.path(), &responses_server.uri(), realtime_server.uri(), - /*realtime_enabled*/ true, StartupContextConfig::Generated, )?; @@ -872,7 +874,6 @@ async fn realtime_conversation_streams_v2_notifications() -> Result<()> { codex_home.path(), &responses_server.uri(), realtime_server.uri(), - /*realtime_enabled*/ true, StartupContextConfig::Generated, )?; @@ -1445,7 +1446,6 @@ async fn realtime_start_can_skip_startup_context() -> Result<()> { codex_home.path(), &responses_server.uri(), realtime_server.uri(), - /*realtime_enabled*/ true, StartupContextConfig::Generated, )?; @@ -1543,7 +1543,6 @@ async fn realtime_text_output_modality_requests_text_output_and_final_transcript codex_home.path(), &responses_server.uri(), realtime_server.uri(), - /*realtime_enabled*/ true, StartupContextConfig::Generated, )?; @@ -1652,7 +1651,6 @@ async fn realtime_list_voices_returns_supported_names() -> Result<()> { codex_home.path(), "http://127.0.0.1:1", "ws://127.0.0.1:1", - /*realtime_enabled*/ true, StartupContextConfig::Generated, )?; @@ -1722,7 +1720,6 @@ async fn realtime_conversation_stop_emits_closed_notification() -> Result<()> { codex_home.path(), &responses_server.uri(), realtime_server.uri(), - /*realtime_enabled*/ true, StartupContextConfig::Generated, )?; @@ -1930,7 +1927,6 @@ async fn realtime_webrtc_start_emits_sdp_notification() -> Result<()> { codex_home.path(), &responses_server.uri(), realtime_server.uri(), - /*realtime_enabled*/ true, StartupContextConfig::Override("startup context"), )?; @@ -3756,7 +3752,6 @@ async fn realtime_webrtc_start_surfaces_backend_error() -> Result<()> { codex_home.path(), &responses_server.uri(), realtime_server.uri(), - /*realtime_enabled*/ true, StartupContextConfig::Override("startup context"), )?; @@ -3812,70 +3807,100 @@ async fn realtime_webrtc_start_surfaces_backend_error() -> Result<()> { Ok(()) } +#[derive(Debug, Clone, Copy)] +enum RealtimeThreadLoad { + Fork, + ColdResume, +} + +#[test_case(RealtimeThreadLoad::Fork, None; "fork without opt in")] +#[test_case(RealtimeThreadLoad::ColdResume, None; "cold resume without opt in")] +#[test_case(RealtimeThreadLoad::Fork, Some(false); "fork with legacy disabled flag")] +#[test_case(RealtimeThreadLoad::ColdResume, Some(false); "cold resume with legacy disabled flag")] +#[test_case(RealtimeThreadLoad::Fork, Some(true); "fork with legacy enabled flag")] #[tokio::test] -async fn realtime_conversation_requires_feature_flag() -> Result<()> { +async fn realtime_starts_on_forked_and_resumed_threads( + load: RealtimeThreadLoad, + legacy_flag: Option, +) -> Result<()> { skip_if_no_network!(Ok(())); - let responses_server = create_mock_responses_server_sequence_unchecked(Vec::new()).await; - let realtime_server = start_websocket_server(vec![vec![]]).await; + let mut harness = RealtimeE2eHarness::new( + RealtimeTestVersion::V1, + main_loop_responses(vec![create_final_assistant_message_sse_response("Done")?]), + realtime_sideband(vec![open_realtime_sideband_connection(vec![vec![ + session_updated("sess_fork"), + ]])]), + ) + .await?; - let codex_home = TempDir::new()?; - create_config_toml( - codex_home.path(), - &responses_server.uri(), - realtime_server.uri(), - /*realtime_enabled*/ false, - StartupContextConfig::Generated, - )?; - - let mut mcp = TestAppServer::builder() - .with_codex_home(codex_home.path()) - .build_initialized_with_timeout(DEFAULT_TIMEOUT) - .await?; - - let thread_start_request_id = mcp - .send_thread_start_request_with_auto_env(ThreadStartParams::default()) - .await?; - let thread_start: ThreadStartResponse = - timeout(DEFAULT_TIMEOUT, mcp.read_response(thread_start_request_id)).await??; - - let start_request_id = mcp - .send_thread_realtime_start_request(ThreadRealtimeStartParams { - client_managed_handoffs: None, - delegation_ack_filler: None, - flush_transcript_tail_on_session_end: None, - codex_responses_as_items: None, - codex_response_item_prefix: None, - codex_response_handoff_mode: None, - codex_response_handoff_channel_prefixes: None, - thread_id: thread_start.thread.id.clone(), - model: None, - output_modality: RealtimeOutputModality::Audio, - include_startup_context: None, - initial_items: None, - realtime_start_instructions: None, - realtime_end_instructions: None, - prompt: Some(Some("backend prompt".to_string())), - realtime_session_id: None, - transport: None, - version: None, - voice: None, - }) - .await?; - let error = timeout( + // Persist ordinary text history before forking it into a Voice-capable thread. + timeout( DEFAULT_TIMEOUT, - mcp.read_stream_until_error_message(RequestId::Integer(start_request_id)), + harness + .mcp + .start_turn_and_wait_for_completion(TurnStartParams { + thread_id: harness.thread_id.clone(), + input: vec![V2UserInput::Text { + text: "hello".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }), ) .await??; - assert_invalid_request( - error, - format!( - "thread {} does not support realtime conversation", - thread_start.thread.id - ), - ); + let config = legacy_flag.map(|enabled| { + HashMap::from([("features.realtime_conversation".to_string(), json!(enabled))]) + }); + let fork_id = harness + .mcp + .send_thread_fork_request(ThreadForkParams { + thread_id: harness.thread_id.clone(), + config: config.clone(), + ..Default::default() + }) + .await?; + let fork: ThreadForkResponse = + timeout(DEFAULT_TIMEOUT, harness.mcp.read_response(fork_id)).await??; + assert_ne!(fork.thread.id, harness.thread_id); + harness.thread_id = fork.thread.id; - realtime_server.shutdown().await; + if let RealtimeThreadLoad::ColdResume = load { + timeout(DEFAULT_TIMEOUT, harness.mcp.shutdown_gracefully()).await??; + harness.mcp = TestAppServer::builder() + .with_codex_home(harness._codex_home.path()) + .build_initialized_with_timeout(DEFAULT_TIMEOUT) + .await?; + let resume_id = harness + .mcp + .send_thread_resume_request(ThreadResumeParams { + thread_id: harness.thread_id.clone(), + config, + ..Default::default() + }) + .await?; + let resumed: ThreadResumeResponse = + timeout(DEFAULT_TIMEOUT, harness.mcp.read_response(resume_id)).await??; + assert_eq!(resumed.thread.id, harness.thread_id); + } + + let started = harness.start_webrtc_realtime("v=offer\r\n").await?; + assert_eq!( + started, + StartedWebrtcRealtime { + started: ThreadRealtimeStartedNotification { + thread_id: harness.thread_id.clone(), + realtime_session_id: Some(harness.thread_id.clone()), + version: RealtimeConversationVersion::V1, + }, + sdp: ThreadRealtimeSdpNotification { + thread_id: harness.thread_id.clone(), + sdp: "v=answer\r\n".to_string(), + }, + } + ); + assert_v1_session_update(&harness.sideband_outbound_request(/*request_index*/ 0).await)?; + harness.shutdown().await; Ok(()) } @@ -4140,14 +4165,12 @@ fn create_config_toml( codex_home: &Path, responses_server_uri: &str, realtime_server_uri: &str, - realtime_enabled: bool, startup_context: StartupContextConfig<'_>, ) -> std::io::Result<()> { create_config_toml_with_realtime_version( codex_home, responses_server_uri, realtime_server_uri, - realtime_enabled, startup_context, RealtimeTestVersion::V2, RealtimeTestSandbox::ReadOnly, @@ -4158,7 +4181,6 @@ fn create_config_toml_with_realtime_version( codex_home: &Path, responses_server_uri: &str, realtime_server_uri: &str, - realtime_enabled: bool, startup_context: StartupContextConfig<'_>, realtime_version: RealtimeTestVersion, sandbox: RealtimeTestSandbox, @@ -4179,11 +4201,6 @@ fn create_config_toml_with_realtime_version( "experimental_realtime_ws_startup_context = {context:?}" )); } - config = if realtime_enabled { - config.enable_feature(Feature::RealtimeConversation) - } else { - config.disable_feature(Feature::RealtimeConversation) - }; config.write(codex_home) } diff --git a/codex-rs/app-server/tests/suite/v2/turn_start.rs b/codex-rs/app-server/tests/suite/v2/turn_start.rs index f38b5cddfe..6af12cea2e 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -4324,7 +4324,6 @@ async fn direct_input_to_multi_agent_v2_subagent_is_rejected( MockResponsesConfig::new(&server.uri()) .enable_feature(Feature::MultiAgentV2) .enable_feature(Feature::Goals) - .enable_feature(Feature::RealtimeConversation) .with_root_config(&format!("chatgpt_base_url = \"{}\"", server.uri())) .write(codex_home.path())?; write_models_cache(codex_home.path())?; diff --git a/codex-rs/features/src/lib.rs b/codex-rs/features/src/lib.rs index 1cf7523478..552792fdf8 100644 --- a/codex-rs/features/src/lib.rs +++ b/codex-rs/features/src/lib.rs @@ -330,7 +330,7 @@ pub enum Feature { FastMode, /// Enable explicitly requested model changes for later step captures. StepModelSwitching, - /// Enable experimental realtime voice conversation mode in the TUI. + /// Removed compatibility flag. Realtime sessions no longer require a per-thread opt-in. RealtimeConversation, /// Prevent idle system sleep while a turn is actively running. PreventIdleSleep, @@ -1612,7 +1612,7 @@ pub const FEATURES: &[FeatureSpec] = &[ FeatureSpec { id: Feature::RealtimeConversation, key: "realtime_conversation", - stage: Stage::UnderDevelopment, + stage: Stage::Removed, default_enabled: false, }, FeatureSpec {