Improve realtime startup context, recovery, and queue visibility

codex-rs/core/src/realtime_context.rs

@@ -1,8 +1,11 @@
 use crate::codex::Session;
+use crate::compact::content_items_to_text;
+use crate::event_mapping::is_contextual_user_message_content;
 use crate::git_info::resolve_root_git_project_for_trust;
 use crate::truncate::TruncationPolicy;
 use crate::truncate::truncate_text;
 use chrono::Utc;
+use codex_protocol::models::ResponseItem;
 use codex_state::SortKey;
 use codex_state::ThreadMetadata;
 use dirs::home_dir;
@@ -19,9 +22,11 @@ use tracing::info;
 use tracing::warn;
 
 const STARTUP_CONTEXT_HEADER: &str = "Startup context from Codex.\nThis is background context about recent work and machine/workspace layout. It may be incomplete or stale. Use it to inform responses, and do not repeat it back unless relevant.";
+const CURRENT_THREAD_SECTION_TOKEN_BUDGET: usize = 1_200;
 const RECENT_WORK_SECTION_TOKEN_BUDGET: usize = 2_200;
 const WORKSPACE_SECTION_TOKEN_BUDGET: usize = 1_600;
 const NOTES_SECTION_TOKEN_BUDGET: usize = 300;
+const MAX_CURRENT_THREAD_TURNS: usize = 2;
 const MAX_RECENT_THREADS: usize = 40;
 const MAX_RECENT_WORK_GROUPS: usize = 8;
 const MAX_CURRENT_CWD_ASKS: usize = 8;
@@ -49,20 +54,33 @@ pub(crate) async fn build_realtime_startup_context(
 ) -> Option<String> {
     let config = sess.get_config().await;
     let cwd = config.cwd.clone();
+    let history = sess.clone_history().await;
+    let current_thread_section = build_current_thread_section(history.raw_items());
     let recent_threads = load_recent_threads(sess).await;
     let recent_work_section = build_recent_work_section(&cwd, &recent_threads);
     let workspace_section = build_workspace_section(&cwd);
 
-    if recent_work_section.is_none() && workspace_section.is_none() {
+    if current_thread_section.is_none()
+        && recent_work_section.is_none()
+        && workspace_section.is_none()
+    {
         debug!("realtime startup context unavailable; skipping injection");
         return None;
     }
 
     let mut parts = vec![STARTUP_CONTEXT_HEADER.to_string()];
 
+    let has_current_thread_section = current_thread_section.is_some();
     let has_recent_work_section = recent_work_section.is_some();
     let has_workspace_section = workspace_section.is_some();
 
+    if let Some(section) = format_section(
+        "Current Thread",
+        current_thread_section,
+        CURRENT_THREAD_SECTION_TOKEN_BUDGET,
+    ) {
+        parts.push(section);
+    }
     if let Some(section) = format_section(
         "Recent Work",
         recent_work_section,
@@ -79,7 +97,7 @@ pub(crate) async fn build_realtime_startup_context(
     }
     if let Some(section) = format_section(
         "Notes",
-        Some("Built at realtime startup from persisted thread metadata in the state DB and a bounded local workspace scan. This excludes repo memory instructions, AGENTS files, project-doc prompt blends, and memory summaries.".to_string()),
+        Some("Built at realtime startup from the current thread history, persisted thread metadata in the state DB, and a bounded local workspace scan. This excludes repo memory instructions, AGENTS files, project-doc prompt blends, and memory summaries.".to_string()),
         NOTES_SECTION_TOKEN_BUDGET,
     ) {
         parts.push(section);
@@ -89,6 +107,7 @@ pub(crate) async fn build_realtime_startup_context(
     debug!(
         approx_tokens = approx_token_count(&context),
         bytes = context.len(),
+        has_current_thread_section,
         has_recent_work_section,
         has_workspace_section,
         "built realtime startup context"
@@ -167,6 +186,90 @@ fn build_recent_work_section(cwd: &Path, recent_threads: &[ThreadMetadata]) -> O
     (!sections.is_empty()).then(|| sections.join("\n\n"))
 }
 
+fn build_current_thread_section(items: &[ResponseItem]) -> Option<String> {
+    let mut turns = Vec::new();
+    let mut current_user = Vec::new();
+    let mut current_assistant = Vec::new();
+
+    for item in items {
+        match item {
+            ResponseItem::Message { role, content, .. } if role == "user" => {
+                if is_contextual_user_message_content(content) {
+                    continue;
+                }
+                let Some(text) = content_items_to_text(content)
+                    .map(|text| text.trim().to_string())
+                    .filter(|text| !text.is_empty())
+                else {
+                    continue;
+                };
+                if !current_user.is_empty() || !current_assistant.is_empty() {
+                    turns.push((
+                        std::mem::take(&mut current_user),
+                        std::mem::take(&mut current_assistant),
+                    ));
+                }
+                current_user.push(text);
+            }
+            ResponseItem::Message { role, content, .. } if role == "assistant" => {
+                let Some(text) = content_items_to_text(content)
+                    .map(|text| text.trim().to_string())
+                    .filter(|text| !text.is_empty())
+                else {
+                    continue;
+                };
+                if current_user.is_empty() && current_assistant.is_empty() {
+                    continue;
+                }
+                current_assistant.push(text);
+            }
+            _ => {}
+        }
+    }
+
+    if !current_user.is_empty() || !current_assistant.is_empty() {
+        turns.push((current_user, current_assistant));
+    }
+
+    let retained_turns = turns
+        .into_iter()
+        .rev()
+        .take(MAX_CURRENT_THREAD_TURNS)
+        .collect::<Vec<_>>()
+        .into_iter()
+        .rev()
+        .collect::<Vec<_>>();
+    if retained_turns.is_empty() {
+        return None;
+    }
+
+    let mut lines = vec![
+        "Most recent user/assistant turns from this exact thread. Use them for continuity when responding.".to_string(),
+    ];
+
+    let retained_turn_count = retained_turns.len();
+    for (index, (user_messages, assistant_messages)) in retained_turns.into_iter().enumerate() {
+        lines.push(String::new());
+        if retained_turn_count == 1 || index + 1 == retained_turn_count {
+            lines.push("### Latest turn".to_string());
+        } else {
+            lines.push(format!("### Prior turn {}", index + 1));
+        }
+
+        if !user_messages.is_empty() {
+            lines.push("User:".to_string());
+            lines.push(user_messages.join("\n\n"));
+        }
+        if !assistant_messages.is_empty() {
+            lines.push(String::new());
+            lines.push("Assistant:".to_string());
+            lines.push(assistant_messages.join("\n\n"));
+        }
+    }
+
+    Some(lines.join("\n"))
+}
+
 fn build_workspace_section(cwd: &Path) -> Option<String> {
     build_workspace_section_with_user_root(cwd, home_dir())
 }
@@ -396,12 +499,16 @@ fn approx_token_count(text: &str) -> usize {
 
 #[cfg(test)]
 mod tests {
+    use super::build_current_thread_section;
     use super::build_recent_work_section;
     use super::build_workspace_section;
     use super::build_workspace_section_with_user_root;
+    use crate::contextual_user_message::ENVIRONMENT_CONTEXT_FRAGMENT;
     use chrono::TimeZone;
     use chrono::Utc;
     use codex_protocol::ThreadId;
+    use codex_protocol::models::ContentItem;
+    use codex_protocol::models::ResponseItem;
     use codex_state::ThreadMetadata;
     use pretty_assertions::assert_eq;
     use std::fs;
@@ -529,4 +636,119 @@ mod tests {
         assert!(section.contains(&format!("### Directory: {}", outside.display())));
         assert!(section.contains(&format!("- {}: Inspect flaky test", outside.display())));
     }
+
+    #[test]
+    fn current_thread_section_includes_last_two_turns() {
+        let items = vec![
+            ResponseItem::Message {
+                id: None,
+                role: "user".to_string(),
+                content: vec![ContentItem::InputText {
+                    text: "first question".to_string(),
+                }],
+                end_turn: None,
+                phase: None,
+            },
+            ResponseItem::Message {
+                id: None,
+                role: "assistant".to_string(),
+                content: vec![ContentItem::OutputText {
+                    text: "first answer".to_string(),
+                }],
+                end_turn: None,
+                phase: None,
+            },
+            ResponseItem::Message {
+                id: None,
+                role: "user".to_string(),
+                content: vec![ContentItem::InputText {
+                    text: "second question".to_string(),
+                }],
+                end_turn: None,
+                phase: None,
+            },
+            ResponseItem::Message {
+                id: None,
+                role: "assistant".to_string(),
+                content: vec![ContentItem::OutputText {
+                    text: "second answer".to_string(),
+                }],
+                end_turn: None,
+                phase: None,
+            },
+            ResponseItem::Message {
+                id: None,
+                role: "user".to_string(),
+                content: vec![ContentItem::InputText {
+                    text: "third question".to_string(),
+                }],
+                end_turn: None,
+                phase: None,
+            },
+            ResponseItem::Message {
+                id: None,
+                role: "assistant".to_string(),
+                content: vec![ContentItem::OutputText {
+                    text: "third answer".to_string(),
+                }],
+                end_turn: None,
+                phase: None,
+            },
+        ];
+
+        let section = build_current_thread_section(&items).expect("current thread section");
+        assert!(section.contains("second question"));
+        assert!(section.contains("second answer"));
+        assert!(section.contains("third question"));
+        assert!(section.contains("third answer"));
+        assert!(!section.contains("first question"));
+        assert!(!section.contains("first answer"));
+    }
+
+    #[test]
+    fn current_thread_section_skips_contextual_user_messages() {
+        let items = vec![
+            ResponseItem::Message {
+                id: None,
+                role: "user".to_string(),
+                content: vec![ContentItem::InputText {
+                    text: ENVIRONMENT_CONTEXT_FRAGMENT.wrap("synthetic context".to_string()),
+                }],
+                end_turn: None,
+                phase: None,
+            },
+            ResponseItem::Message {
+                id: None,
+                role: "assistant".to_string(),
+                content: vec![ContentItem::OutputText {
+                    text: "assistant reply".to_string(),
+                }],
+                end_turn: None,
+                phase: None,
+            },
+            ResponseItem::Message {
+                id: None,
+                role: "user".to_string(),
+                content: vec![ContentItem::InputText {
+                    text: "real question".to_string(),
+                }],
+                end_turn: None,
+                phase: None,
+            },
+            ResponseItem::Message {
+                id: None,
+                role: "assistant".to_string(),
+                content: vec![ContentItem::OutputText {
+                    text: "real answer".to_string(),
+                }],
+                end_turn: None,
+                phase: None,
+            },
+        ];
+
+        let section = build_current_thread_section(&items).expect("current thread section");
+        assert!(!section.contains("synthetic context"));
+        assert!(section.contains("real question"));
+        assert!(section.contains("real answer"));
+    }
 }

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

@@ -920,6 +920,75 @@ async fn conversation_start_injects_startup_context_from_thread_history() -> Res
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn conversation_start_injects_recent_current_thread_turns_into_startup_context() -> Result<()>
+{
+    skip_if_no_network!(Ok(()));
+
+    let server = start_mock_server().await;
+    responses::mount_sse_once(
+        &server,
+        responses::sse(vec![
+            responses::ev_response_created("resp-thread-1"),
+            responses::ev_assistant_message("msg-thread-1", "first answer"),
+            responses::ev_completed("resp-thread-1"),
+        ]),
+    )
+    .await;
+    responses::mount_sse_once(
+        &server,
+        responses::sse(vec![
+            responses::ev_response_created("resp-thread-2"),
+            responses::ev_assistant_message("msg-thread-2", "second answer"),
+            responses::ev_completed("resp-thread-2"),
+        ]),
+    )
+    .await;
+
+    let realtime_server = start_websocket_server(vec![vec![vec![json!({
+        "type": "session.updated",
+        "session": { "id": "sess_current_thread", "instructions": "backend prompt" }
+    })]]])
+    .await;
+
+    let mut builder = test_codex().with_config({
+        let realtime_base_url = realtime_server.uri().to_string();
+        move |config| {
+            config.experimental_realtime_ws_base_url = Some(realtime_base_url);
+        }
+    });
+    let test = builder.build(&server).await?;
+
+    test.submit_turn("first question").await?;
+    test.submit_turn("second question").await?;
+
+    test.codex
+        .submit(Op::RealtimeConversationStart(ConversationStartParams {
+            prompt: "backend prompt".to_string(),
+            session_id: None,
+        }))
+        .await?;
+
+    let startup_context_request = wait_for_matching_websocket_request(
+        &realtime_server,
+        "current thread startup context request with instructions",
+        |request| websocket_request_instructions(request).is_some(),
+    )
+    .await;
+    let startup_context = websocket_request_instructions(&startup_context_request)
+        .expect("startup context request should contain instructions");
+
+    assert!(startup_context.contains("## Current Thread"));
+    assert!(startup_context.contains("Most recent user/assistant turns from this exact thread."));
+    assert!(startup_context.contains("first question"));
+    assert!(startup_context.contains("first answer"));
+    assert!(startup_context.contains("second question"));
+    assert!(startup_context.contains("second answer"));
+
+    realtime_server.shutdown().await;
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn conversation_startup_context_falls_back_to_workspace_map() -> Result<()> {
     skip_if_no_network!(Ok(()));

codex-rs/tui/src/chatwidget.rs

@@ -2086,6 +2086,8 @@ impl ChatWidget {
             self.pending_steers.clear();
             self.queued_user_messages.clear();
             self.refresh_pending_input_preview();
+            self.maybe_close_realtime_conversation_when_idle();
+            self.request_redraw();
             return;
         }
         if send_pending_steers_immediately {
@@ -3886,6 +3888,16 @@ impl ChatWidget {
                             .bottom_pane
                             .take_recent_submission_mention_bindings(),
                     };
+                    if self.realtime_conversation.is_live() {
+                        if user_message.text.is_empty()
+                            && user_message.local_images.is_empty()
+                            && user_message.remote_image_urls.is_empty()
+                        {
+                            return;
+                        }
+                        self.queue_user_message(user_message);
+                        return;
+                    }
                     let Some(user_message) =
                         self.maybe_submit_user_message_for_realtime(user_message)
                     else {

codex-rs/tui/src/chatwidget/realtime.rs

@@ -41,6 +41,7 @@ const REALTIME_CONVERSATION_PROMPT: &str = concat!(
     "Do not read long code snippets, long file paths, diffs, or command output aloud unless the ",
     "user explicitly asks for that."
 );
+const MAX_REALTIME_RECONNECT_ATTEMPTS: u32 = 3;
 
 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
 pub(super) enum RealtimeConversationPhase {
@@ -57,6 +58,7 @@ pub(super) struct RealtimeConversationUiState {
     requested_close: bool,
     close_when_idle: bool,
     session_id: Option<String>,
+    reconnect_attempts: u32,
     warned_unsupported_composer_submission: bool,
     meter_generation: u64,
     meter_text: Option<String>,
@@ -89,6 +91,10 @@ impl RealtimeConversationUiState {
         self.phase = phase;
     }
 
+    pub(super) fn set_session_id_for_test(&mut self, session_id: Option<String>) {
+        self.session_id = session_id;
+    }
+
     pub(super) fn set_meter_generation_for_test(&mut self, generation: u64) {
         self.meter_generation = generation;
     }
@@ -104,6 +110,10 @@ impl RealtimeConversationUiState {
     pub(super) fn close_when_idle_for_test(&self) -> bool {
         self.close_when_idle
     }
+
+    pub(super) fn reconnect_attempts_for_test(&self) -> u32 {
+        self.reconnect_attempts
+    }
 }
 
 #[derive(Clone, Debug, PartialEq)]
@@ -344,17 +354,21 @@ impl ChatWidget {
     }
 
     pub(super) fn start_realtime_conversation(&mut self) {
+        self.start_realtime_conversation_with_session_id(None);
+    }
+
+    fn start_realtime_conversation_with_session_id(&mut self, session_id: Option<String>) {
         self.realtime_conversation.phase = RealtimeConversationPhase::Starting;
         self.realtime_conversation.requested_close = false;
         self.realtime_conversation.close_when_idle = false;
-        self.realtime_conversation.session_id = None;
+        self.realtime_conversation.session_id = session_id.clone();
         self.realtime_conversation.meter_text = None;
         self.realtime_conversation
             .warned_unsupported_composer_submission = false;
         self.sync_realtime_status_label();
         self.submit_op(Op::RealtimeConversationStart(ConversationStartParams {
             prompt: REALTIME_CONVERSATION_PROMPT.to_string(),
-            session_id: None,
+            session_id,
         }));
         self.request_redraw();
     }
@@ -426,6 +440,7 @@ impl ChatWidget {
         self.realtime_conversation.requested_close = false;
         self.realtime_conversation.close_when_idle = false;
         self.realtime_conversation.session_id = None;
+        self.realtime_conversation.reconnect_attempts = 0;
         self.realtime_conversation.meter_text = None;
         self.realtime_conversation
             .warned_unsupported_composer_submission = false;
@@ -443,6 +458,7 @@ impl ChatWidget {
         }
         self.realtime_conversation.phase = RealtimeConversationPhase::Active;
         self.realtime_conversation.session_id = ev.session_id;
+        self.realtime_conversation.reconnect_attempts = 0;
         self.realtime_conversation
             .warned_unsupported_composer_submission = false;
         self.sync_realtime_status_label();
@@ -496,9 +512,24 @@ impl ChatWidget {
     pub(super) fn on_realtime_conversation_closed(&mut self, ev: RealtimeConversationClosedEvent) {
         let requested = self.realtime_conversation.requested_close;
         let reason = ev.reason;
+        let session_id = self.realtime_conversation.session_id.clone();
+        let next_reconnect_attempt = self.realtime_conversation.reconnect_attempts + 1;
+        let should_reconnect =
+            !requested && next_reconnect_attempt <= MAX_REALTIME_RECONNECT_ATTEMPTS;
         self.reset_realtime_conversation_state();
-        if !requested && let Some(reason) = reason {
-            self.add_info_message(format!("Realtime voice mode closed: {reason}"), None);
+        if should_reconnect {
+            self.realtime_conversation.reconnect_attempts = next_reconnect_attempt;
+            self.add_info_message(
+                format!(
+                    "Realtime voice mode connection dropped. Reconnecting (attempt {}/{MAX_REALTIME_RECONNECT_ATTEMPTS}).",
+                    self.realtime_conversation.reconnect_attempts
+                ),
+                None,
+            );
+            self.start_realtime_conversation_with_session_id(session_id);
+        } else if !requested {
+            let close_reason = reason.unwrap_or_else(|| "transport_closed".to_string());
+            self.add_info_message(format!("Realtime voice mode closed: {close_reason}"), None);
         }
         self.request_redraw();
     }

codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__realtime_queue_shortcut_pending_preview.snap

@@ -0,0 +1,14 @@
+---
+source: tui/src/chatwidget/tests.rs
+expression: term.backend().vt100().screen().contents()
+---
+
+• Working (0s • esc to interrupt)
+
+• Queued follow-up messages
+  ↳ queued via realtime
+    ⌥ + ↑ edit last queued message
+
+› Ask Codex to do anything
+
+  gpt-5.4 · /Users/pbakkum/code/codex

codex-rs/tui/src/chatwidget/tests.rs

@@ -65,6 +65,7 @@ use codex_protocol::protocol::BackgroundEventEvent;
 use codex_protocol::protocol::CodexErrorInfo;
 use codex_protocol::protocol::CollabAgentSpawnBeginEvent;
 use codex_protocol::protocol::CollabAgentSpawnEndEvent;
+use codex_protocol::protocol::ConversationStartParams;
 use codex_protocol::protocol::CreditsSnapshot;
 use codex_protocol::protocol::Event;
 use codex_protocol::protocol::EventMsg;
@@ -1965,6 +1966,23 @@ fn next_realtime_close_op(op_rx: &mut tokio::sync::mpsc::UnboundedReceiver<Op>)
     }
 }
 
+fn next_realtime_start_op(
+    op_rx: &mut tokio::sync::mpsc::UnboundedReceiver<Op>,
+) -> ConversationStartParams {
+    loop {
+        match op_rx.try_recv() {
+            Ok(Op::RealtimeConversationStart(params)) => return params,
+            Ok(_) => continue,
+            Err(TryRecvError::Empty) => {
+                panic!("expected realtime start op but queue was empty")
+            }
+            Err(TryRecvError::Disconnected) => {
+                panic!("expected realtime start op but channel closed")
+            }
+        }
+    }
+}
+
 fn next_app_event(rx: &mut tokio::sync::mpsc::UnboundedReceiver<AppEvent>) -> AppEvent {
     rx.try_recv().expect("expected app event")
 }
@@ -10506,7 +10524,7 @@ async fn realtime_recording_meter_renders_in_footer_snapshot() {
 }
 
 #[tokio::test]
-async fn queue_shortcut_submits_realtime_text_while_realtime_mode_is_live() {
+async fn queue_shortcut_queues_local_draft_while_realtime_mode_is_live() {
     let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(None).await;
     chat.realtime_conversation
         .set_phase_for_test(super::realtime::RealtimeConversationPhase::Active);
@@ -10517,16 +10535,46 @@ async fn queue_shortcut_submits_realtime_text_while_realtime_mode_is_live() {
     chat.handle_key_event(KeyEvent::new(KeyCode::Tab, KeyModifiers::NONE));
 
     assert_eq!(chat.bottom_pane.composer_text(), "");
-    assert!(chat.queued_user_messages.is_empty());
-    assert_eq!(next_realtime_text(&mut op_rx), "queued via realtime");
+    assert_eq!(chat.queued_user_messages.len(), 1);
+    assert_eq!(
+        chat.queued_user_messages
+            .front()
+            .map(|message| message.text.as_str()),
+        Some("queued via realtime")
+    );
+    assert_matches!(op_rx.try_recv(), Err(TryRecvError::Empty));
+    assert!(drain_insert_history(&mut rx).is_empty());
+}
 
-    let inserted = drain_insert_history(&mut rx);
-    let rendered = inserted
-        .iter()
-        .map(|lines| lines_to_single_string(lines))
-        .find(|text| text.contains("queued via realtime"))
-        .expect("expected rendered realtime text prompt");
-    assert!(rendered.contains("queued via realtime"));
+#[tokio::test]
+async fn realtime_queue_shortcut_renders_pending_preview_snapshot() {
+    let (mut chat, _rx, _op_rx) = make_chatwidget_manual(None).await;
+    chat.realtime_conversation
+        .set_phase_for_test(super::realtime::RealtimeConversationPhase::Active);
+    chat.bottom_pane.set_status_line_enabled(true);
+    chat.set_status_line(Some(ratatui::text::Line::from(
+        "gpt-5.4 · /Users/pbakkum/code/codex",
+    )));
+    chat.on_task_started();
+
+    chat.bottom_pane
+        .set_composer_text("queued via realtime".to_string(), Vec::new(), Vec::new());
+    chat.handle_key_event(KeyEvent::new(KeyCode::Tab, KeyModifiers::NONE));
+
+    let width: u16 = 80;
+    let height: u16 = 10;
+    let backend = VT100Backend::new(width, height);
+    let mut term = crate::custom_terminal::Terminal::with_options(backend).expect("terminal");
+    let desired_height = chat.desired_height(width).min(height);
+    term.set_viewport_area(Rect::new(0, height - desired_height, width, desired_height));
+    term.draw(|f| {
+        chat.render(f.area(), f.buffer_mut());
+    })
+    .unwrap();
+    assert_snapshot!(
+        "realtime_queue_shortcut_pending_preview",
+        term.backend().vt100().screen().contents()
+    );
 }
 
 #[tokio::test]
@@ -10722,6 +10770,64 @@ async fn realtime_close_request_closes_immediately_when_idle() {
     assert!(rendered.contains("Realtime voice mode turned off."));
 }
 
+#[tokio::test]
+async fn unexpected_realtime_close_restarts_with_last_known_session_id() {
+    let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(None).await;
+
+    chat.realtime_conversation
+        .set_phase_for_test(super::realtime::RealtimeConversationPhase::Active);
+    chat.realtime_conversation
+        .set_session_id_for_test(Some("sess_reconnect".to_string()));
+
+    chat.on_realtime_conversation_closed(
+        codex_protocol::protocol::RealtimeConversationClosedEvent {
+            reason: Some("transport_closed".to_string()),
+        },
+    );
+
+    assert_eq!(
+        chat.realtime_conversation.phase_for_test(),
+        super::realtime::RealtimeConversationPhase::Starting
+    );
+    assert_eq!(chat.realtime_conversation.reconnect_attempts_for_test(), 1);
+    let restart = next_realtime_start_op(&mut op_rx);
+    assert_eq!(restart.session_id.as_deref(), Some("sess_reconnect"));
+    assert!(restart.prompt.contains("realtime voice conversation"));
+
+    let inserted = drain_insert_history(&mut rx);
+    let rendered = inserted
+        .iter()
+        .map(|lines| lines_to_single_string(lines))
+        .find(|text| text.contains("Realtime voice mode connection dropped. Reconnecting"))
+        .expect("expected reconnect info message");
+    assert!(rendered.contains("attempt 1/3"));
+}
+
+#[tokio::test]
+async fn requested_realtime_close_does_not_restart() {
+    let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(None).await;
+
+    chat.realtime_conversation
+        .set_phase_for_test(super::realtime::RealtimeConversationPhase::Active);
+    chat.realtime_conversation
+        .set_session_id_for_test(Some("sess_requested_close".to_string()));
+    chat.request_realtime_conversation_close(None);
+    next_realtime_close_op(&mut op_rx);
+
+    chat.on_realtime_conversation_closed(
+        codex_protocol::protocol::RealtimeConversationClosedEvent {
+            reason: Some("transport_closed".to_string()),
+        },
+    );
+
+    assert_eq!(
+        chat.realtime_conversation.phase_for_test(),
+        super::realtime::RealtimeConversationPhase::Inactive
+    );
+    assert!(matches!(op_rx.try_recv(), Err(TryRecvError::Empty)));
+    assert!(drain_insert_history(&mut rx).is_empty());
+}
+
 #[tokio::test]
 async fn realtime_manage_message_queue_updates_queue_and_completes_tool_call() {
     let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await;