fix(app-server): report upstream realtime session ID

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).

codex-rs/app-server/tests/suite/v2/realtime_conversation.rs

@@ -336,7 +336,7 @@ impl RealtimeE2eHarness {
         codex_responses_as_items: Option<bool>,
     ) -> Result<StartedWebrtcRealtime> {
         // 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::<ThreadRealtimeStartedNotification>("thread/realtime/started")
-            .await?;
         let sdp = self
             .read_notification::<ThreadRealtimeSdpNotification>("thread/realtime/sdp")
             .await?;
+        let started = self
+            .read_notification::<ThreadRealtimeStartedNotification>("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::<ThreadRealtimeStartedNotification>(&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::<ThreadRealtimeStartedNotification>(&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.

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]

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<RealtimeEvent> {
+pub(super) fn parse_session_event(parsed: &Value) -> Option<RealtimeEvent> {
     let session_id = parsed
         .get("session")
         .and_then(Value::as_object)

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<RealtimeEvent> {
     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")

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<RealtimeEvent> {
     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)
         }

codex-rs/core/src/realtime_conversation.rs

@@ -678,7 +678,6 @@ struct PreparedRealtimeConversationStart {
     codex_responses_as_items: bool,
     codex_response_item_prefix: Option<String>,
     realtime_call_api_provider: Option<ApiProvider>,
-    requested_realtime_session_id: Option<String>,
     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;
             }

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 {