diff --git a/codex-rs/core/src/realtime_context.rs b/codex-rs/core/src/realtime_context.rs index e15adabc4e..703fa93f1f 100644 --- a/codex-rs/core/src/realtime_context.rs +++ b/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 { 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 { + 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::>() + .into_iter() + .rev() + .collect::>(); + 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 { 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")); + } } diff --git a/codex-rs/core/tests/suite/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index da9c341bd2..f7f237742f 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/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(())); diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index 06551807fa..80e1c1ec8a 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/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 { diff --git a/codex-rs/tui/src/chatwidget/realtime.rs b/codex-rs/tui/src/chatwidget/realtime.rs index 91b50deb4c..d5992e9606 100644 --- a/codex-rs/tui/src/chatwidget/realtime.rs +++ b/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, + reconnect_attempts: u32, warned_unsupported_composer_submission: bool, meter_generation: u64, meter_text: Option, @@ -89,6 +91,10 @@ impl RealtimeConversationUiState { self.phase = phase; } + pub(super) fn set_session_id_for_test(&mut self, session_id: Option) { + 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) { 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(); } diff --git a/codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__realtime_queue_shortcut_pending_preview.snap b/codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__realtime_queue_shortcut_pending_preview.snap new file mode 100644 index 0000000000..1db2522f45 --- /dev/null +++ b/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 diff --git a/codex-rs/tui/src/chatwidget/tests.rs b/codex-rs/tui/src/chatwidget/tests.rs index 7825da3bc9..0923389913 100644 --- a/codex-rs/tui/src/chatwidget/tests.rs +++ b/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) } } +fn next_realtime_start_op( + op_rx: &mut tokio::sync::mpsc::UnboundedReceiver, +) -> 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 { 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;