codex: address PR review feedback (#14902)

Co-authored-by: Codex <noreply@openai.com>

codex-rs/core/src/realtime_conversation.rs

@@ -352,7 +352,23 @@ pub(crate) async fn handle_start(
     sub_id: String,
     params: ConversationStartParams,
 ) -> CodexResult<()> {
-    if let Err(err) = handle_start_inner(sess, &sub_id, params).await {
+    let prepared_start = match prepare_realtime_start(sess, params).await {
+        Ok(prepared_start) => prepared_start,
+        Err(err) => {
+            error!("failed to prepare realtime conversation: {err}");
+            let message = err.to_string();
+            sess.send_event_raw(Event {
+                id: sub_id,
+                msg: EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
+                    payload: RealtimeEvent::Error(message),
+                }),
+            })
+            .await;
+            return Ok(());
+        }
+    };
+
+    if let Err(err) = handle_start_inner(sess, &sub_id, prepared_start).await {
         error!("failed to start realtime conversation: {err}");
         let message = err.to_string();
         sess.send_event_raw(Event {
@@ -367,11 +383,17 @@ pub(crate) async fn handle_start(
     Ok(())
 }
 
-async fn handle_start_inner(
+struct PreparedRealtimeConversationStart {
+    api_provider: ApiProvider,
+    extra_headers: Option<HeaderMap>,
+    requested_session_id: Option<String>,
+    session_config: RealtimeSessionConfig,
+}
+
+async fn prepare_realtime_start(
     sess: &Arc<Session>,
-    sub_id: &str,
     params: ConversationStartParams,
-) -> CodexResult<()> {
+) -> CodexResult<PreparedRealtimeConversationStart> {
     let provider = sess.provider().await;
     let auth = sess.services.auth_manager.auth().await;
     let realtime_api_key = realtime_api_key(auth.as_ref(), &provider)?;
@@ -418,6 +440,25 @@ async fn handle_start_inner(
     };
     let extra_headers =
         realtime_request_headers(requested_session_id.as_deref(), realtime_api_key.as_str())?;
+    Ok(PreparedRealtimeConversationStart {
+        api_provider,
+        extra_headers,
+        requested_session_id,
+        session_config,
+    })
+}
+
+async fn handle_start_inner(
+    sess: &Arc<Session>,
+    sub_id: &str,
+    prepared_start: PreparedRealtimeConversationStart,
+) -> CodexResult<()> {
+    let PreparedRealtimeConversationStart {
+        api_provider,
+        extra_headers,
+        requested_session_id,
+        session_config,
+    } = prepared_start;
     info!("starting realtime conversation");
     let (events_rx, realtime_active) = sess
         .conversation

codex-rs/core/tests/suite/realtime_conversation.rs

@@ -434,10 +434,10 @@ async fn conversation_audio_before_start_emits_error() -> Result<()> {
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn conversation_start_failure_emits_realtime_error_and_closed() -> Result<()> {
+async fn conversation_start_preflight_failure_emits_realtime_error_only() -> Result<()> {
     if std::env::var_os(REALTIME_CONVERSATION_TEST_SUBPROCESS_ENV_VAR).is_none() {
         return run_realtime_conversation_test_in_subprocess(
-            "suite::realtime_conversation::conversation_start_failure_emits_realtime_error_and_closed",
+            "suite::realtime_conversation::conversation_start_preflight_failure_emits_realtime_error_only",
             None,
         );
     }
@@ -464,6 +464,46 @@ async fn conversation_start_failure_emits_realtime_error_and_closed() -> Result<
     .await;
     assert_eq!(err, "realtime conversation requires API key auth");
 
+    let closed = timeout(Duration::from_millis(200), async {
+        wait_for_event_match(&test.codex, |msg| match msg {
+            EventMsg::RealtimeConversationClosed(closed) => Some(closed.clone()),
+            _ => None,
+        })
+        .await
+    })
+    .await;
+    assert!(closed.is_err(), "preflight failure should not emit closed");
+
+    server.shutdown().await;
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn conversation_start_connect_failure_emits_realtime_error_and_closed() -> Result<()> {
+    skip_if_no_network!(Ok(()));
+
+    let server = start_websocket_server(vec![]).await;
+    let mut builder = test_codex().with_config(|config| {
+        config.experimental_realtime_ws_base_url = Some("http://127.0.0.1:1".to_string());
+    });
+    let test = builder.build_with_websocket_server(&server).await?;
+
+    test.codex
+        .submit(Op::RealtimeConversationStart(ConversationStartParams {
+            prompt: "backend prompt".to_string(),
+            session_id: None,
+        }))
+        .await?;
+
+    let err = wait_for_event_match(&test.codex, |msg| match msg {
+        EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
+            payload: RealtimeEvent::Error(message),
+        }) => Some(message.clone()),
+        _ => None,
+    })
+    .await;
+    assert!(!err.is_empty());
+
     let closed = wait_for_event_match(&test.codex, |msg| match msg {
         EventMsg::RealtimeConversationClosed(closed) => Some(closed.clone()),
         _ => None,