Support attaching to existing realtime calls (#39876)

## Why

Clients that create and negotiate realtime calls themselves need Codex to join
those calls without creating another call or overwriting the client-owned
session configuration.

## What changed

- Add the `existingCall` transport to `thread/realtime/start`, accepting a
  client-provided `callId` and optional `realtimeSessionId`.
- Attach Codex over the call's sideband WebSocket without SDP negotiation or a
  session update, and reconnect that sideband while preserving transcript
  state.
- Default existing calls to realtime v1, support v1 and v3, and reject v2 or
  options that would reconfigure the client-owned session.
- Encode call IDs as a single URL path segment for v3 sideband connections.

## Testing

Add protocol, app-server, core, reconnection, authentication-header, and URL
encoding coverage for existing-call attachments.

GitOrigin-RevId: 2046b70566f98efa55381bb1e461ea3cc68256c5
This commit is contained in:
Zahan Malkani
2026-08-21 06:41:28 +00:00
committed by copyberry
parent 2aaefa32b0
commit 536f86e5cc
18 changed files with 765 additions and 34 deletions
@@ -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"
}
]
},
@@ -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"
}
]
},
@@ -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"
}
]
},
@@ -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, };
@@ -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.
+21
View File
@@ -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
@@ -1082,6 +1082,36 @@ impl TurnRequestProcessor {
request_id: &ConnectionRequestId,
params: ThreadRealtimeStartParams,
) -> Result<Option<ThreadRealtimeStartResponse>, JSONRPCErrorError> {
let attaches_existing_call = matches!(
&params.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, &params.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,
@@ -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<RealtimeConversationVersion>,
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::<ThreadRealtimeErrorNotification>("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::<ThreadRealtimeStartedNotification>("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(()));
@@ -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<RealtimeWebsocketConnection, ApiError> {
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<RealtimeWebsocketConnection, ApiError> {
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<RealtimeWebsocketConnection, ApiError> {
// 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<RealtimeWebsocketConnection, ApiError> {
// 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<RealtimeWebsocketConnection, ApiError> {
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("<none>"),
"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 {
+14
View File
@@ -686,6 +686,20 @@ impl ModelClient {
})
}
pub(crate) async fn realtime_sideband_headers(
&self,
mut extra_headers: ApiHeaderMap,
) -> Result<ApiHeaderMap> {
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`.
+19
View File
@@ -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) =
+70 -9
View File
@@ -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<String>,
existing_call_id: Option<String>,
}
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, &params, 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<RealtimeEvent>,
handoff_state: RealtimeHandoffState,
@@ -1767,6 +1821,13 @@ struct RealtimeWebrtcSidebandInputTask {
stop_token: CancellationToken,
}
enum RealtimeSidebandSessionInitialization {
CoreCreated,
ExistingCall {
initial_connection: Option<RealtimeWebsocketConnection>,
},
}
async fn run_realtime_input_task(
input: RealtimeInputTask,
pending_outbound: Option<Box<RealtimePendingOutbound>>,
@@ -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<RealtimeEvent>,
pub(super) handoff_state: RealtimeHandoffState,
pub(super) session_kind: RealtimeSessionKind,
pub(super) event_parser: RealtimeEventParser,
pub(super) realtime_active: Arc<AtomicBool>,
pub(super) transcript_tail_flush: RealtimeTranscriptTailFlush,
pub(super) stop_token: CancellationToken,
}
pub(super) async fn attach(attachment: ExistingCallAttachment) -> CodexResult<JoinHandle<()>> {
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,
},
))
}
@@ -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) => {
@@ -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,
}))
+1
View File
@@ -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)]