diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index 047f8267fc..8a39fdc299 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/codex-rs/app-server-protocol/schema/json/ClientRequest.json @@ -4526,6 +4526,27 @@ ], "title": "WebrtcThreadRealtimeStartTransport", "type": "object" + }, + { + "properties": { + "callId": { + "description": "Identifier of a realtime call already created and negotiated by the client.", + "type": "string" + }, + "type": { + "enum": [ + "existingCall" + ], + "title": "ExistingCallThreadRealtimeStartTransportType", + "type": "string" + } + }, + "required": [ + "callId", + "type" + ], + "title": "ExistingCallThreadRealtimeStartTransport", + "type": "object" } ] }, diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index 3edb865667..04c3525d2e 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -21178,6 +21178,27 @@ ], "title": "WebrtcThreadRealtimeStartTransport", "type": "object" + }, + { + "properties": { + "callId": { + "description": "Identifier of a realtime call already created and negotiated by the client.", + "type": "string" + }, + "type": { + "enum": [ + "existingCall" + ], + "title": "ExistingCallThreadRealtimeStartTransportType", + "type": "string" + } + }, + "required": [ + "callId", + "type" + ], + "title": "ExistingCallThreadRealtimeStartTransport", + "type": "object" } ] }, diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index 032dfbc38a..6542337a86 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -18930,6 +18930,27 @@ ], "title": "WebrtcThreadRealtimeStartTransport", "type": "object" + }, + { + "properties": { + "callId": { + "description": "Identifier of a realtime call already created and negotiated by the client.", + "type": "string" + }, + "type": { + "enum": [ + "existingCall" + ], + "title": "ExistingCallThreadRealtimeStartTransportType", + "type": "string" + } + }, + "required": [ + "callId", + "type" + ], + "title": "ExistingCallThreadRealtimeStartTransport", + "type": "object" } ] }, diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst index a5690d7291..92acb98d4a 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index 2e37790d0b..9b92646d4c 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeStartTransport.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeStartTransport.ts index 339e1b1b17..eabc5eae85 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeStartTransport.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadRealtimeStartTransport.ts @@ -10,4 +10,8 @@ export type ThreadRealtimeStartTransport = { "type": "websocket" } | { "type": " * SDP offer generated by a WebRTC RTCPeerConnection after configuring audio and the * realtime events data channel. */ -sdp: string, }; +sdp: string, } | { "type": "existingCall", +/** + * Identifier of a realtime call already created and negotiated by the client. + */ +callId: string, }; diff --git a/codex-rs/app-server-protocol/src/protocol/v2/realtime.rs b/codex-rs/app-server-protocol/src/protocol/v2/realtime.rs index a53f001838..12355c96d3 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/realtime.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/realtime.rs @@ -155,6 +155,12 @@ pub enum ThreadRealtimeStartTransport { /// realtime events data channel. sdp: String, }, + ExistingCall { + /// Identifier of a realtime call already created and negotiated by the client. + #[serde(rename = "callId")] + #[ts(rename = "callId")] + call_id: String, + }, } /// EXPERIMENTAL - response for starting thread realtime. diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 0578eb560a..92e83d1309 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -1079,6 +1079,27 @@ Then send `offer.sdp` to app-server. Core uses `experimental_realtime_ws_backend } } ``` +Clients that create and negotiate the realtime call themselves can instead pass its call ID: + +```json +{ "method": "thread/realtime/start", "id": 41, "params": { + "threadId": "thr_123", + "outputModality": "audio", + "version": "v3", + "realtimeSessionId": "sess_123", + "transport": { "type": "existingCall", "callId": "rtc_123" } +} } +{ "id": 41, "result": {} } +``` + +The existing-call transport attaches Codex to the call over its sideband WebSocket without creating +another call or emitting `thread/realtime/sdp`. The client owns the SDP negotiation and the initial +realtime session configuration. Codex startup context is disabled by default for existing calls; +`includeStartupContext: true`, `prompt`, nonempty `initialItems`, `model`, `voice`, and +`delegationAckFiller` are rejected because they would change the client-owned session. Supply +`realtimeSessionId` when the upstream session ID is known; otherwise the +`thread/realtime/started` notification reports `realtimeSessionId: null`. + Omit `prompt` to use Codex's default realtime backend prompt. Send `prompt: null` or `prompt: ""` when the session should start without that default backend prompt. Pass `realtimeStartInstructions` to provide the developer instructions given to 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 7664364175..a7f96ed316 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/codex-rs/app-server/src/request_processors/turn_processor.rs @@ -1082,6 +1082,36 @@ impl TurnRequestProcessor { request_id: &ConnectionRequestId, params: ThreadRealtimeStartParams, ) -> Result, JSONRPCErrorError> { + let attaches_existing_call = matches!( + ¶ms.transport, + Some(ThreadRealtimeStartTransport::ExistingCall { .. }) + ); + if attaches_existing_call { + let unsupported_option = if params.include_startup_context == Some(true) { + Some("includeStartupContext") + } else if params.prompt.is_some() { + Some("prompt") + } else if params + .initial_items + .as_ref() + .is_some_and(|items| !items.is_empty()) + { + Some("initialItems") + } else if params.model.is_some() { + Some("model") + } else if params.voice.is_some() { + Some("voice") + } else if params.delegation_ack_filler.is_some() { + Some("delegationAckFiller") + } else { + None + }; + if let Some(option) = unsupported_option { + return Err(invalid_request(format!( + "existingCall transport does not support {option}" + ))); + } + } let Some((_, thread)) = self .prepare_realtime_conversation_thread(request_id, ¶ms.thread_id) .await? @@ -1104,7 +1134,9 @@ impl TurnRequestProcessor { .codex_response_handoff_channel_prefixes, model: params.model, output_modality: params.output_modality, - include_startup_context: params.include_startup_context.unwrap_or(true), + include_startup_context: params + .include_startup_context + .unwrap_or(!attaches_existing_call), initial_items: params .initial_items .unwrap_or_default() @@ -1125,6 +1157,9 @@ impl TurnRequestProcessor { ThreadRealtimeStartTransport::Webrtc { sdp } => { ConversationStartTransport::Webrtc { sdp } } + ThreadRealtimeStartTransport::ExistingCall { call_id } => { + ConversationStartTransport::ExistingCall { call_id } + } }), version: params.version, voice: params.voice, 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 9164a258a4..bb1b5bf175 100644 --- a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs +++ b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs @@ -68,6 +68,7 @@ use std::sync::Mutex; use std::sync::mpsc; use std::time::Duration; use tempfile::TempDir; +use test_case::test_case; use tokio::time::timeout; use wiremock::Match; use wiremock::Mock; @@ -1682,6 +1683,191 @@ async fn webrtc_v1_start_posts_offer_returns_sdp_and_joins_sideband() -> Result< Ok(()) } +#[test_case( + None, + None, + RealtimeConversationVersion::V1, + "/v1/realtime?intent=quicksilver&call_id=rtc_existing"; + "defaults to v1" +)] +#[test_case( + Some(RealtimeConversationVersion::V3), + Some("sess_client_owned"), + RealtimeConversationVersion::V3, + "/v1/live/rtc_existing"; + "supports v3" +)] +#[test_case( + Some(RealtimeConversationVersion::V2), + None, + RealtimeConversationVersion::V2, + ""; + "rejects v2" +)] +#[tokio::test] +async fn existing_call_attaches_without_reinitializing_the_client_session( + version: Option, + realtime_session_id: Option<&str>, + expected_version: RealtimeConversationVersion, + expected_handshake_uri: &str, +) -> Result<()> { + skip_if_no_network!(Ok(())); + + let mut harness = RealtimeE2eHarness::new( + RealtimeTestVersion::V1, + no_main_loop_responses(), + realtime_sideband(vec![open_realtime_sideband_connection(vec![vec![]])]), + ) + .await?; + let request_id = harness + .mcp + .send_thread_realtime_start_request(ThreadRealtimeStartParams { + thread_id: harness.thread_id.clone(), + client_managed_handoffs: None, + delegation_ack_filler: None, + flush_transcript_tail_on_session_end: None, + codex_response_item_prefix: None, + codex_response_handoff_mode: None, + codex_response_handoff_channel_prefixes: None, + codex_responses_as_items: None, + model: None, + output_modality: RealtimeOutputModality::Audio, + include_startup_context: None, + initial_items: None, + realtime_start_instructions: None, + realtime_end_instructions: None, + prompt: None, + realtime_session_id: realtime_session_id.map(str::to_string), + transport: Some(ThreadRealtimeStartTransport::ExistingCall { + call_id: "rtc_existing".to_string(), + }), + version, + voice: None, + }) + .await?; + let _: ThreadRealtimeStartResponse = + timeout(DEFAULT_TIMEOUT, harness.mcp.read_response(request_id)).await??; + if expected_version == RealtimeConversationVersion::V2 { + let error = harness + .read_notification::("thread/realtime/error") + .await?; + assert_eq!( + error.message, + "AVAS realtime calls require realtime v1 or v3" + ); + assert!(harness.realtime_server.handshakes().is_empty()); + harness.shutdown().await; + return Ok(()); + } + let started = harness + .read_notification::("thread/realtime/started") + .await?; + + assert_eq!( + started, + ThreadRealtimeStartedNotification { + thread_id: harness.thread_id.clone(), + realtime_session_id: realtime_session_id.map(str::to_string), + version: expected_version, + } + ); + assert_eq!( + harness.realtime_server.single_handshake().uri(), + expected_handshake_uri + ); + assert!( + harness.realtime_server.single_connection().is_empty(), + "attaching to an existing call must not overwrite the client session" + ); + assert!( + harness + .call_capture + .requests + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .is_empty(), + "an existing call must not issue a second call-create request" + ); + + harness.shutdown().await; + Ok(()) +} + +#[test_case("includeStartupContext"; "rejects startup context")] +#[test_case("prompt"; "rejects prompt")] +#[test_case("initialItems"; "rejects initial items")] +#[test_case("model"; "rejects model")] +#[test_case("voice"; "rejects voice")] +#[test_case("delegationAckFiller"; "rejects delegation acknowledgement filler")] +#[tokio::test] +async fn existing_call_rejects_client_owned_session_configuration(option: &str) -> Result<()> { + skip_if_no_network!(Ok(())); + + let mut harness = RealtimeE2eHarness::new( + RealtimeTestVersion::V1, + no_main_loop_responses(), + realtime_sideband(Vec::new()), + ) + .await?; + let mut params = ThreadRealtimeStartParams { + thread_id: harness.thread_id.clone(), + client_managed_handoffs: None, + delegation_ack_filler: None, + flush_transcript_tail_on_session_end: None, + codex_response_item_prefix: None, + codex_response_handoff_mode: None, + codex_response_handoff_channel_prefixes: None, + codex_responses_as_items: None, + model: None, + output_modality: RealtimeOutputModality::Audio, + include_startup_context: None, + initial_items: None, + realtime_start_instructions: None, + realtime_end_instructions: None, + prompt: None, + realtime_session_id: None, + transport: Some(ThreadRealtimeStartTransport::ExistingCall { + call_id: "rtc_existing".to_string(), + }), + version: Some(RealtimeConversationVersion::V3), + voice: None, + }; + match option { + "includeStartupContext" => params.include_startup_context = Some(true), + "prompt" => params.prompt = Some(Some("backend prompt".to_string())), + "initialItems" => { + params.initial_items = Some(vec![ThreadRealtimeInitialItem { + role: ConversationTextRole::User, + text: "client-owned history".to_string(), + }]); + } + "model" => params.model = Some("another-model".to_string()), + "voice" => params.voice = Some(RealtimeVoice::Cove), + "delegationAckFiller" => params.delegation_ack_filler = Some(true), + option => anyhow::bail!("unsupported test option: {option}"), + } + + let request_id = harness + .mcp + .send_thread_realtime_start_request(params) + .await?; + let error = timeout( + DEFAULT_TIMEOUT, + harness + .mcp + .read_stream_until_error_message(RequestId::Integer(request_id)), + ) + .await??; + assert_invalid_request( + error, + format!("existingCall transport does not support {option}"), + ); + assert!(harness.realtime_server.handshakes().is_empty()); + + harness.shutdown().await; + Ok(()) +} + #[tokio::test] async fn webrtc_v3_start_posts_live_session_and_joins_without_session_update() -> Result<()> { skip_if_no_network!(Ok(())); diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs index 4a7bd31b7c..7dd84e5366 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs @@ -767,6 +767,13 @@ pub struct RealtimeWebsocketClient { webrtc_sideband_base_url: String, } +#[derive(Clone, Copy)] +enum RealtimeSessionInitialization { + NewSession, + LegacyWebrtcSideband, + ExistingCall, +} + impl RealtimeWebsocketClient { pub fn new(provider: Provider) -> Self { Self { @@ -799,7 +806,7 @@ impl RealtimeWebsocketClient { config, extra_headers, default_headers, - /*initialize_session*/ true, + RealtimeSessionInitialization::NewSession, RealtimeTranscriptState::default(), ) .await @@ -812,6 +819,46 @@ impl RealtimeWebsocketClient { extra_headers: HeaderMap, default_headers: HeaderMap, transcript_state: RealtimeTranscriptState, + ) -> Result { + self.connect_sideband( + config, + call_id, + extra_headers, + default_headers, + RealtimeSessionInitialization::LegacyWebrtcSideband, + transcript_state, + ) + .await + } + + /// Attaches to a client-created call without overwriting its session configuration. + pub async fn connect_existing_call_sideband( + &self, + config: RealtimeSessionConfig, + call_id: &str, + extra_headers: HeaderMap, + default_headers: HeaderMap, + transcript_state: RealtimeTranscriptState, + ) -> Result { + self.connect_sideband( + config, + call_id, + extra_headers, + default_headers, + RealtimeSessionInitialization::ExistingCall, + transcript_state, + ) + .await + } + + async fn connect_sideband( + &self, + config: RealtimeSessionConfig, + call_id: &str, + extra_headers: HeaderMap, + default_headers: HeaderMap, + session_initialization: RealtimeSessionInitialization, + transcript_state: RealtimeTranscriptState, ) -> Result { // The WebRTC call already exists; this loop only retries joining its sideband control // socket. Once joined, the returned connection is the same reader/writer state that the @@ -823,6 +870,7 @@ impl RealtimeWebsocketClient { call_id, extra_headers.clone(), default_headers.clone(), + session_initialization, transcript_state.clone(), ) .await; @@ -854,6 +902,7 @@ impl RealtimeWebsocketClient { call_id: &str, extra_headers: HeaderMap, default_headers: HeaderMap, + session_initialization: RealtimeSessionInitialization, transcript_state: RealtimeTranscriptState, ) -> Result { // Keep the parser/session query shaping from standalone realtime while replacing the model @@ -864,7 +913,7 @@ impl RealtimeWebsocketClient { config, extra_headers, default_headers, - /*initialize_session*/ false, + session_initialization, transcript_state, ) .await @@ -891,7 +940,7 @@ impl RealtimeWebsocketClient { config: RealtimeSessionConfig, extra_headers: HeaderMap, default_headers: HeaderMap, - initialize_session: bool, + session_initialization: RealtimeSessionInitialization, transcript_state: RealtimeTranscriptState, ) -> Result { ensure_rustls_crypto_provider(); @@ -934,7 +983,14 @@ impl RealtimeWebsocketClient { config.event_parser, transcript_state, ); - if initialize_session || config.event_parser != RealtimeEventParser::FramelessBidi { + let initialize_session = match session_initialization { + RealtimeSessionInitialization::NewSession => true, + RealtimeSessionInitialization::LegacyWebrtcSideband => { + config.event_parser != RealtimeEventParser::FramelessBidi + } + RealtimeSessionInitialization::ExistingCall => false, + }; + if initialize_session { debug!( session_id = config.session_id.as_deref().unwrap_or(""), "realtime websocket sending session.update" @@ -951,7 +1007,11 @@ impl RealtimeWebsocketClient { ) .await?; } - if initialize_session && config.event_parser == RealtimeEventParser::FramelessBidi { + if matches!( + session_initialization, + RealtimeSessionInitialization::NewSession + ) && config.event_parser == RealtimeEventParser::FramelessBidi + { connection.events.wait_for_session_started().await?; } Ok(connection) @@ -1079,8 +1139,19 @@ fn websocket_url_from_api_url_for_call( )?; match event_parser { RealtimeEventParser::FramelessBidi => { - let path = format!("{}/{}", url.path().trim_end_matches('/'), call_id); - url.set_path(&path); + if matches!(call_id, "." | "..") { + return Err(ApiError::InvalidRequest { + message: format!("invalid realtime call id: {call_id}"), + }); + } + url.path_segments_mut() + .map_err(|()| { + ApiError::Stream( + "realtime sideband URL cannot contain path segments".to_string(), + ) + })? + .pop_if_empty() + .push(call_id); } RealtimeEventParser::V1 | RealtimeEventParser::RealtimeV2 => { url.query_pairs_mut().append_pair("call_id", call_id); @@ -2051,6 +2122,43 @@ mod tests { ); } + #[test] + fn frameless_websocket_url_encodes_call_id_as_one_path_segment() { + let url = websocket_url_from_api_url_for_call( + "https://example.com/proxy/v1/live?trace=1", + /*query_params*/ None, + RealtimeEventParser::FramelessBidi, + RealtimeSessionMode::Conversational, + "../../admin", + ) + .expect("build existing-call websocket url"); + + assert_eq!( + url.as_str(), + "wss://example.com/proxy/v1/live/..%2F..%2Fadmin?trace=1" + ); + } + + #[test] + fn frameless_websocket_url_rejects_dot_only_call_ids() { + for call_id in [".", ".."] { + let error = websocket_url_from_api_url_for_call( + "https://example.com/v1/live", + /*query_params*/ None, + RealtimeEventParser::FramelessBidi, + RealtimeSessionMode::Conversational, + call_id, + ) + .expect_err("dot-only call IDs must not target the parent sideband URL"); + + assert!(matches!( + error, + ApiError::InvalidRequest { message } + if message == format!("invalid realtime call id: {call_id}") + )); + } + } + #[test] fn webrtc_frameless_sideband_ignores_provider_base_url() { let client = RealtimeWebsocketClient::new(Provider { diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index f26d41955e..fa4a82086e 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -686,6 +686,20 @@ impl ModelClient { }) } + pub(crate) async fn realtime_sideband_headers( + &self, + mut extra_headers: ApiHeaderMap, + ) -> Result { + let client_setup = self.current_client_setup().await?; + if let Some(header_value) = self.generate_attestation_header_for().await { + extra_headers.insert(X_OAI_ATTESTATION_HEADER, header_value); + } + extra_headers.extend(sideband_websocket_auth_headers( + client_setup.api_auth.as_ref(), + )); + Ok(extra_headers) + } + /// Builds memory summaries for each provided normalized raw memory. /// /// This is a unary call (no streaming) to `/v1/memories/trace_summarize`. diff --git a/codex-rs/core/src/client_tests.rs b/codex-rs/core/src/client_tests.rs index 58f2f5f72e..a6908cfe44 100644 --- a/codex-rs/core/src/client_tests.rs +++ b/codex-rs/core/src/client_tests.rs @@ -974,6 +974,25 @@ async fn websocket_handshake_includes_attestation_for_chatgpt_codex_responses() assert_eq!(attestation_calls.load(Ordering::Relaxed), 1); } +#[tokio::test] +async fn existing_call_sideband_headers_include_attestation() { + let (model_client, attestation_calls) = + model_client_with_counting_attestation(/*include_attestation*/ true); + + let headers = model_client + .realtime_sideband_headers(http::HeaderMap::new()) + .await + .expect("existing call sideband headers should build"); + + assert_eq!( + headers + .get(crate::attestation::X_OAI_ATTESTATION_HEADER) + .and_then(|value| value.to_str().ok()), + Some("v1.header-1"), + ); + assert_eq!(attestation_calls.load(Ordering::Relaxed), 1); +} + #[tokio::test] async fn non_chatgpt_codex_endpoints_omit_attestation_generation() { let (model_client, attestation_calls) = diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index f9c96bea3b..e158c0d941 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -22,6 +22,7 @@ use codex_api::RealtimeEventParser; use codex_api::RealtimeSessionConfig; use codex_api::RealtimeSessionMode; use codex_api::RealtimeWebsocketClient; +use codex_api::RealtimeWebsocketConnection; use codex_api::RealtimeWebsocketEvents; use codex_api::RealtimeWebsocketWriter; use codex_api::build_session_headers; @@ -80,6 +81,7 @@ use tracing::info; use tracing::warn; mod bem; +mod existing_call; mod sideband; use self::bem::ChannelParser as BemChannelParser; @@ -478,6 +480,7 @@ struct RealtimeStart { session_config: RealtimeSessionConfig, model_client: ModelClient, sdp: Option, + existing_call_id: Option, } struct RealtimeStartOutput { @@ -550,6 +553,7 @@ impl RealtimeConversationManager { session_config, model_client, sdp, + existing_call_id, } = start; let event_parser = session_config.event_parser; let session_kind = match event_parser { @@ -612,6 +616,7 @@ impl RealtimeConversationManager { session_config, call_id: call.call_id, sideband_headers: call.sideband_headers, + session_initialization: RealtimeSidebandSessionInitialization::CoreCreated, input_channels, events_tx, handoff_state: handoff.clone(), @@ -622,6 +627,24 @@ impl RealtimeConversationManager { stop_token: stop_token.clone(), }); (task, Some(call.sdp)) + } else if let Some(call_id) = existing_call_id { + let task = existing_call::attach(existing_call::ExistingCallAttachment { + client, + model_client, + session_config, + call_id, + extra_headers: extra_headers.unwrap_or_default(), + input_channels, + events_tx, + handoff_state: handoff.clone(), + session_kind, + event_parser, + realtime_active: Arc::clone(&realtime_active), + transcript_tail_flush, + stop_token: stop_token.clone(), + }) + .await?; + (task, None) } else { let connection = client .connect( @@ -1173,19 +1196,46 @@ async fn prepare_realtime_start( }; let version = params.version.unwrap_or(match &transport { ConversationStartTransport::Websocket => config.realtime.version, - ConversationStartTransport::Webrtc { .. } => RealtimeWsVersion::V1, + ConversationStartTransport::Webrtc { .. } + | ConversationStartTransport::ExistingCall { .. } => RealtimeWsVersion::V1, }); - if matches!(transport, ConversationStartTransport::Webrtc { .. }) { - validate_avas_webrtc_start(version, config.realtime.session_type)?; + match &transport { + ConversationStartTransport::Webrtc { .. } => { + validate_avas_webrtc_start(version, config.realtime.session_type)?; + } + ConversationStartTransport::ExistingCall { .. } => { + if version == RealtimeWsVersion::V2 { + return Err(CodexErr::InvalidRequest( + "AVAS realtime calls require realtime v1 or v3".to_string(), + )); + } + if params.include_startup_context + || params.prompt.is_some() + || !params.initial_items.is_empty() + || params.model.is_some() + || params.voice.is_some() + || params.delegation_ack_filler.is_some() + { + return Err(CodexErr::InvalidRequest( + "existing realtime calls do not support session configuration options" + .to_string(), + )); + } + } + ConversationStartTransport::Websocket => {} } let configured_voice = match (&transport, params.version) { - (ConversationStartTransport::Webrtc { .. }, None) => ConfiguredRealtimeVoice::Ignore, + (ConversationStartTransport::ExistingCall { .. }, _) + | (ConversationStartTransport::Webrtc { .. }, None) => ConfiguredRealtimeVoice::Ignore, (ConversationStartTransport::Webrtc { .. } | ConversationStartTransport::Websocket, _) => { ConfiguredRealtimeVoice::Use } }; - let session_config = + let mut session_config = build_realtime_session_config(sess, ¶ms, version, configured_voice).await?; + if matches!(&transport, ConversationStartTransport::ExistingCall { .. }) { + session_config.session_id = params.realtime_session_id.clone(); + } let requested_realtime_session_id = session_config.session_id.clone(); let event_parser = session_config.event_parser; let originator = sess.originator().await; @@ -1199,7 +1249,8 @@ async fn prepare_realtime_start( originator.as_str(), )? } - ConversationStartTransport::Webrtc { .. } => { + ConversationStartTransport::Webrtc { .. } + | ConversationStartTransport::ExistingCall { .. } => { realtime_request_headers( requested_realtime_session_id.as_deref(), /*api_key*/ None, @@ -1454,9 +1505,10 @@ async fn handle_start_inner( transport, } = prepared_start; info!("starting realtime conversation"); - let sdp = match transport { - ConversationStartTransport::Websocket => None, - ConversationStartTransport::Webrtc { sdp } => Some(sdp), + let (sdp, existing_call_id) = match transport { + ConversationStartTransport::Websocket => (None, None), + ConversationStartTransport::Webrtc { sdp } => (Some(sdp), None), + ConversationStartTransport::ExistingCall { call_id } => (None, Some(call_id)), }; let mode_instructions = RealtimeModeInstructions { start: realtime_start_instructions, @@ -1476,6 +1528,7 @@ async fn handle_start_inner( session_config, model_client: sess.services.model_client.clone(), sdp, + existing_call_id, }; let start_output = sess.conversation.start(start, mode_instructions).await?; @@ -1757,6 +1810,7 @@ struct RealtimeWebrtcSidebandInputTask { session_config: RealtimeSessionConfig, call_id: String, sideband_headers: HeaderMap, + session_initialization: RealtimeSidebandSessionInitialization, input_channels: RealtimeInputChannels, events_tx: Sender, handoff_state: RealtimeHandoffState, @@ -1767,6 +1821,13 @@ struct RealtimeWebrtcSidebandInputTask { stop_token: CancellationToken, } +enum RealtimeSidebandSessionInitialization { + CoreCreated, + ExistingCall { + initial_connection: Option, + }, +} + async fn run_realtime_input_task( input: RealtimeInputTask, pending_outbound: Option>, diff --git a/codex-rs/core/src/realtime_conversation/existing_call.rs b/codex-rs/core/src/realtime_conversation/existing_call.rs new file mode 100644 index 0000000000..21ee18dcb3 --- /dev/null +++ b/codex-rs/core/src/realtime_conversation/existing_call.rs @@ -0,0 +1,90 @@ +use super::RealtimeHandoffState; +use super::RealtimeInputChannels; +use super::RealtimeSessionKind; +use super::RealtimeSidebandSessionInitialization; +use super::RealtimeTranscriptTailFlush; +use super::RealtimeWebrtcSidebandInputTask; +use super::spawn_webrtc_sideband_input_task; +use crate::client::ModelClient; +use async_channel::Sender; +use codex_api::RealtimeEvent; +use codex_api::RealtimeEventParser; +use codex_api::RealtimeSessionConfig; +use codex_api::RealtimeTranscriptState; +use codex_api::RealtimeWebsocketClient; +use codex_api::map_api_error; +use codex_login::default_client::default_headers; +use codex_protocol::error::Result as CodexResult; +use http::HeaderMap; +use std::sync::Arc; +use std::sync::atomic::AtomicBool; +use tokio::task::JoinHandle; +use tokio_util::sync::CancellationToken; + +pub(super) struct ExistingCallAttachment { + pub(super) client: RealtimeWebsocketClient, + pub(super) model_client: ModelClient, + pub(super) session_config: RealtimeSessionConfig, + pub(super) call_id: String, + pub(super) extra_headers: HeaderMap, + pub(super) input_channels: RealtimeInputChannels, + pub(super) events_tx: Sender, + pub(super) handoff_state: RealtimeHandoffState, + pub(super) session_kind: RealtimeSessionKind, + pub(super) event_parser: RealtimeEventParser, + pub(super) realtime_active: Arc, + pub(super) transcript_tail_flush: RealtimeTranscriptTailFlush, + pub(super) stop_token: CancellationToken, +} + +pub(super) async fn attach(attachment: ExistingCallAttachment) -> CodexResult> { + let ExistingCallAttachment { + client, + model_client, + session_config, + call_id, + extra_headers, + input_channels, + events_tx, + handoff_state, + session_kind, + event_parser, + realtime_active, + transcript_tail_flush, + stop_token, + } = attachment; + let sideband_headers = model_client + .realtime_sideband_headers(extra_headers) + .await?; + let transcript_state = RealtimeTranscriptState::default(); + let connection = client + .connect_existing_call_sideband( + session_config.clone(), + &call_id, + sideband_headers.clone(), + default_headers(), + transcript_state, + ) + .await + .map_err(map_api_error)?; + + Ok(spawn_webrtc_sideband_input_task( + RealtimeWebrtcSidebandInputTask { + client, + session_config, + call_id, + sideband_headers, + session_initialization: RealtimeSidebandSessionInitialization::ExistingCall { + initial_connection: Some(connection), + }, + input_channels, + events_tx, + handoff_state, + session_kind, + event_parser, + realtime_active, + transcript_tail_flush, + stop_token, + }, + )) +} diff --git a/codex-rs/core/src/realtime_conversation/sideband.rs b/codex-rs/core/src/realtime_conversation/sideband.rs index 7ce1456301..08d2aeb5a3 100644 --- a/codex-rs/core/src/realtime_conversation/sideband.rs +++ b/codex-rs/core/src/realtime_conversation/sideband.rs @@ -1,5 +1,6 @@ use super::RealtimeInputTask; use super::RealtimeInputTaskExit; +use super::RealtimeSidebandSessionInitialization; use super::RealtimeWebrtcSidebandInputTask; use super::flush_realtime_transcript_tail; use super::report_realtime_transport_loss; @@ -31,6 +32,7 @@ pub(super) fn spawn_webrtc_sideband_input_task( session_config, call_id, sideband_headers, + mut session_initialization, input_channels, events_tx, handoff_state, @@ -45,18 +47,44 @@ pub(super) fn spawn_webrtc_sideband_input_task( let mut reconnecting = false; let mut rapid_disconnects = 0_u32; let mut pending_outbound = None; - let transcript_state = RealtimeTranscriptState::default(); + let transcript_state = match &session_initialization { + RealtimeSidebandSessionInitialization::ExistingCall { + initial_connection: Some(connection), + } => connection.events().transcript_state(), + RealtimeSidebandSessionInitialization::CoreCreated + | RealtimeSidebandSessionInitialization::ExistingCall { + initial_connection: None, + } => RealtimeTranscriptState::default(), + }; while realtime_active.load(Ordering::Relaxed) { let connection = match tokio::select! { biased; _ = stop_token.cancelled() => break, - connection = client.connect_webrtc_sideband( - session_config.clone(), - &call_id, - sideband_headers.clone(), - default_headers(), - transcript_state.clone(), - ) => connection, + connection = async { + match &mut session_initialization { + RealtimeSidebandSessionInitialization::CoreCreated => { + client.connect_webrtc_sideband( + session_config.clone(), + &call_id, + sideband_headers.clone(), + default_headers(), + transcript_state.clone(), + ).await + } + RealtimeSidebandSessionInitialization::ExistingCall { initial_connection } => { + match initial_connection.take() { + Some(connection) => Ok(connection), + None => client.connect_existing_call_sideband( + session_config.clone(), + &call_id, + sideband_headers.clone(), + default_headers(), + transcript_state.clone(), + ).await, + } + } + } + } => connection, } { Ok(connection) => connection, Err(err) => { diff --git a/codex-rs/core/tests/suite/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index fa5ee2fb66..d4bdb73ab2 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/codex-rs/core/tests/suite/realtime_conversation.rs @@ -1,6 +1,7 @@ use anyhow::Context; use anyhow::Result; use chrono::Utc; +use codex_config::config_toml::RealtimeWsMode; use codex_config::config_toml::RealtimeWsVersion; use codex_core::TurnInputRequest; use codex_core::test_support::auth_manager_from_auth; @@ -54,6 +55,7 @@ use std::process::Command; use std::sync::Arc; use std::sync::Mutex; use std::time::Duration; +use test_case::test_case; use tokio::io::AsyncReadExt; use tokio::io::AsyncWriteExt; use tokio::sync::oneshot; @@ -507,6 +509,85 @@ async fn conversation_start_defaults_to_v2_and_gpt_realtime_1_5() -> Result<()> Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn conversation_existing_call_attaches_without_creating_another_call() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let realtime_server = start_websocket_server(vec![vec![vec![]]]).await; + let realtime_ws_base_url = realtime_server.uri().to_string(); + let mut builder = test_codex().with_config(move |config| { + config.experimental_realtime_ws_backend_prompt = Some("backend prompt".to_string()); + config.experimental_realtime_ws_base_url = Some(realtime_ws_base_url); + config.realtime.session_type = RealtimeWsMode::Transcription; + }); + let test = builder.build_with_auto_env(&server).await?; + + test.codex + .submit(Op::RealtimeConversationStart(ConversationStartParams { + client_managed_handoffs: false, + delegation_ack_filler: None, + flush_transcript_tail_on_session_end: false, + codex_responses_as_items: false, + codex_response_item_prefix: None, + codex_response_handoff_mode: + codex_protocol::protocol::CodexResponseHandoffMode::Thinking, + codex_response_handoff_channel_prefixes: None, + model: None, + output_modality: RealtimeOutputModality::Audio, + include_startup_context: false, + initial_items: Vec::new(), + realtime_start_instructions: None, + realtime_end_instructions: None, + prompt: None, + realtime_session_id: None, + transport: Some(ConversationStartTransport::ExistingCall { + call_id: "rtc_existing".to_string(), + }), + version: Some(RealtimeConversationVersion::V3), + voice: None, + })) + .await?; + + let started = wait_for_event_match(&test.codex, |msg| match msg { + EventMsg::RealtimeConversationStarted(started) => Some(Ok(started.clone())), + EventMsg::Error(err) => Some(Err(err.clone())), + _ => None, + }) + .await + .expect("existing call sideband attachment failed"); + assert_eq!( + (started.version, started.realtime_session_id), + (RealtimeConversationVersion::V3, None) + ); + + let handshake = realtime_server.single_handshake(); + assert_eq!(handshake.uri(), "/v1/live/rtc_existing"); + assert_eq!( + handshake.header("authorization").as_deref(), + Some("Bearer dummy") + ); + assert_eq!(handshake.header("x-session-id"), None); + assert!( + server + .received_requests() + .await + .context("mock server should record requests")? + .iter() + .all(|request| !request.url.path().ends_with("/realtime/calls")), + "attaching to an existing call must not create another realtime call" + ); + + test.codex.submit(Op::RealtimeConversationClose).await?; + let _closed = wait_for_event_match(&test.codex, |msg| match msg { + EventMsg::RealtimeConversationClosed(closed) => Some(closed.clone()), + _ => None, + }) + .await; + realtime_server.shutdown().await; + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn conversation_webrtc_frameless_chatgpt_sends_codex_headers_to_backend() -> Result<()> { skip_if_no_network!(Ok(())); @@ -823,10 +904,22 @@ async fn conversation_webrtc_start_posts_generated_session() -> Result<()> { Ok(()) } +#[test_case( + ConversationStartTransport::Webrtc { sdp: "v=offer\r\n".to_string() }; + "core-created webrtc" +)] +#[test_case( + ConversationStartTransport::ExistingCall { call_id: "rtc_reconnect".to_string() }; + "client-created existing call" +)] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn conversation_webrtc_live_reconnects_sideband_after_unclean_disconnect() -> Result<()> { +async fn conversation_webrtc_live_reconnects_sideband_after_unclean_disconnect( + transport: ConversationStartTransport, +) -> Result<()> { skip_if_no_network!(Ok(())); + let attaches_existing_call = + matches!(&transport, ConversationStartTransport::ExistingCall { .. }); let server = start_mock_server().await; Mock::given(method("POST")) .and(path("/v1/live")) @@ -979,7 +1072,7 @@ async fn conversation_webrtc_live_reconnects_sideband_after_unclean_disconnect() config.experimental_realtime_ws_base_url = Some(realtime_ws_base_url); config.realtime.version = RealtimeWsVersion::V3; }); - let test = builder.build(&server).await?; + let test = builder.build_with_auto_env(&server).await?; test.codex .submit(Op::RealtimeConversationStart(ConversationStartParams { @@ -993,15 +1086,17 @@ async fn conversation_webrtc_live_reconnects_sideband_after_unclean_disconnect() codex_response_handoff_channel_prefixes: None, model: None, output_modality: RealtimeOutputModality::Audio, - include_startup_context: true, + include_startup_context: !attaches_existing_call, initial_items: Vec::new(), realtime_start_instructions: None, realtime_end_instructions: None, - prompt: Some(Some("backend prompt".to_string())), - realtime_session_id: None, - transport: Some(ConversationStartTransport::Webrtc { - sdp: "v=offer\r\n".to_string(), - }), + prompt: if attaches_existing_call { + None + } else { + Some(Some("backend prompt".to_string())) + }, + realtime_session_id: attaches_existing_call.then(|| "sess_client_owned".to_string()), + transport: Some(transport), version: Some(RealtimeConversationVersion::V3), voice: None, })) diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index b2230b0aea..8c142856c3 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -251,6 +251,7 @@ pub struct ConversationStartParams { pub enum ConversationStartTransport { Websocket, Webrtc { sdp: String }, + ExistingCall { call_id: String }, } #[derive(Debug, Clone, Copy, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)]