diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 336f769d71..eaffff3d03 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -858,7 +858,7 @@ const offer = await pc.createOffer(); await pc.setLocalDescription(offer); ``` -Then send `offer.sdp` to app-server. Core uses `experimental_realtime_ws_backend_prompt` for the backend instructions and the thread conversation id as the default Realtime API session identifier. This `realtimeSessionId` value refers to the upstream Realtime API session, not a Codex session/thread-group id. The start response is `{}`; the remote answer SDP arrives later as `thread/realtime/sdp` and should be passed to `setRemoteDescription()`: +Then send `offer.sdp` to app-server. Core uses `experimental_realtime_ws_backend_prompt` for the backend instructions and the thread conversation id as the default session correlation value sent upstream. The start response is `{}`; the remote answer SDP arrives later as `thread/realtime/sdp` and should be passed to `setRemoteDescription()`: ```json { "method": "thread/realtime/start", "id": 40, "params": { @@ -1308,7 +1308,7 @@ The fuzzy file search session API emits per-query notifications: The thread realtime API emits thread-scoped notifications for session lifecycle and streaming media: -- `thread/realtime/started` — `{ threadId, realtimeSessionId }` once realtime starts for the thread (experimental). `realtimeSessionId` is the upstream Realtime API session identifier, not a Codex session/thread-group id. +- `thread/realtime/started` — `{ threadId, realtimeSessionId }` after the upstream Realtime API creates or updates the session (experimental). `realtimeSessionId` is exactly the upstream event's `session.id`, not a Codex thread ID or client-supplied correlation ID. - `thread/realtime/itemAdded` — `{ threadId, item }` for raw non-audio realtime items that do not have a dedicated typed app-server notification, including `handoff_request` (experimental). `item` is forwarded as raw JSON while the upstream websocket item schema remains unstable. - `thread/realtime/transcript/delta` — `{ threadId, role, delta }` for live realtime transcript deltas (experimental). - `thread/realtime/transcript/done` — `{ threadId, role, text }` when realtime emits the final full text for a transcript part (experimental). 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 f88ed799a0..a4c7e728fb 100644 --- a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs +++ b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs @@ -336,7 +336,7 @@ impl RealtimeE2eHarness { codex_responses_as_items: Option, ) -> Result { // Starts realtime through the public JSON-RPC method, then waits for the same client-visible - // notifications a desktop app needs: started first, SDP answer second. + // notifications a desktop app needs: the SDP answer, then the upstream session id. let start_request_id = self .mcp .send_thread_realtime_start_request(ThreadRealtimeStartParams { @@ -366,12 +366,12 @@ impl RealtimeE2eHarness { .await??; let _: ThreadRealtimeStartResponse = to_response(start_response)?; - let started = self - .read_notification::("thread/realtime/started") - .await?; let sdp = self .read_notification::("thread/realtime/sdp") .await?; + let started = self + .read_notification::("thread/realtime/started") + .await?; Ok(StartedWebrtcRealtime { started, sdp }) } @@ -502,6 +502,91 @@ fn session_updated(realtime_session_id: &str) -> Value { }) } +fn session_created(realtime_session_id: &str) -> Value { + json!({ + "type": "session.created", + "session": { "id": realtime_session_id, "instructions": "backend prompt" } + }) +} + +#[tokio::test] +async fn realtime_v1_websocket_started_uses_upstream_session_id() -> 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![ + vec![session_created("sess_frontend_test")], + vec![], + ]]) + .await; + let codex_home = TempDir::new()?; + create_config_toml_with_realtime_version( + codex_home.path(), + &responses_server.uri(), + realtime_server.uri(), + /*realtime_enabled*/ true, + StartupContextConfig::Override("startup context"), + RealtimeTestVersion::V1, + RealtimeTestSandbox::ReadOnly, + )?; + + let mut mcp = TestAppServer::new(codex_home.path()).await?; + timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; + login_with_api_key(&mut mcp, "sk-test-key").await?; + + let thread_start_request_id = mcp + .send_thread_start_request(ThreadStartParams::default()) + .await?; + let thread_start_response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(thread_start_request_id)), + ) + .await??; + let thread_start: ThreadStartResponse = to_response(thread_start_response)?; + + let start_request_id = mcp + .send_thread_realtime_start_request(ThreadRealtimeStartParams { + architecture: None, + codex_responses_as_items: None, + codex_response_item_prefix: None, + thread_id: thread_start.thread.id.clone(), + model: None, + output_modality: RealtimeOutputModality::Audio, + include_startup_context: None, + prompt: None, + realtime_session_id: Some("client-correlation-id".to_string()), + transport: None, + version: Some(RealtimeConversationVersion::V1), + voice: None, + }) + .await?; + let start_response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(start_request_id)), + ) + .await??; + let _: ThreadRealtimeStartResponse = to_response(start_response)?; + + let started = + read_notification::(&mut mcp, "thread/realtime/started") + .await?; + assert_eq!( + started, + ThreadRealtimeStartedNotification { + thread_id: thread_start.thread.id.clone(), + realtime_session_id: Some("sess_frontend_test".to_string()), + version: RealtimeConversationVersion::V1, + } + ); + assert_ne!( + started.realtime_session_id.as_deref(), + Some(thread_start.thread.id.as_str()) + ); + + realtime_server.shutdown().await; + Ok(()) +} + fn v2_background_agent_tool_call(call_id: &str, prompt: &str) -> Value { json!({ "type": "conversation.item.done", @@ -524,10 +609,7 @@ async fn realtime_conversation_streams_v2_notifications() -> Result<()> { ]) .await; let realtime_server = start_websocket_server(vec![vec![ - vec![json!({ - "type": "session.updated", - "session": { "id": "sess_backend", "instructions": "backend prompt" } - })], + vec![session_created("sess_frontend_test")], vec![], vec![ json!({ @@ -634,9 +716,18 @@ async fn realtime_conversation_streams_v2_notifications() -> Result<()> { let started = read_notification::(&mut mcp, "thread/realtime/started") .await?; - assert_eq!(started.thread_id, thread_start.thread.id); - assert!(started.realtime_session_id.is_some()); - assert_eq!(started.version, RealtimeConversationVersion::V2); + assert_eq!( + started, + ThreadRealtimeStartedNotification { + thread_id: thread_start.thread.id.clone(), + realtime_session_id: Some("sess_frontend_test".to_string()), + version: RealtimeConversationVersion::V2, + } + ); + assert_ne!( + started.realtime_session_id.as_deref(), + Some(started.thread_id.as_str()) + ); let startup_context_request = realtime_server .wait_for_request(/*connection_index*/ 0, /*request_index*/ 0) @@ -1385,7 +1476,7 @@ async fn webrtc_v1_start_posts_offer_returns_sdp_and_joins_sideband() -> Result< StartedWebrtcRealtime { started: ThreadRealtimeStartedNotification { thread_id: harness.thread_id.clone(), - realtime_session_id: Some(harness.thread_id.clone()), + realtime_session_id: Some("sess_v1_webrtc".to_string()), version: RealtimeConversationVersion::V1, }, sdp: ThreadRealtimeSdpNotification { @@ -1394,6 +1485,10 @@ async fn webrtc_v1_start_posts_offer_returns_sdp_and_joins_sideband() -> Result< }, } ); + assert_ne!( + started.started.realtime_session_id.as_deref(), + Some(harness.thread_id.as_str()) + ); // Phase 3: verify the HTTP call-create leg, the direct sideband join, and the normal v1 // session.update; the WebRTC transport should remain alive instead of closing after SDP. 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 0aa2feb04e..4febd55094 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs @@ -873,20 +873,22 @@ mod tests { use tokio_tungstenite::tungstenite::Message; #[test] - fn parse_session_updated_event() { - let payload = json!({ - "type": "session.updated", - "session": {"id": "sess_123", "instructions": "backend prompt"} - }) - .to_string(); - - assert_eq!( - parse_realtime_event(payload.as_str(), RealtimeEventParser::V1), - Some(RealtimeEvent::SessionUpdated { - realtime_session_id: "sess_123".to_string(), - instructions: Some("backend prompt".to_string()), + fn parse_session_created_and_updated_events() { + for event_type in ["session.created", "session.updated"] { + let payload = json!({ + "type": event_type, + "session": {"id": "sess_123", "instructions": "backend prompt"} }) - ); + .to_string(); + + assert_eq!( + parse_realtime_event(payload.as_str(), RealtimeEventParser::V1), + Some(RealtimeEvent::SessionUpdated { + realtime_session_id: "sess_123".to_string(), + instructions: Some("backend prompt".to_string()), + }) + ); + } } #[test] diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_common.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_common.rs index 2c96280672..54310eba6d 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_common.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_common.rs @@ -24,7 +24,7 @@ pub(super) fn parse_realtime_payload(payload: &str, parser_name: &str) -> Option Some((parsed, message_type)) } -pub(super) fn parse_session_updated_event(parsed: &Value) -> Option { +pub(super) fn parse_session_event(parsed: &Value) -> Option { let session_id = parsed .get("session") .and_then(Value::as_object) diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v1.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v1.rs index a464852244..da475328b3 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v1.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v1.rs @@ -1,6 +1,6 @@ use crate::endpoint::realtime_websocket::protocol_common::parse_error_event; use crate::endpoint::realtime_websocket::protocol_common::parse_realtime_payload; -use crate::endpoint::realtime_websocket::protocol_common::parse_session_updated_event; +use crate::endpoint::realtime_websocket::protocol_common::parse_session_event; use crate::endpoint::realtime_websocket::protocol_common::parse_transcript_delta_event; use crate::endpoint::realtime_websocket::protocol_common::parse_transcript_done_event; use codex_protocol::protocol::RealtimeAudioFrame; @@ -12,7 +12,7 @@ use tracing::debug; pub(super) fn parse_realtime_event_v1(payload: &str) -> Option { let (parsed, message_type) = parse_realtime_payload(payload, "realtime v1")?; match message_type.as_str() { - "session.updated" => parse_session_updated_event(&parsed), + "session.created" | "session.updated" => parse_session_event(&parsed), "conversation.output_audio.delta" => { let data = parsed .get("delta") diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v2.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v2.rs index ee0bba9946..93fc790cc9 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v2.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v2.rs @@ -1,6 +1,6 @@ use crate::endpoint::realtime_websocket::protocol_common::parse_error_event; use crate::endpoint::realtime_websocket::protocol_common::parse_realtime_payload; -use crate::endpoint::realtime_websocket::protocol_common::parse_session_updated_event; +use crate::endpoint::realtime_websocket::protocol_common::parse_session_event; use crate::endpoint::realtime_websocket::protocol_common::parse_transcript_delta_event; use crate::endpoint::realtime_websocket::protocol_common::parse_transcript_done_event; use codex_protocol::protocol::RealtimeAudioFrame; @@ -25,7 +25,7 @@ pub(super) fn parse_realtime_event_v2(payload: &str) -> Option { let (parsed, message_type) = parse_realtime_payload(payload, "realtime v2")?; match message_type.as_str() { - "session.updated" => parse_session_updated_event(&parsed), + "session.created" | "session.updated" => parse_session_event(&parsed), "response.output_audio.delta" | "response.audio.delta" => { parse_output_audio_delta_event(&parsed) } diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index 2b5eb91a23..03d0172f43 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -678,7 +678,6 @@ struct PreparedRealtimeConversationStart { codex_responses_as_items: bool, codex_response_item_prefix: Option, realtime_call_api_provider: Option, - requested_realtime_session_id: Option, version: RealtimeWsVersion, session_config: RealtimeSessionConfig, transport: ConversationStartTransport, @@ -747,7 +746,6 @@ async fn prepare_realtime_start( codex_responses_as_items: params.codex_responses_as_items, codex_response_item_prefix: params.codex_response_item_prefix, realtime_call_api_provider, - requested_realtime_session_id, version, session_config, transport, @@ -917,7 +915,6 @@ async fn handle_start_inner( codex_responses_as_items, codex_response_item_prefix, realtime_call_api_provider, - requested_realtime_session_id, version, session_config, transport, @@ -938,24 +935,11 @@ async fn handle_start_inner( model_client: sess.services.model_client.clone(), sdp, }; - let start_output = sess.conversation.start(start).await?; - - info!("realtime conversation started"); - - sess.send_event_raw(Event { - id: sub_id.to_string(), - msg: EventMsg::RealtimeConversationStarted(RealtimeConversationStartedEvent { - realtime_session_id: requested_realtime_session_id, - version, - }), - }) - .await; - let RealtimeStartOutput { realtime_active, events_rx, sdp, - } = start_output; + } = sess.conversation.start(start).await?; if let Some(sdp) = sdp { sess.send_event_raw(Event { id: sub_id.to_string(), @@ -964,6 +948,48 @@ async fn handle_start_inner( .await; } + let mut pending_events = Vec::new(); + let realtime_session_id = loop { + match events_rx.recv().await { + Ok(RealtimeEvent::SessionUpdated { + realtime_session_id, + instructions, + }) => { + let started_realtime_session_id = realtime_session_id.clone(); + pending_events.push(RealtimeEvent::SessionUpdated { + realtime_session_id, + instructions, + }); + break started_realtime_session_id; + } + Ok(RealtimeEvent::Error(message)) => { + sess.conversation.finish_if_active(&realtime_active).await; + return Err(CodexErr::Stream(message, None)); + } + Ok(event) => pending_events.push(event), + Err(_) if !realtime_active.load(Ordering::Relaxed) => return Ok(()), + Err(_) => { + sess.conversation.finish_if_active(&realtime_active).await; + return Err(CodexErr::Stream( + "realtime conversation ended before the upstream session was created" + .to_string(), + None, + )); + } + } + }; + + info!(%realtime_session_id, "realtime conversation started"); + + sess.send_event_raw(Event { + id: sub_id.to_string(), + msg: EventMsg::RealtimeConversationStarted(RealtimeConversationStartedEvent { + realtime_session_id: Some(realtime_session_id), + version, + }), + }) + .await; + let sess_clone = Arc::clone(sess); let sub_id = sub_id.to_string(); let fanout_realtime_active = Arc::clone(&realtime_active); @@ -973,7 +999,15 @@ async fn handle_start_inner( msg, }; let mut end = RealtimeConversationEnd::TransportClosed; - while let Ok(event) = events_rx.recv().await { + let mut pending_events = pending_events.into_iter(); + loop { + let event = match pending_events.next() { + Some(event) => event, + None => match events_rx.recv().await { + Ok(event) => event, + Err(_) => break, + }, + }; if !fanout_realtime_active.load(Ordering::Relaxed) { break; } diff --git a/codex-rs/core/tests/suite/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index 61fc38e8b1..13870fbe71 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/codex-rs/core/tests/suite/realtime_conversation.rs @@ -304,7 +304,7 @@ async fn conversation_start_audio_text_close_round_trip() -> Result<()> { }) .await .expect("conversation start failed"); - assert!(started.realtime_session_id.is_some()); + assert_eq!(started.realtime_session_id.as_deref(), Some("sess_1")); assert_eq!(started.version, RealtimeConversationVersion::V1); let session_updated = wait_for_event_match(&test.codex, |msg| match msg { @@ -362,7 +362,7 @@ async fn conversation_start_audio_text_close_round_trip() -> Result<()> { let initial_instructions = websocket_request_instructions(&connection[0]) .expect("initial session update instructions"); assert!(initial_instructions.starts_with("backend prompt")); - assert_eq!( + assert_ne!( server.handshakes()[1] .header("x-session-id") .expect("session.update x-session-id header"), @@ -962,7 +962,7 @@ async fn conversation_webrtc_close_while_sideband_connecting_drops_pending_join( } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn conversation_webrtc_sideband_connect_failure_closes_with_error() -> Result<()> { +async fn conversation_webrtc_sideband_connect_failure_errors_without_started_event() -> Result<()> { skip_if_no_network!(Ok(())); let server = start_mock_server().await; @@ -1005,13 +1005,6 @@ async fn conversation_webrtc_sideband_connect_failure_closes_with_error() -> Res })) .await?; - let started = wait_for_event_match(&test.codex, |msg| match msg { - EventMsg::RealtimeConversationStarted(started) => Some(started.clone()), - _ => None, - }) - .await; - assert!(started.realtime_session_id.is_some()); - let sdp = wait_for_event_match(&test.codex, |msg| match msg { EventMsg::RealtimeConversationSdp(created) => Some(created.sdp.clone()), _ => None, @@ -1019,22 +1012,18 @@ async fn conversation_webrtc_sideband_connect_failure_closes_with_error() -> Res .await; assert_eq!(sdp, "v=answer\r\n"); - let err = wait_for_event_match(&test.codex, |msg| match msg { + let startup_result = wait_for_event_match(&test.codex, |msg| match msg { + EventMsg::RealtimeConversationStarted(started) => Some(Err(started.clone())), EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent { payload: RealtimeEvent::Error(message), - }) => Some(message.clone()), + }) => Some(Ok(message.clone())), _ => None, }) .await; + let err = startup_result + .unwrap_or_else(|started| panic!("unexpected realtime started event: {started:?}")); assert!(!err.is_empty()); - let closed = wait_for_event_match(&test.codex, |msg| match msg { - EventMsg::RealtimeConversationClosed(closed) => Some(closed.clone()), - _ => None, - }) - .await; - assert_eq!(closed.reason.as_deref(), Some("error")); - test.codex .submit(Op::RealtimeConversationText(ConversationTextParams { text: "after sideband failure".to_string(), @@ -1102,7 +1091,7 @@ async fn conversation_start_uses_openai_env_key_fallback_with_chatgpt_auth() -> }) .await .expect("conversation start failed"); - assert!(started.realtime_session_id.is_some()); + assert_eq!(started.realtime_session_id.as_deref(), Some("sess_env")); let session_updated = wait_for_event_match(&test.codex, |msg| match msg { EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {