From 709916ad10b3f8e71051aace123e263d508732dd Mon Sep 17 00:00:00 2001 From: Eric Traut Date: Sat, 14 Mar 2026 10:19:57 -0600 Subject: [PATCH] Part 1 --- codex-rs/tui/src/app.rs | 734 ++++++------ codex-rs/tui/src/app/app_server_adapter.rs | 1267 +++++++++++++++++++- codex-rs/tui/src/chatwidget.rs | 149 +-- codex-rs/tui/src/chatwidget/agent.rs | 16 + codex-rs/tui/src/chatwidget/tests.rs | 113 +- codex-rs/tui/src/collaboration_modes.rs | 43 +- 6 files changed, 1817 insertions(+), 505 deletions(-) diff --git a/codex-rs/tui/src/app.rs b/codex-rs/tui/src/app.rs index ca27a2ad1e..25b4c66c5d 100644 --- a/codex-rs/tui/src/app.rs +++ b/codex-rs/tui/src/app.rs @@ -41,9 +41,8 @@ use crate::version::CODEX_CLI_VERSION; use codex_ansi_escape::ansi_escape_line; use codex_app_server_client::InProcessAppServerClient; use codex_app_server_protocol::ConfigLayerSource; +use codex_app_server_protocol::RequestId; use codex_core::AuthManager; -use codex_core::CodexAuth; -use codex_core::ThreadManager; use codex_core::config::Config; use codex_core::config::ConfigBuilder; use codex_core::config::ConfigOverrides; @@ -53,7 +52,6 @@ use codex_core::config::types::ApprovalsReviewer; use codex_core::config::types::ModelAvailabilityNuxConfig; use codex_core::config_loader::ConfigLayerStackOrdering; use codex_core::features::Feature; -use codex_core::models_manager::manager::RefreshStrategy; use codex_core::models_manager::model_presets::HIDE_GPT_5_1_CODEX_MAX_MIGRATION_PROMPT_CONFIG; use codex_core::models_manager::model_presets::HIDE_GPT5_1_MIGRATION_PROMPT_CONFIG; #[cfg(target_os = "windows")] @@ -61,6 +59,7 @@ use codex_core::windows_sandbox::WindowsSandboxLevelExt; use codex_otel::SessionTelemetry; use codex_otel::TelemetryAuthMode; use codex_protocol::ThreadId; +use codex_protocol::config_types::CollaborationModeMask; use codex_protocol::config_types::Personality; #[cfg(target_os = "windows")] use codex_protocol::config_types::WindowsSandboxLevel; @@ -104,7 +103,6 @@ use std::time::Duration; use std::time::Instant; use tokio::select; use tokio::sync::Mutex; -use tokio::sync::broadcast; use tokio::sync::mpsc; use tokio::sync::mpsc::error::TryRecvError; use tokio::sync::mpsc::error::TrySendError; @@ -660,11 +658,12 @@ async fn handle_model_migration_prompt_if_needed( } pub(crate) struct App { - pub(crate) server: Arc, pub(crate) session_telemetry: SessionTelemetry, pub(crate) app_event_tx: AppEventSender, pub(crate) chat_widget: ChatWidget, pub(crate) auth_manager: Arc, + available_models: Vec, + available_collaboration_modes: Vec, /// Config is stored here so we can recreate ChatWidgets as needed. pub(crate) config: Config, pub(crate) active_profile: Option, @@ -727,6 +726,15 @@ pub(crate) struct App { primary_thread_id: Option, primary_session_configured: Option, pending_primary_events: VecDeque, + next_app_server_request_id: i64, + active_turn_ids: HashMap, + pending_exec_approval_request_ids: HashMap, + pending_patch_approval_request_ids: HashMap, + pending_elicitation_request_ids: HashMap<(String, String), RequestId>, + pending_elicitation_request_queues: HashMap>, + pending_permissions_request_ids: HashMap, + pending_user_input_request_ids: HashMap, + pending_dynamic_tool_request_ids: HashMap, } #[derive(Default)] @@ -767,7 +775,8 @@ impl App { initial_user_message: None, enhanced_keys_supported: self.enhanced_keys_supported, auth_manager: self.auth_manager.clone(), - models_manager: self.server.get_models_manager(), + available_models: self.available_models.clone(), + available_collaboration_modes: self.available_collaboration_modes.clone(), feedback: self.feedback.clone(), is_first_run: false, feedback_audience: self.feedback_audience, @@ -1207,14 +1216,18 @@ impl App { self.backtrack_render_pending = false; } - async fn shutdown_current_thread(&mut self) { + async fn shutdown_current_thread(&mut self, app_server: &InProcessAppServerClient) { if let Some(thread_id) = self.chat_widget.thread_id() { // Clear any in-flight rollback guard when switching threads. self.backtrack.pending_rollback = None; self.suppress_shutdown_complete = true; - self.chat_widget.submit_op(Op::Shutdown); - self.server.remove_thread(&thread_id).await; - self.abort_thread_event_listener(thread_id); + if let Err(err) = self + .unsubscribe_thread_via_app_server(app_server, thread_id) + .await + { + self.chat_widget + .add_error_message(format!("Failed to shut down thread {thread_id}: {err}")); + } } } @@ -1437,31 +1450,16 @@ impl App { } } - async fn submit_op_to_thread(&mut self, thread_id: ThreadId, op: Op) { + async fn submit_op_to_thread( + &mut self, + app_server: &InProcessAppServerClient, + thread_id: ThreadId, + op: Op, + ) { let replay_state_op = ThreadEventStore::op_can_change_pending_replay_state(&op).then(|| op.clone()); - let submitted = if self.active_thread_id == Some(thread_id) { - self.chat_widget.submit_op(op) - } else { - crate::session_log::log_outbound_op(&op); - match self.server.get_thread(thread_id).await { - Ok(thread) => match thread.submit(op).await { - Ok(_) => true, - Err(err) => { - self.chat_widget.add_error_message(format!( - "Failed to submit op to thread {thread_id}: {err}" - )); - false - } - }, - Err(err) => { - self.chat_widget.add_error_message(format!( - "Failed to find thread {thread_id} for approval response: {err}" - )); - false - } - } - }; + crate::session_log::log_outbound_op(&op); + let submitted = self.submit_app_server_op(app_server, thread_id, op).await; if submitted && let Some(op) = replay_state_op.as_ref() { self.note_thread_outbound_op(thread_id, op).await; self.refresh_pending_thread_approvals().await; @@ -1557,8 +1555,13 @@ impl App { event: Event, ) -> Result<()> { if !self.thread_event_channels.contains_key(&thread_id) { - tracing::debug!("dropping stale event for untracked thread {thread_id}"); - return Ok(()); + let should_attach_channel = matches!(event.msg, EventMsg::SessionConfigured(_)) + || self.agent_navigation.get(&thread_id).is_some(); + if !should_attach_channel { + tracing::debug!("dropping routed event for unknown thread {thread_id}"); + return Ok(()); + } + self.ensure_thread_channel(thread_id); } self.enqueue_thread_event(thread_id, event).await @@ -1595,22 +1598,17 @@ impl App { /// historical id now" and converted into closed picker entries instead of deleting them, so /// the stable traversal order remains intact for review and keyboard navigation. async fn open_agent_picker(&mut self) { - let thread_ids: Vec = self.thread_event_channels.keys().cloned().collect(); + let thread_ids: Vec<_> = self.thread_event_channels.keys().copied().collect(); for thread_id in thread_ids { - match self.server.get_thread(thread_id).await { - Ok(thread) => { - let session_source = thread.config_snapshot().await.session_source; - self.upsert_agent_picker_thread( - thread_id, - session_source.get_nickname(), - session_source.get_agent_role(), - false, - ); - } - Err(_) => { - self.mark_agent_picker_thread_closed(thread_id); - } - } + let existing = self.agent_navigation.get(&thread_id).cloned(); + self.upsert_agent_picker_thread( + thread_id, + existing + .as_ref() + .and_then(|entry| entry.agent_nickname.clone()), + existing.as_ref().and_then(|entry| entry.agent_role.clone()), + self.active_thread_id != Some(thread_id), + ); } let has_non_primary_agent_thread = self @@ -1700,21 +1698,10 @@ impl App { return Ok(()); } - let live_thread = match self.server.get_thread(thread_id).await { - Ok(thread) => Some(thread), - Err(err) => { - if self.thread_event_channels.contains_key(&thread_id) { - self.mark_agent_picker_thread_closed(thread_id); - None - } else { - self.chat_widget.add_error_message(format!( - "Failed to attach to agent thread {thread_id}: {err}" - )); - return Ok(()); - } - } - }; - let is_replay_only = live_thread.is_none(); + let is_replay_only = self + .agent_navigation + .get(&thread_id) + .is_some_and(|entry| entry.is_closed); let previous_thread_id = self.active_thread_id; self.store_active_thread_receiver().await; @@ -1732,12 +1719,7 @@ impl App { self.active_thread_rx = Some(receiver); let init = self.chatwidget_init_for_forked_or_resumed_thread(tui, self.config.clone()); - let codex_op_tx = if let Some(thread) = live_thread { - crate::chatwidget::spawn_op_forwarder(thread) - } else { - let (tx, _rx) = unbounded_channel(); - tx - }; + let codex_op_tx = crate::chatwidget::spawn_app_event_forwarder(self.app_event_tx.clone()); self.chat_widget = ChatWidget::new_with_op_sender(init, codex_op_tx); self.sync_active_agent_label(); @@ -1779,7 +1761,11 @@ impl App { self.sync_active_agent_label(); } - async fn start_fresh_session_with_summary_hint(&mut self, tui: &mut tui::Tui) { + async fn start_fresh_session_with_summary_hint( + &mut self, + tui: &mut tui::Tui, + app_server: &InProcessAppServerClient, + ) { // Start a fresh in-memory session while preserving resumability via persisted rollout // history. self.refresh_in_memory_config_from_disk_best_effort("starting a new thread") @@ -1791,18 +1777,24 @@ impl App { self.chat_widget.thread_id(), self.chat_widget.thread_name(), ); - self.shutdown_current_thread().await; - let report = self - .server - .shutdown_all_threads_bounded(Duration::from_secs(10)) - .await; - if !report.submit_failed.is_empty() || !report.timed_out.is_empty() { - tracing::warn!( - submit_failed = report.submit_failed.len(), - timed_out = report.timed_out.len(), - "failed to close all threads" - ); + self.shutdown_current_thread(app_server).await; + let thread_ids: Vec = self.thread_event_channels.keys().copied().collect(); + for thread_id in thread_ids { + if let Err(err) = self + .unsubscribe_thread_via_app_server(app_server, thread_id) + .await + { + tracing::warn!(thread_id = %thread_id, error = %err, "failed to close thread"); + } } + let session_configured = match self.thread_start_via_app_server(app_server, &config).await { + Ok(session_configured) => session_configured, + Err(err) => { + self.chat_widget + .add_error_message(format!("Failed to start a new session: {err}")); + return; + } + }; let init = crate::chatwidget::ChatWidgetInit { config, frame_requester: tui.frame_requester(), @@ -1811,7 +1803,8 @@ impl App { initial_user_message: None, enhanced_keys_supported: self.enhanced_keys_supported, auth_manager: self.auth_manager.clone(), - models_manager: self.server.get_models_manager(), + available_models: self.available_models.clone(), + available_collaboration_modes: self.available_collaboration_modes.clone(), feedback: self.feedback.clone(), is_first_run: false, feedback_audience: self.feedback_audience, @@ -1820,8 +1813,21 @@ impl App { status_line_invalid_items_warned: self.status_line_invalid_items_warned.clone(), session_telemetry: self.session_telemetry.clone(), }; - self.chat_widget = ChatWidget::new(init, self.server.clone()); + self.chat_widget = ChatWidget::new_with_op_sender( + init, + crate::chatwidget::spawn_app_event_forwarder(self.app_event_tx.clone()), + ); self.reset_thread_event_state(); + if let Err(err) = self + .enqueue_primary_event(Event { + id: String::new(), + msg: EventMsg::SessionConfigured(session_configured), + }) + .await + { + self.chat_widget + .add_error_message(format!("Failed to initialize new session state: {err}")); + } if let Some(summary) = summary { let mut lines: Vec> = vec![summary.usage_line.clone().into()]; if let Some(command) = summary.resume_command { @@ -1957,15 +1963,26 @@ impl App { let harness_overrides = normalize_harness_overrides_for_cwd(harness_overrides, &config.cwd)?; let auth_manager = app_server.auth_manager(); - let thread_manager = app_server.thread_manager(); - let mut model = thread_manager - .get_models_manager() - .get_default_model(&config.model, RefreshStrategy::Offline) - .await; - let available_models = thread_manager - .get_models_manager() - .list_models(RefreshStrategy::Offline) - .await; + let available_models = App::list_models_via_app_server(&app_server) + .await + .map_err(|err| color_eyre::eyre::eyre!(err))?; + let available_collaboration_modes = + App::list_collaboration_modes_via_app_server(&app_server) + .await + .map_err(|err| color_eyre::eyre::eyre!(err))?; + let mut model = available_models + .iter() + .find(|preset| { + config + .model + .as_deref() + .is_some_and(|configured| preset.model == configured) + }) + .or_else(|| available_models.iter().find(|preset| preset.is_default)) + .or_else(|| available_models.iter().find(|preset| preset.show_in_picker)) + .map(|preset| preset.model.clone()) + .or_else(|| config.model.clone()) + .unwrap_or_default(); let exit_info = handle_model_migration_prompt_if_needed( tui, &mut config, @@ -1987,28 +2004,35 @@ impl App { if let Some(updated_model) = config.model.clone() { model = updated_model; } - let auth = auth_manager.auth().await; - let auth_ref = auth.as_ref(); + let account = App::read_account_via_app_server(&app_server) + .await + .map_err(|err| color_eyre::eyre::eyre!(err))?; // Determine who should see internal Slack routing. We treat // `@openai.com` emails as employees and default to `External` when the // email is unavailable (for example, API key auth). - let feedback_audience = if auth_ref - .and_then(CodexAuth::get_account_email) + let account_email = account.as_ref().and_then(|account| match account { + codex_app_server_protocol::Account::Chatgpt { email, .. } => Some(email.clone()), + codex_app_server_protocol::Account::ApiKey {} => None, + }); + let account_id = None; + let auth_mode = account.as_ref().map(|account| match account { + codex_app_server_protocol::Account::ApiKey {} => TelemetryAuthMode::ApiKey, + codex_app_server_protocol::Account::Chatgpt { .. } => TelemetryAuthMode::Chatgpt, + }); + let feedback_audience = if account_email + .as_deref() .is_some_and(|email| email.ends_with("@openai.com")) { FeedbackAudience::OpenAiEmployee } else { FeedbackAudience::External }; - let auth_mode = auth_ref - .map(CodexAuth::auth_mode) - .map(TelemetryAuthMode::from); let session_telemetry = SessionTelemetry::new( ThreadId::new(), model.as_str(), model.as_str(), - auth_ref.and_then(CodexAuth::get_account_id), - auth_ref.and_then(CodexAuth::get_account_email), + account_id, + account_email, auth_mode, codex_core::default_client::originator().value, config.otel.log_user_prompt, @@ -2028,109 +2052,37 @@ impl App { let enhanced_keys_supported = tui.enhanced_keys_supported(); let wait_for_initial_session_configured = Self::should_wait_for_initial_session(&session_selection); - let mut chat_widget = match session_selection { - SessionSelection::StartFresh | SessionSelection::Exit => { - let startup_tooltip_override = - prepare_startup_tooltip_override(&mut config, &available_models, is_first_run) - .await; - let init = crate::chatwidget::ChatWidgetInit { - config: config.clone(), - frame_requester: tui.frame_requester(), - app_event_tx: app_event_tx.clone(), - initial_user_message: crate::chatwidget::create_initial_user_message( - initial_prompt.clone(), - initial_images.clone(), - // CLI prompt args are plain strings, so they don't provide element ranges. - Vec::new(), - ), - enhanced_keys_supported, - auth_manager: auth_manager.clone(), - models_manager: thread_manager.get_models_manager(), - feedback: feedback.clone(), - is_first_run, - feedback_audience, - model: Some(model.clone()), - startup_tooltip_override, - status_line_invalid_items_warned: status_line_invalid_items_warned.clone(), - session_telemetry: session_telemetry.clone(), - }; - ChatWidget::new(init, thread_manager.clone()) - } - SessionSelection::Resume(target_session) => { - let resumed = thread_manager - .resume_thread_from_rollout( - config.clone(), - target_session.path.clone(), - auth_manager.clone(), - None, - ) - .await - .wrap_err_with(|| { - let path_display = target_session.path.display(); - format!("Failed to resume session from {path_display}") - })?; - let init = crate::chatwidget::ChatWidgetInit { - config: config.clone(), - frame_requester: tui.frame_requester(), - app_event_tx: app_event_tx.clone(), - initial_user_message: crate::chatwidget::create_initial_user_message( - initial_prompt.clone(), - initial_images.clone(), - // CLI prompt args are plain strings, so they don't provide element ranges. - Vec::new(), - ), - enhanced_keys_supported, - auth_manager: auth_manager.clone(), - models_manager: thread_manager.get_models_manager(), - feedback: feedback.clone(), - is_first_run, - feedback_audience, - model: config.model.clone(), - startup_tooltip_override: None, - status_line_invalid_items_warned: status_line_invalid_items_warned.clone(), - session_telemetry: session_telemetry.clone(), - }; - ChatWidget::new_from_existing(init, resumed.thread, resumed.session_configured) - } - SessionSelection::Fork(target_session) => { - session_telemetry.counter("codex.thread.fork", 1, &[("source", "cli_subcommand")]); - let forked = thread_manager - .fork_thread( - usize::MAX, - config.clone(), - target_session.path.clone(), - false, - None, - ) - .await - .wrap_err_with(|| { - let path_display = target_session.path.display(); - format!("Failed to fork session from {path_display}") - })?; - let init = crate::chatwidget::ChatWidgetInit { - config: config.clone(), - frame_requester: tui.frame_requester(), - app_event_tx: app_event_tx.clone(), - initial_user_message: crate::chatwidget::create_initial_user_message( - initial_prompt.clone(), - initial_images.clone(), - // CLI prompt args are plain strings, so they don't provide element ranges. - Vec::new(), - ), - enhanced_keys_supported, - auth_manager: auth_manager.clone(), - models_manager: thread_manager.get_models_manager(), - feedback: feedback.clone(), - is_first_run, - feedback_audience, - model: config.model.clone(), - startup_tooltip_override: None, - status_line_invalid_items_warned: status_line_invalid_items_warned.clone(), - session_telemetry: session_telemetry.clone(), - }; - ChatWidget::new_from_existing(init, forked.thread, forked.session_configured) - } + let shared_op_tx = crate::chatwidget::spawn_app_event_forwarder(app_event_tx.clone()); + let startup_tooltip_override = if matches!( + session_selection, + SessionSelection::StartFresh | SessionSelection::Exit + ) { + prepare_startup_tooltip_override(&mut config, &available_models, is_first_run).await + } else { + None }; + let init = crate::chatwidget::ChatWidgetInit { + config: config.clone(), + frame_requester: tui.frame_requester(), + app_event_tx: app_event_tx.clone(), + initial_user_message: crate::chatwidget::create_initial_user_message( + initial_prompt.clone(), + initial_images.clone(), + Vec::new(), + ), + enhanced_keys_supported, + auth_manager: auth_manager.clone(), + available_models: available_models.clone(), + available_collaboration_modes: available_collaboration_modes.clone(), + feedback: feedback.clone(), + is_first_run, + feedback_audience, + model: Some(model.clone()), + startup_tooltip_override, + status_line_invalid_items_warned: status_line_invalid_items_warned.clone(), + session_telemetry: session_telemetry.clone(), + }; + let mut chat_widget = ChatWidget::new_with_op_sender(init, shared_op_tx); chat_widget .maybe_prompt_windows_sandbox_enable(should_prompt_windows_sandbox_nux_at_startup); @@ -2140,11 +2092,12 @@ impl App { let upgrade_version = crate::updates::get_upgrade_version(&config); let mut app = Self { - server: thread_manager.clone(), session_telemetry: session_telemetry.clone(), app_event_tx, chat_widget, auth_manager: auth_manager.clone(), + available_models, + available_collaboration_modes, config, active_profile, cli_kv_overrides, @@ -2175,8 +2128,68 @@ impl App { primary_thread_id: None, primary_session_configured: None, pending_primary_events: VecDeque::new(), + next_app_server_request_id: 1, + active_turn_ids: HashMap::new(), + pending_exec_approval_request_ids: HashMap::new(), + pending_patch_approval_request_ids: HashMap::new(), + pending_elicitation_request_ids: HashMap::new(), + pending_elicitation_request_queues: HashMap::new(), + pending_permissions_request_ids: HashMap::new(), + pending_user_input_request_ids: HashMap::new(), + pending_dynamic_tool_request_ids: HashMap::new(), }; + let initial_session_configured = match &session_selection { + SessionSelection::StartFresh | SessionSelection::Exit => { + let startup_config = app.config.clone(); + Some( + app.thread_start_via_app_server(&app_server, &startup_config) + .await + .map_err(|err| color_eyre::eyre::eyre!(err))?, + ) + } + SessionSelection::Resume(target_session) => { + let resume_config = app.config.clone(); + Some( + app.thread_resume_via_app_server( + &app_server, + &resume_config, + target_session.path.clone(), + ) + .await + .map_err(|err| { + let path_display = target_session.path.display(); + color_eyre::eyre::eyre!( + "Failed to resume session from {path_display}: {err}" + ) + })?, + ) + } + SessionSelection::Fork(target_session) => { + session_telemetry.counter("codex.thread.fork", 1, &[("source", "cli_subcommand")]); + let fork_config = app.config.clone(); + Some( + app.thread_fork_via_app_server( + &app_server, + &fork_config, + target_session.path.clone(), + ) + .await + .map_err(|err| { + let path_display = target_session.path.display(); + color_eyre::eyre::eyre!("Failed to fork session from {path_display}: {err}") + })?, + ) + } + }; + if let Some(session_configured) = initial_session_configured { + app.enqueue_primary_event(Event { + id: String::new(), + msg: EventMsg::SessionConfigured(session_configured), + }) + .await?; + } + // On startup, if Agent mode (workspace-write) or ReadOnly is active, warn about world-writable dirs on Windows. #[cfg(target_os = "windows")] { @@ -2207,8 +2220,6 @@ impl App { tui.frame_requester().schedule_frame(); - let mut thread_created_rx = thread_manager.subscribe_thread_created(); - let mut listen_for_threads = true; let mut listen_for_app_server_events = true; let mut waiting_for_initial_session_configured = wait_for_initial_session_configured; @@ -2217,6 +2228,7 @@ impl App { let control = app .handle_event( tui, + &app_server, AppEvent::InsertHistoryCell(Box::new(UpdateAvailableHistoryCell::new( latest_version, crate::update_action::get_update_action(), @@ -2239,7 +2251,7 @@ impl App { loop { let control = select! { Some(event) = app_event_rx.recv() => { - match app.handle_event(tui, event).await { + match app.handle_event(tui, &app_server, event).await { Ok(control) => control, Err(err) => break Err(err), } @@ -2279,23 +2291,6 @@ impl App { } AppRunControl::Continue } - // Listen on new thread creation due to collab tools. - created = thread_created_rx.recv(), if listen_for_threads => { - match created { - Ok(thread_id) => { - if let Err(err) = app.handle_thread_created(thread_id).await { - break Err(err); - } - } - Err(broadcast::error::RecvError::Lagged(_)) => { - tracing::warn!("thread_created receiver lagged; skipping resync"); - } - Err(broadcast::error::RecvError::Closed) => { - listen_for_threads = false; - } - } - AppRunControl::Continue - } }; if App::should_stop_waiting_for_initial_session( waiting_for_initial_session_configured, @@ -2395,16 +2390,23 @@ impl App { Ok(AppRunControl::Continue) } - async fn handle_event(&mut self, tui: &mut tui::Tui, event: AppEvent) -> Result { + async fn handle_event( + &mut self, + tui: &mut tui::Tui, + app_server: &InProcessAppServerClient, + event: AppEvent, + ) -> Result { match event { AppEvent::NewSession => { - self.start_fresh_session_with_summary_hint(tui).await; + self.start_fresh_session_with_summary_hint(tui, app_server) + .await; } AppEvent::ClearUi => { self.clear_terminal_ui(tui, false)?; self.reset_app_ui_state_after_clear(); - self.start_fresh_session_with_summary_hint(tui).await; + self.start_fresh_session_with_summary_hint(tui, app_server) + .await; } AppEvent::OpenResumePicker => { match crate::resume_picker::run_resume_picker(tui, &self.config, false).await? { @@ -2446,17 +2448,15 @@ impl App { self.chat_widget.thread_name(), ); match self - .server - .resume_thread_from_rollout( - resume_config.clone(), + .thread_resume_via_app_server( + app_server, + &resume_config, target_session.path.clone(), - self.auth_manager.clone(), - None, ) .await { - Ok(resumed) => { - self.shutdown_current_thread().await; + Ok(session_configured) => { + self.shutdown_current_thread(app_server).await; self.config = resume_config; tui.set_notification_method(self.config.tui_notification_method); self.file_search.update_search_dir(self.config.cwd.clone()); @@ -2464,12 +2464,18 @@ impl App { tui, self.config.clone(), ); - self.chat_widget = ChatWidget::new_from_existing( + self.chat_widget = ChatWidget::new_with_op_sender( init, - resumed.thread, - resumed.session_configured, + crate::chatwidget::spawn_app_event_forwarder( + self.app_event_tx.clone(), + ), ); self.reset_thread_event_state(); + self.enqueue_primary_event(Event { + id: String::new(), + msg: EventMsg::SessionConfigured(session_configured), + }) + .await?; if let Some(summary) = summary { let mut lines: Vec> = vec![summary.usage_line.clone().into()]; @@ -2518,23 +2524,29 @@ impl App { // Fresh threads expose a precomputed path, but the file is // materialized lazily on first user message. if path.exists() { + let fork_config = self.config.clone(); match self - .server - .fork_thread(usize::MAX, self.config.clone(), path.clone(), false, None) + .thread_fork_via_app_server(app_server, &fork_config, path.clone()) .await { - Ok(forked) => { - self.shutdown_current_thread().await; + Ok(session_configured) => { + self.shutdown_current_thread(app_server).await; let init = self.chatwidget_init_for_forked_or_resumed_thread( tui, self.config.clone(), ); - self.chat_widget = ChatWidget::new_from_existing( + self.chat_widget = ChatWidget::new_with_op_sender( init, - forked.thread, - forked.session_configured, + crate::chatwidget::spawn_app_event_forwarder( + self.app_event_tx.clone(), + ), ); self.reset_thread_event_state(); + self.enqueue_primary_event(Event { + id: String::new(), + msg: EventMsg::SessionConfigured(session_configured), + }) + .await?; if let Some(summary) = summary { let mut lines: Vec> = vec![summary.usage_line.clone().into()]; @@ -2630,7 +2642,7 @@ impl App { self.handle_routed_thread_event(thread_id, event).await?; } AppEvent::Exit(mode) => { - return Ok(self.handle_exit_mode(mode)); + return Ok(self.handle_exit_mode(app_server, mode).await); } AppEvent::FatalExitRequest(message) => { return Ok(AppRunControl::Exit(ExitReason::Fatal(message))); @@ -2638,14 +2650,22 @@ impl App { AppEvent::CodexOp(op) => { let replay_state_op = ThreadEventStore::op_can_change_pending_replay_state(&op).then(|| op.clone()); - let submitted = self.chat_widget.submit_op(op); + let submitted = if let Some(thread_id) = + self.active_thread_id.or(self.chat_widget.thread_id()) + { + self.submit_app_server_op(app_server, thread_id, op).await + } else { + self.chat_widget + .add_error_message("No active thread is available.".to_string()); + false + }; if submitted && let Some(op) = replay_state_op.as_ref() { self.note_active_thread_outbound_op(op).await; self.refresh_pending_thread_approvals().await; } } AppEvent::SubmitThreadOp { thread_id, op } => { - self.submit_op_to_thread(thread_id, op).await; + self.submit_op_to_thread(app_server, thread_id, op).await; } AppEvent::DiffResult(text) => { // Clear the in-progress state in the bottom pane @@ -3711,14 +3731,24 @@ impl App { Ok(AppRunControl::Continue) } - fn handle_exit_mode(&mut self, mode: ExitMode) -> AppRunControl { + async fn handle_exit_mode( + &mut self, + app_server: &InProcessAppServerClient, + mode: ExitMode, + ) -> AppRunControl { match mode { ExitMode::ShutdownFirst => { // Mark the thread we are explicitly shutting down for exit so // its shutdown completion does not trigger agent failover. self.pending_shutdown_exit_thread_id = self.active_thread_id.or(self.chat_widget.thread_id()); - if self.chat_widget.submit_op(Op::Shutdown) { + let submitted = if let Some(thread_id) = self.pending_shutdown_exit_thread_id { + self.submit_app_server_op(app_server, thread_id, Op::Shutdown) + .await + } else { + false + }; + if submitted { AppRunControl::Continue } else { self.pending_shutdown_exit_thread_id = None; @@ -3814,62 +3844,9 @@ impl App { } async fn handle_thread_created(&mut self, thread_id: ThreadId) -> Result<()> { - if self.thread_event_channels.contains_key(&thread_id) { - return Ok(()); - } - let thread = match self.server.get_thread(thread_id).await { - Ok(thread) => thread, - Err(err) => { - tracing::warn!("failed to attach listener for thread {thread_id}: {err}"); - return Ok(()); - } - }; - let config_snapshot = thread.config_snapshot().await; - self.upsert_agent_picker_thread( - thread_id, - config_snapshot.session_source.get_nickname(), - config_snapshot.session_source.get_agent_role(), - false, + tracing::debug!( + "handle_thread_created({thread_id}) should be unreachable after app-server migration" ); - let event = Event { - id: String::new(), - msg: EventMsg::SessionConfigured(SessionConfiguredEvent { - session_id: thread_id, - forked_from_id: None, - thread_name: None, - model: config_snapshot.model, - model_provider_id: config_snapshot.model_provider_id, - service_tier: config_snapshot.service_tier, - approval_policy: config_snapshot.approval_policy, - approvals_reviewer: config_snapshot.approvals_reviewer, - sandbox_policy: config_snapshot.sandbox_policy, - cwd: config_snapshot.cwd, - reasoning_effort: config_snapshot.reasoning_effort, - history_log_id: 0, - history_entry_count: 0, - initial_messages: None, - network_proxy: None, - rollout_path: thread.rollout_path(), - }), - }; - let channel = - ThreadEventChannel::new_with_session_configured(THREAD_EVENT_CHANNEL_CAPACITY, event); - let app_event_tx = self.app_event_tx.clone(); - self.thread_event_channels.insert(thread_id, channel); - let listener_handle = tokio::spawn(async move { - loop { - let event = match thread.next_event().await { - Ok(event) => event, - Err(err) => { - tracing::debug!("external thread {thread_id} listener stopped: {err}"); - break; - } - }; - app_event_tx.send(AppEvent::ThreadEvent { thread_id, event }); - } - }); - self.thread_event_listener_tasks - .insert(thread_id, listener_handle); Ok(()) } @@ -4182,10 +4159,13 @@ mod tests { use crate::history_cell::new_session_info; use crate::multi_agents::AgentPickerThreadEntry; use assert_matches::assert_matches; + use codex_arg0::Arg0DispatchPaths; use codex_core::CodexAuth; use codex_core::config::ConfigBuilder; use codex_core::config::ConfigOverrides; use codex_core::config::types::ModelAvailabilityNuxConfig; + use codex_core::config_loader::CloudRequirementsLoader; + use codex_core::config_loader::LoaderOverrides; use codex_otel::SessionTelemetry; use codex_protocol::ThreadId; use codex_protocol::config_types::CollaborationMode; @@ -4218,6 +4198,27 @@ mod tests { use tempfile::tempdir; use tokio::time; + async fn start_test_app_server(config: Config) -> InProcessAppServerClient { + InProcessAppServerClient::start(codex_app_server_client::InProcessClientStartArgs { + arg0_paths: Arg0DispatchPaths::default(), + config: Arc::new(config), + cli_overrides: Vec::new(), + loader_overrides: LoaderOverrides::default(), + cloud_requirements: CloudRequirementsLoader::default(), + feedback: codex_feedback::CodexFeedback::new(), + config_warnings: Vec::new(), + session_source: SessionSource::Cli, + enable_codex_api_key_env: false, + client_name: "codex-tui-test".to_string(), + client_version: "0.0.0-test".to_string(), + experimental_api: true, + opt_out_notification_methods: Vec::new(), + channel_capacity: codex_app_server_client::DEFAULT_IN_PROCESS_CHANNEL_CAPACITY, + }) + .await + .expect("in-process app server should start") + } + #[test] fn normalize_harness_overrides_resolves_relative_add_dirs() -> Result<()> { let temp_dir = tempdir()?; @@ -6327,12 +6328,19 @@ smart_approvals = true async fn make_test_app() -> App { let (chat_widget, app_event_tx, _rx, _op_rx) = make_chatwidget_manual_with_sender().await; let config = chat_widget.config_ref().clone(); - let server = Arc::new( + let thread_manager = Arc::new( codex_core::test_support::thread_manager_with_models_provider( CodexAuth::from_api_key("Test API Key"), config.model_provider.clone(), ), ); + let available_models = thread_manager + .get_models_manager() + .try_list_models() + .expect("test models should be available"); + let available_collaboration_modes = thread_manager + .get_models_manager() + .list_collaboration_modes(); let auth_manager = codex_core::test_support::auth_manager_from_auth( CodexAuth::from_api_key("Test API Key"), ); @@ -6341,11 +6349,12 @@ smart_approvals = true let session_telemetry = test_session_telemetry(&config, model.as_str()); App { - server, session_telemetry, app_event_tx, chat_widget, auth_manager, + available_models, + available_collaboration_modes, config, active_profile: None, cli_kv_overrides: Vec::new(), @@ -6376,6 +6385,15 @@ smart_approvals = true primary_thread_id: None, primary_session_configured: None, pending_primary_events: VecDeque::new(), + next_app_server_request_id: 1, + active_turn_ids: HashMap::new(), + pending_exec_approval_request_ids: HashMap::new(), + pending_patch_approval_request_ids: HashMap::new(), + pending_elicitation_request_ids: HashMap::new(), + pending_elicitation_request_queues: HashMap::new(), + pending_permissions_request_ids: HashMap::new(), + pending_user_input_request_ids: HashMap::new(), + pending_dynamic_tool_request_ids: HashMap::new(), } } @@ -6386,12 +6404,19 @@ smart_approvals = true ) { let (chat_widget, app_event_tx, rx, op_rx) = make_chatwidget_manual_with_sender().await; let config = chat_widget.config_ref().clone(); - let server = Arc::new( + let thread_manager = Arc::new( codex_core::test_support::thread_manager_with_models_provider( CodexAuth::from_api_key("Test API Key"), config.model_provider.clone(), ), ); + let available_models = thread_manager + .get_models_manager() + .try_list_models() + .expect("test models should be available"); + let available_collaboration_modes = thread_manager + .get_models_manager() + .list_collaboration_modes(); let auth_manager = codex_core::test_support::auth_manager_from_auth( CodexAuth::from_api_key("Test API Key"), ); @@ -6401,11 +6426,12 @@ smart_approvals = true ( App { - server, session_telemetry, app_event_tx, chat_widget, auth_manager, + available_models, + available_collaboration_modes, config, active_profile: None, cli_kv_overrides: Vec::new(), @@ -6436,6 +6462,15 @@ smart_approvals = true primary_thread_id: None, primary_session_configured: None, pending_primary_events: VecDeque::new(), + next_app_server_request_id: 1, + active_turn_ids: HashMap::new(), + pending_exec_approval_request_ids: HashMap::new(), + pending_patch_approval_request_ids: HashMap::new(), + pending_elicitation_request_ids: HashMap::new(), + pending_elicitation_request_queues: HashMap::new(), + pending_permissions_request_ids: HashMap::new(), + pending_user_input_request_ids: HashMap::new(), + pending_dynamic_tool_request_ids: HashMap::new(), }, rx, op_rx, @@ -7412,70 +7447,77 @@ smart_approvals = true #[tokio::test] async fn new_session_requests_shutdown_for_previous_conversation() { let (mut app, mut app_event_rx, mut op_rx) = make_test_app_with_channels().await; - - let thread_id = ThreadId::new(); - let event = SessionConfiguredEvent { - session_id: thread_id, - forked_from_id: None, - thread_name: None, - model: "gpt-test".to_string(), - model_provider_id: "test-provider".to_string(), - service_tier: None, - approval_policy: AskForApproval::Never, - approvals_reviewer: ApprovalsReviewer::User, - sandbox_policy: SandboxPolicy::new_read_only_policy(), - cwd: PathBuf::from("/home/user/project"), - reasoning_effort: None, - history_log_id: 0, - history_entry_count: 0, - initial_messages: None, - network_proxy: None, - rollout_path: Some(PathBuf::new()), - }; + let app_server = start_test_app_server(app.config.clone()).await; + let event = app + .thread_start_via_app_server(&app_server, &app.config.clone()) + .await + .expect("thread/start should succeed"); app.chat_widget.handle_codex_event(Event { id: String::new(), - msg: EventMsg::SessionConfigured(event), + msg: EventMsg::SessionConfigured(event.clone()), }); while app_event_rx.try_recv().is_ok() {} while op_rx.try_recv().is_ok() {} - app.shutdown_current_thread().await; + app.shutdown_current_thread(&app_server).await; - match op_rx.try_recv() { - Ok(Op::Shutdown) => {} - Ok(other) => panic!("expected Op::Shutdown, got {other:?}"), - Err(_) => panic!("expected shutdown op to be sent"), - } + assert!(app.suppress_shutdown_complete); + assert_eq!(app.chat_widget.thread_id(), Some(event.session_id)); + assert!(op_rx.try_recv().is_err()); + app_server + .shutdown() + .await + .expect("shutdown should complete"); } #[tokio::test] - async fn shutdown_first_exit_returns_immediate_exit_when_shutdown_submit_fails() { + async fn shutdown_first_exit_returns_immediate_exit_without_an_active_thread() { let mut app = make_test_app().await; - let thread_id = ThreadId::new(); - app.active_thread_id = Some(thread_id); + let app_server = start_test_app_server(app.config.clone()).await; - let control = app.handle_exit_mode(ExitMode::ShutdownFirst); + let control = app + .handle_exit_mode(&app_server, ExitMode::ShutdownFirst) + .await; assert_eq!(app.pending_shutdown_exit_thread_id, None); assert!(matches!( control, AppRunControl::Exit(ExitReason::UserRequested) )); + app_server + .shutdown() + .await + .expect("shutdown should complete"); } #[tokio::test] async fn shutdown_first_exit_waits_for_shutdown_when_submit_succeeds() { let (mut app, _app_event_rx, mut op_rx) = make_test_app_with_channels().await; - let thread_id = ThreadId::new(); - app.active_thread_id = Some(thread_id); + let app_server = start_test_app_server(app.config.clone()).await; + let event = app + .thread_start_via_app_server(&app_server, &app.config.clone()) + .await + .expect("thread/start should succeed"); + app.chat_widget.handle_codex_event(Event { + id: String::new(), + msg: EventMsg::SessionConfigured(event.clone()), + }); + app.active_thread_id = Some(event.session_id); + while op_rx.try_recv().is_ok() {} - let control = app.handle_exit_mode(ExitMode::ShutdownFirst); + let control = app + .handle_exit_mode(&app_server, ExitMode::ShutdownFirst) + .await; - assert_eq!(app.pending_shutdown_exit_thread_id, Some(thread_id)); + assert_eq!(app.pending_shutdown_exit_thread_id, Some(event.session_id)); assert!(matches!(control, AppRunControl::Continue)); - assert_eq!(op_rx.try_recv(), Ok(Op::Shutdown)); + assert!(op_rx.try_recv().is_err()); + app_server + .shutdown() + .await + .expect("shutdown should complete"); } #[tokio::test] diff --git a/codex-rs/tui/src/app/app_server_adapter.rs b/codex-rs/tui/src/app/app_server_adapter.rs index 48caeb1380..b3c1a5e58e 100644 --- a/codex-rs/tui/src/app/app_server_adapter.rs +++ b/codex-rs/tui/src/app/app_server_adapter.rs @@ -1,22 +1,783 @@ -/* -This module holds the temporary adapter layer between the TUI and the app -server during the hybrid migration period. - -For now, the TUI still owns its existing direct-core behavior, but startup -allocates a local in-process app server and drains its event stream. Keeping -the app-server-specific wiring here keeps that transitional logic out of the -main `app.rs` orchestration path. - -As more TUI flows move onto the app-server surface directly, this adapter -should shrink and eventually disappear. -*/ +use std::collections::HashMap; +use std::path::PathBuf; use super::App; use codex_app_server_client::InProcessAppServerClient; use codex_app_server_client::InProcessServerEvent; +use codex_app_server_protocol::ClientRequest; +use codex_app_server_protocol::CollaborationModeListParams; +use codex_app_server_protocol::CollaborationModeListResponse; +use codex_app_server_protocol::CommandExecutionRequestApprovalResponse; +use codex_app_server_protocol::FileChangeApprovalDecision; +use codex_app_server_protocol::FileChangeRequestApprovalResponse; +use codex_app_server_protocol::GetAccountParams; +use codex_app_server_protocol::GetAccountResponse; use codex_app_server_protocol::JSONRPCErrorError; +use codex_app_server_protocol::JSONRPCNotification; +use codex_app_server_protocol::McpServerElicitationAction; +use codex_app_server_protocol::McpServerElicitationRequestResponse; +use codex_app_server_protocol::McpServerRefreshResponse; +use codex_app_server_protocol::ModelListParams; +use codex_app_server_protocol::ModelListResponse; +use codex_app_server_protocol::PermissionsRequestApprovalResponse; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ReviewStartParams; +use codex_app_server_protocol::ReviewStartResponse; +use codex_app_server_protocol::ReviewTarget as ApiReviewTarget; +use codex_app_server_protocol::SandboxMode; +use codex_app_server_protocol::ServerNotification; +use codex_app_server_protocol::ServerRequest; +use codex_app_server_protocol::SkillsListParams; +use codex_app_server_protocol::SkillsListResponse; +use codex_app_server_protocol::ThreadBackgroundTerminalsCleanParams; +use codex_app_server_protocol::ThreadBackgroundTerminalsCleanResponse; +use codex_app_server_protocol::ThreadCompactStartParams; +use codex_app_server_protocol::ThreadCompactStartResponse; +use codex_app_server_protocol::ThreadForkParams; +use codex_app_server_protocol::ThreadForkResponse; +use codex_app_server_protocol::ThreadRealtimeAppendAudioParams; +use codex_app_server_protocol::ThreadRealtimeAppendAudioResponse; +use codex_app_server_protocol::ThreadRealtimeAppendTextParams; +use codex_app_server_protocol::ThreadRealtimeAppendTextResponse; +use codex_app_server_protocol::ThreadRealtimeStartParams; +use codex_app_server_protocol::ThreadRealtimeStartResponse; +use codex_app_server_protocol::ThreadRealtimeStopParams; +use codex_app_server_protocol::ThreadRealtimeStopResponse; +use codex_app_server_protocol::ThreadResumeParams; +use codex_app_server_protocol::ThreadResumeResponse; +use codex_app_server_protocol::ThreadRollbackParams; +use codex_app_server_protocol::ThreadRollbackResponse; +use codex_app_server_protocol::ThreadSetNameParams; +use codex_app_server_protocol::ThreadSetNameResponse; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::ThreadUnsubscribeParams; +use codex_app_server_protocol::ThreadUnsubscribeResponse; +use codex_app_server_protocol::ToolRequestUserInputResponse as ApiToolRequestUserInputResponse; +use codex_app_server_protocol::TurnInterruptParams; +use codex_app_server_protocol::TurnInterruptResponse; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::TurnStartResponse; +use codex_app_server_protocol::TurnSteerParams; +use codex_app_server_protocol::TurnSteerResponse; +use codex_core::config::Config; +use codex_protocol::ThreadId; +use codex_protocol::approvals::ElicitationAction; +use codex_protocol::config_types::CollaborationModeMask; +use codex_protocol::openai_models::ModelAvailabilityNux; +use codex_protocol::openai_models::ModelPreset; +use codex_protocol::openai_models::ModelUpgrade; +use codex_protocol::openai_models::ReasoningEffortPreset; +use codex_protocol::protocol::AskForApproval; +use codex_protocol::protocol::Event; +use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::ListSkillsResponseEvent; +use codex_protocol::protocol::Op; +use codex_protocol::protocol::ReviewTarget; +use codex_protocol::protocol::SandboxPolicy; +use codex_protocol::protocol::SessionConfiguredEvent; +use codex_protocol::request_permissions::PermissionGrantScope; +use serde::de::DeserializeOwned; +use serde_json::Value; impl App { + pub(super) async fn list_models_via_app_server( + app_server_client: &InProcessAppServerClient, + ) -> Result, String> { + let response: ModelListResponse = send_request_with_response( + app_server_client, + ClientRequest::ModelList { + request_id: RequestId::Integer(0), + params: ModelListParams::default(), + }, + "model/list", + ) + .await?; + Ok(response + .data + .into_iter() + .map(model_preset_from_api) + .collect()) + } + + pub(super) async fn list_collaboration_modes_via_app_server( + app_server_client: &InProcessAppServerClient, + ) -> Result, String> { + let response: CollaborationModeListResponse = send_request_with_response( + app_server_client, + ClientRequest::CollaborationModeList { + request_id: RequestId::Integer(0), + params: CollaborationModeListParams::default(), + }, + "collaborationMode/list", + ) + .await?; + Ok(response + .data + .into_iter() + .map(collaboration_mode_mask_from_api) + .collect()) + } + + pub(super) async fn read_account_via_app_server( + app_server_client: &InProcessAppServerClient, + ) -> Result, String> { + let response: GetAccountResponse = send_request_with_response( + app_server_client, + ClientRequest::GetAccount { + request_id: RequestId::Integer(0), + params: GetAccountParams { + refresh_token: false, + }, + }, + "account/read", + ) + .await?; + Ok(response.account) + } + + pub(super) fn next_app_server_request_id(&mut self) -> RequestId { + let request_id = self.next_app_server_request_id; + self.next_app_server_request_id = self.next_app_server_request_id.saturating_add(1); + RequestId::Integer(request_id) + } + + pub(super) async fn thread_start_via_app_server( + &mut self, + app_server_client: &InProcessAppServerClient, + config: &Config, + ) -> Result { + let response: ThreadStartResponse = send_request_with_response( + app_server_client, + ClientRequest::ThreadStart { + request_id: self.next_app_server_request_id(), + params: thread_start_params_from_config(config), + }, + "thread/start", + ) + .await?; + session_configured_from_thread_start_response(&response) + } + + pub(super) async fn thread_resume_via_app_server( + &mut self, + app_server_client: &InProcessAppServerClient, + config: &Config, + path: PathBuf, + ) -> Result { + let response: ThreadResumeResponse = send_request_with_response( + app_server_client, + ClientRequest::ThreadResume { + request_id: self.next_app_server_request_id(), + params: thread_resume_params_from_config(config, path), + }, + "thread/resume", + ) + .await?; + session_configured_from_thread_resume_response(&response) + } + + pub(super) async fn thread_fork_via_app_server( + &mut self, + app_server_client: &InProcessAppServerClient, + config: &Config, + path: PathBuf, + ) -> Result { + let response: ThreadForkResponse = send_request_with_response( + app_server_client, + ClientRequest::ThreadFork { + request_id: self.next_app_server_request_id(), + params: thread_fork_params_from_config(config, path), + }, + "thread/fork", + ) + .await?; + session_configured_from_thread_fork_response(&response) + } + + pub(super) async fn unsubscribe_thread_via_app_server( + &mut self, + app_server_client: &InProcessAppServerClient, + thread_id: ThreadId, + ) -> Result<(), String> { + let _: ThreadUnsubscribeResponse = send_request_with_response( + app_server_client, + ClientRequest::ThreadUnsubscribe { + request_id: self.next_app_server_request_id(), + params: ThreadUnsubscribeParams { + thread_id: thread_id.to_string(), + }, + }, + "thread/unsubscribe", + ) + .await?; + self.active_turn_ids.remove(&thread_id); + Ok(()) + } + + pub(super) async fn submit_app_server_op( + &mut self, + app_server_client: &InProcessAppServerClient, + thread_id: ThreadId, + op: Op, + ) -> bool { + if let Err(err) = self + .submit_app_server_op_inner(app_server_client, thread_id, op) + .await + { + self.chat_widget.add_error_message(err); + return false; + } + true + } + + async fn submit_app_server_op_inner( + &mut self, + app_server_client: &InProcessAppServerClient, + thread_id: ThreadId, + op: Op, + ) -> Result<(), String> { + match op { + Op::Interrupt => { + let Some(turn_id) = self.active_turn_ids.get(&thread_id).cloned() else { + return Err("No active turn to interrupt.".to_string()); + }; + let _: TurnInterruptResponse = send_request_with_response( + app_server_client, + ClientRequest::TurnInterrupt { + request_id: self.next_app_server_request_id(), + params: TurnInterruptParams { + thread_id: thread_id.to_string(), + turn_id, + }, + }, + "turn/interrupt", + ) + .await?; + } + Op::CleanBackgroundTerminals => { + let _: ThreadBackgroundTerminalsCleanResponse = send_request_with_response( + app_server_client, + ClientRequest::ThreadBackgroundTerminalsClean { + request_id: self.next_app_server_request_id(), + params: ThreadBackgroundTerminalsCleanParams { + thread_id: thread_id.to_string(), + }, + }, + "thread/backgroundTerminals/clean", + ) + .await?; + } + Op::RealtimeConversationStart(params) => { + let _: ThreadRealtimeStartResponse = send_request_with_response( + app_server_client, + ClientRequest::ThreadRealtimeStart { + request_id: self.next_app_server_request_id(), + params: ThreadRealtimeStartParams { + thread_id: thread_id.to_string(), + prompt: params.prompt, + session_id: params.session_id, + }, + }, + "thread/realtime/start", + ) + .await?; + } + Op::RealtimeConversationAudio(params) => { + let _: ThreadRealtimeAppendAudioResponse = send_request_with_response( + app_server_client, + ClientRequest::ThreadRealtimeAppendAudio { + request_id: self.next_app_server_request_id(), + params: ThreadRealtimeAppendAudioParams { + thread_id: thread_id.to_string(), + audio: params.frame.into(), + }, + }, + "thread/realtime/appendAudio", + ) + .await?; + } + Op::RealtimeConversationText(params) => { + let _: ThreadRealtimeAppendTextResponse = send_request_with_response( + app_server_client, + ClientRequest::ThreadRealtimeAppendText { + request_id: self.next_app_server_request_id(), + params: ThreadRealtimeAppendTextParams { + thread_id: thread_id.to_string(), + text: params.text, + }, + }, + "thread/realtime/appendText", + ) + .await?; + } + Op::RealtimeConversationClose => { + let _: ThreadRealtimeStopResponse = send_request_with_response( + app_server_client, + ClientRequest::ThreadRealtimeStop { + request_id: self.next_app_server_request_id(), + params: ThreadRealtimeStopParams { + thread_id: thread_id.to_string(), + }, + }, + "thread/realtime/stop", + ) + .await?; + } + Op::UserTurn { + items, + cwd, + approval_policy, + sandbox_policy, + model, + effort, + summary, + service_tier, + final_output_json_schema, + collaboration_mode, + personality, + } => { + if let Some(expected_turn_id) = self.active_turn_ids.get(&thread_id).cloned() { + let _: TurnSteerResponse = send_request_with_response( + app_server_client, + ClientRequest::TurnSteer { + request_id: self.next_app_server_request_id(), + params: TurnSteerParams { + thread_id: thread_id.to_string(), + input: items.into_iter().map(Into::into).collect(), + expected_turn_id, + }, + }, + "turn/steer", + ) + .await?; + } else { + let response: TurnStartResponse = send_request_with_response( + app_server_client, + ClientRequest::TurnStart { + request_id: self.next_app_server_request_id(), + params: TurnStartParams { + thread_id: thread_id.to_string(), + input: items.into_iter().map(Into::into).collect(), + cwd: Some(cwd), + approval_policy: Some(approval_policy.into()), + approvals_reviewer: Some(self.config.approvals_reviewer.into()), + sandbox_policy: Some(sandbox_policy.into()), + model: Some(model), + service_tier, + effort, + summary, + personality, + output_schema: final_output_json_schema, + collaboration_mode, + }, + }, + "turn/start", + ) + .await?; + self.active_turn_ids.insert(thread_id, response.turn.id); + } + } + Op::UserInput { + items, + final_output_json_schema, + } => { + if let Some(expected_turn_id) = self.active_turn_ids.get(&thread_id).cloned() { + let _: TurnSteerResponse = send_request_with_response( + app_server_client, + ClientRequest::TurnSteer { + request_id: self.next_app_server_request_id(), + params: TurnSteerParams { + thread_id: thread_id.to_string(), + input: items.into_iter().map(Into::into).collect(), + expected_turn_id, + }, + }, + "turn/steer", + ) + .await?; + } else { + let response: TurnStartResponse = send_request_with_response( + app_server_client, + ClientRequest::TurnStart { + request_id: self.next_app_server_request_id(), + params: TurnStartParams { + thread_id: thread_id.to_string(), + input: items.into_iter().map(Into::into).collect(), + cwd: Some(self.config.cwd.clone()), + approval_policy: Some( + self.config.permissions.approval_policy.value().into(), + ), + approvals_reviewer: Some(self.config.approvals_reviewer.into()), + sandbox_policy: Some( + self.config.permissions.sandbox_policy.get().clone().into(), + ), + model: self.config.model.clone(), + service_tier: self.config.service_tier.map(Some), + effort: self.config.model_reasoning_effort, + summary: None, + personality: self.config.personality, + output_schema: final_output_json_schema, + collaboration_mode: None, + }, + }, + "turn/start", + ) + .await?; + self.active_turn_ids.insert(thread_id, response.turn.id); + } + } + Op::ExecApproval { id, decision, .. } => { + let Some(request_id) = self.pending_exec_approval_request_ids.remove(&id) else { + return Err(format!( + "Missing app-server approval request for exec approval {id}." + )); + }; + let response = CommandExecutionRequestApprovalResponse { + decision: decision.into(), + }; + resolve_server_request( + app_server_client, + request_id, + response, + "item/commandExecution/requestApproval", + ) + .await?; + } + Op::PatchApproval { id, decision } => { + let Some(request_id) = self.pending_patch_approval_request_ids.remove(&id) else { + return Err(format!( + "Missing app-server approval request for patch approval {id}." + )); + }; + let response = FileChangeRequestApprovalResponse { + decision: review_decision_to_file_change_decision(decision), + }; + resolve_server_request( + app_server_client, + request_id, + response, + "item/fileChange/requestApproval", + ) + .await?; + } + Op::ResolveElicitation { + server_name, + request_id, + decision, + content, + meta, + } => { + let key = (server_name.clone(), request_id.to_string()); + let Some(server_request_id) = self.pending_elicitation_request_ids.remove(&key) + else { + return Err(format!( + "Missing app-server request for MCP elicitation {server_name}/{request_id}." + )); + }; + let response = McpServerElicitationRequestResponse { + action: elicitation_action_to_api(decision), + content, + meta, + }; + resolve_server_request( + app_server_client, + server_request_id, + response, + "mcpServer/elicitation/request", + ) + .await?; + } + Op::UserInputAnswer { id, response } => { + let Some(request_id) = self.pending_user_input_request_ids.remove(&id) else { + return Err(format!( + "Missing app-server request for user input turn {id}." + )); + }; + let response: ApiToolRequestUserInputResponse = + serde_json::from_value(serde_json::to_value(response).map_err(|err| { + format!("failed to encode request_user_input response: {err}") + })?) + .map_err(|err| { + format!("failed to convert request_user_input response: {err}") + })?; + resolve_server_request( + app_server_client, + request_id, + response, + "item/tool/requestUserInput", + ) + .await?; + } + Op::RequestPermissionsResponse { id, response } => { + let Some(request_id) = self.pending_permissions_request_ids.remove(&id) else { + return Err(format!( + "Missing app-server request for permissions approval {id}." + )); + }; + let response = PermissionsRequestApprovalResponse { + permissions: granted_permission_profile_from_request(response.permissions), + scope: permission_grant_scope_to_api(response.scope), + }; + resolve_server_request( + app_server_client, + request_id, + response, + "item/permissions/requestApproval", + ) + .await?; + } + Op::DynamicToolResponse { id, response } => { + let Some(request_id) = self.pending_dynamic_tool_request_ids.remove(&id) else { + return Err(format!( + "Missing app-server request for dynamic tool call {id}." + )); + }; + resolve_server_request(app_server_client, request_id, response, "item/tool/call") + .await?; + } + Op::AddToHistory { text } => { + let _ = (thread_id, text); + // TODO(app-server): expose message-history append/lookup APIs and migrate + // `AddToHistory`/`GetHistoryEntryRequest` together. + } + Op::ListSkills { cwds, force_reload } => { + let response: SkillsListResponse = send_request_with_response( + app_server_client, + ClientRequest::SkillsList { + request_id: self.next_app_server_request_id(), + params: SkillsListParams { + cwds, + force_reload, + per_cwd_extra_user_roots: None, + }, + }, + "skills/list", + ) + .await?; + let skills = serde_json::from_value( + serde_json::to_value(response.data) + .map_err(|err| format!("failed to encode skills/list response: {err}"))?, + ) + .map_err(|err| format!("failed to convert skills/list response: {err}"))?; + self.enqueue_primary_event(Event { + id: String::new(), + msg: EventMsg::ListSkillsResponse(ListSkillsResponseEvent { skills }), + }) + .await + .map_err(|err| err.to_string())?; + } + Op::RefreshMcpServers { config } => { + let _: McpServerRefreshResponse = send_request_with_response( + app_server_client, + ClientRequest::McpServerRefresh { + request_id: self.next_app_server_request_id(), + params: None, + }, + "config/mcpServer/reload", + ) + .await?; + let _ = config; + } + Op::ReloadUserConfig | Op::OverrideTurnContext { .. } => { + // TODO(app-server): add a thread-scoped override/context refresh API so the TUI + // does not need to treat these as local-only state updates. + } + Op::Compact => { + let _: ThreadCompactStartResponse = send_request_with_response( + app_server_client, + ClientRequest::ThreadCompactStart { + request_id: self.next_app_server_request_id(), + params: ThreadCompactStartParams { + thread_id: thread_id.to_string(), + }, + }, + "thread/compact/start", + ) + .await?; + } + Op::SetThreadName { name } => { + let _: ThreadSetNameResponse = send_request_with_response( + app_server_client, + ClientRequest::ThreadSetName { + request_id: self.next_app_server_request_id(), + params: ThreadSetNameParams { + thread_id: thread_id.to_string(), + name, + }, + }, + "thread/name/set", + ) + .await?; + } + Op::ThreadRollback { num_turns } => { + let _: ThreadRollbackResponse = send_request_with_response( + app_server_client, + ClientRequest::ThreadRollback { + request_id: self.next_app_server_request_id(), + params: ThreadRollbackParams { + thread_id: thread_id.to_string(), + num_turns, + }, + }, + "thread/rollback", + ) + .await?; + } + Op::Review { review_request } => { + let response: ReviewStartResponse = send_request_with_response( + app_server_client, + ClientRequest::ReviewStart { + request_id: self.next_app_server_request_id(), + params: ReviewStartParams { + thread_id: thread_id.to_string(), + target: review_target_to_api(review_request.target), + delivery: None, + }, + }, + "review/start", + ) + .await?; + self.active_turn_ids.insert(thread_id, response.turn.id); + } + Op::Shutdown => { + self.unsubscribe_thread_via_app_server(app_server_client, thread_id) + .await?; + } + Op::ListCustomPrompts + | Op::Undo + | Op::DropMemories + | Op::UpdateMemories + | Op::RunUserShellCommand { .. } + | Op::GetHistoryEntryRequest { .. } + | Op::ListMcpTools => { + // TODO(app-server): migrate these legacy-only TUI features once app-server grows + // equivalent APIs. For now, keep the UI explicit instead of silently dropping them. + return Err(format!( + "This TUI feature is not yet available over app-server: {}", + legacy_op_name(&op) + )); + } + Op::ListRemoteSkills { .. } | Op::DownloadRemoteSkill { .. } | Op::ListModels => { + return Err(format!( + "This TUI feature is not yet routed through app-server: {}", + legacy_op_name(&op) + )); + } + _ => { + return Err(format!( + "This TUI operation is not yet supported over app-server: {}", + legacy_op_name(&op) + )); + } + } + + Ok(()) + } + + fn note_server_request(&mut self, request: &ServerRequest) { + match request { + ServerRequest::CommandExecutionRequestApproval { request_id, params } => { + let approval_id = params + .approval_id + .clone() + .unwrap_or_else(|| params.item_id.clone()); + self.pending_exec_approval_request_ids + .insert(approval_id, request_id.clone()); + } + ServerRequest::FileChangeRequestApproval { request_id, params } => { + self.pending_patch_approval_request_ids + .insert(params.item_id.clone(), request_id.clone()); + } + ServerRequest::McpServerElicitationRequest { request_id, params } => { + self.pending_elicitation_request_queues + .entry(params.server_name.clone()) + .or_default() + .push_back(request_id.clone()); + } + ServerRequest::PermissionsRequestApproval { request_id, params } => { + self.pending_permissions_request_ids + .insert(params.item_id.clone(), request_id.clone()); + } + ServerRequest::ToolRequestUserInput { request_id, params } => { + self.pending_user_input_request_ids + .insert(params.turn_id.clone(), request_id.clone()); + } + ServerRequest::DynamicToolCall { request_id, params } => { + self.pending_dynamic_tool_request_ids + .insert(params.call_id.clone(), request_id.clone()); + } + ServerRequest::ApplyPatchApproval { .. } + | ServerRequest::ExecCommandApproval { .. } => { + // These legacy server requests are not expected on the turn/start path that the + // TUI now uses. Keep them explicit so regressions are obvious during cleanup. + } + ServerRequest::ChatgptAuthTokensRefresh { .. } => {} + } + } + + async fn handle_server_notification( + &mut self, + _app_server_client: &InProcessAppServerClient, + notification: ServerNotification, + ) { + match notification { + ServerNotification::TurnStarted(payload) => { + if let Ok(thread_id) = ThreadId::from_string(&payload.thread_id) { + self.active_turn_ids.insert(thread_id, payload.turn.id); + } + } + ServerNotification::TurnCompleted(payload) => { + if let Ok(thread_id) = ThreadId::from_string(&payload.thread_id) + && self.active_turn_ids.get(&thread_id) == Some(&payload.turn.id) + { + self.active_turn_ids.remove(&thread_id); + } + } + ServerNotification::ThreadStarted(payload) => { + if let Ok(thread_id) = ThreadId::from_string(&payload.thread.id) { + self.upsert_agent_picker_thread( + thread_id, + payload.thread.agent_nickname, + payload.thread.agent_role, + false, + ); + } + } + ServerNotification::ThreadClosed(payload) => { + if let Ok(thread_id) = ThreadId::from_string(&payload.thread_id) { + self.mark_agent_picker_thread_closed(thread_id); + self.active_turn_ids.remove(&thread_id); + } + } + ServerNotification::ThreadNameUpdated(payload) => { + if let Ok(thread_id) = ThreadId::from_string(&payload.thread_id) + && let Some(channel) = self.thread_event_channels.get(&thread_id) + { + let mut store = channel.store.lock().await; + if let Some(event) = store.session_configured.as_mut() + && let EventMsg::SessionConfigured(session) = &mut event.msg + { + session.thread_name = payload.thread_name.clone(); + } + } + } + _ => {} + } + } + + fn maybe_correlate_legacy_server_request(&mut self, event: &Event) { + if let EventMsg::ElicitationRequest(ev) = &event.msg + && let Some(queue) = self + .pending_elicitation_request_queues + .get_mut(&ev.server_name) + && let Some(request_id) = queue.pop_front() + { + self.pending_elicitation_request_ids + .insert((ev.server_name.clone(), ev.id.to_string()), request_id); + } + } + pub(super) async fn handle_app_server_event( &mut self, app_server_client: &InProcessAppServerClient, @@ -28,22 +789,70 @@ impl App { skipped, "app-server event consumer lagged; dropping ignored events" ); + self.chat_widget + .add_error_message(lagged_event_warning_message(skipped)); + } + InProcessServerEvent::ServerNotification(notification) => { + self.handle_server_notification(app_server_client, notification) + .await; + } + InProcessServerEvent::LegacyNotification(notification) => { + match decode_legacy_notification(notification) { + Ok(decoded) => { + self.maybe_correlate_legacy_server_request(&decoded.event); + match decoded.conversation_id { + Some(conversation_id) => { + match ThreadId::from_string(&conversation_id) { + Ok(thread_id) => { + if self.primary_thread_id.is_none() + || self.primary_thread_id == Some(thread_id) + { + if let Err(err) = + self.enqueue_primary_event(decoded.event).await + { + tracing::warn!("{err}"); + } + } else if let Err(err) = self + .handle_routed_thread_event(thread_id, decoded.event) + .await + { + tracing::warn!("{err}"); + } + } + Err(err) => { + tracing::warn!( + conversation_id, + error = %err, + "failed to parse app-server conversation id" + ); + } + } + } + None => { + if let Err(err) = self.enqueue_primary_event(decoded.event).await { + tracing::warn!("{err}"); + } + } + } + } + Err(err) => { + tracing::warn!("{err}"); + } + } } - InProcessServerEvent::ServerNotification(_) => {} - InProcessServerEvent::LegacyNotification(_) => {} InProcessServerEvent::ServerRequest(request) => { - let request_id = request.id().clone(); - tracing::warn!( - ?request_id, - "rejecting app-server request while TUI still uses direct core APIs" - ); - if let Err(err) = self - .reject_app_server_request( - app_server_client, - request_id, - "TUI client does not yet handle this app-server server request".to_string(), - ) - .await + self.note_server_request(&request); + if let ServerRequest::ChatgptAuthTokensRefresh { + request_id, + params: _, + } = request + && let Err(err) = self + .reject_app_server_request( + app_server_client, + request_id, + "TUI does not yet handle auth refresh server requests".to_string(), + ) + .await { tracing::warn!("{err}"); } @@ -54,9 +863,9 @@ impl App { async fn reject_app_server_request( &self, app_server_client: &InProcessAppServerClient, - request_id: codex_app_server_protocol::RequestId, + request_id: RequestId, reason: String, - ) -> std::result::Result<(), String> { + ) -> Result<(), String> { app_server_client .reject_server_request( request_id, @@ -70,3 +879,405 @@ impl App { .map_err(|err| format!("failed to reject app-server request: {err}")) } } + +async fn send_request_with_response( + client: &InProcessAppServerClient, + request: ClientRequest, + method: &str, +) -> Result +where + T: DeserializeOwned, +{ + client.request_typed(request).await.map_err(|err| { + if method.is_empty() { + err.to_string() + } else { + format!("{method}: {err}") + } + }) +} + +fn thread_start_params_from_config(config: &Config) -> ThreadStartParams { + ThreadStartParams { + model: config.model.clone(), + model_provider: Some(config.model_provider_id.clone()), + service_tier: config.service_tier.map(Some), + cwd: Some(config.cwd.to_string_lossy().to_string()), + approval_policy: Some(config.permissions.approval_policy.value().into()), + approvals_reviewer: Some(config.approvals_reviewer.into()), + sandbox: sandbox_mode_from_policy(config.permissions.sandbox_policy.get()), + config: config_request_overrides_from_config(config), + base_instructions: None, + developer_instructions: None, + personality: config.personality, + ephemeral: Some(config.ephemeral), + dynamic_tools: None, + mock_experimental_field: None, + experimental_raw_events: false, + persist_extended_history: true, + service_name: None, + } +} + +fn thread_resume_params_from_config(config: &Config, path: PathBuf) -> ThreadResumeParams { + ThreadResumeParams { + thread_id: "resume".to_string(), + history: None, + path: Some(path), + model: config.model.clone(), + model_provider: Some(config.model_provider_id.clone()), + service_tier: config.service_tier.map(Some), + cwd: Some(config.cwd.to_string_lossy().to_string()), + approval_policy: Some(config.permissions.approval_policy.value().into()), + approvals_reviewer: Some(config.approvals_reviewer.into()), + sandbox: sandbox_mode_from_policy(config.permissions.sandbox_policy.get()), + config: config_request_overrides_from_config(config), + base_instructions: None, + developer_instructions: None, + personality: config.personality, + persist_extended_history: true, + } +} + +fn thread_fork_params_from_config(config: &Config, path: PathBuf) -> ThreadForkParams { + ThreadForkParams { + thread_id: "fork".to_string(), + path: Some(path), + model: config.model.clone(), + model_provider: Some(config.model_provider_id.clone()), + service_tier: config.service_tier.map(Some), + cwd: Some(config.cwd.to_string_lossy().to_string()), + approval_policy: Some(config.permissions.approval_policy.value().into()), + approvals_reviewer: Some(config.approvals_reviewer.into()), + sandbox: sandbox_mode_from_policy(config.permissions.sandbox_policy.get()), + config: config_request_overrides_from_config(config), + base_instructions: None, + developer_instructions: None, + ephemeral: false, + persist_extended_history: true, + } +} + +fn config_request_overrides_from_config(config: &Config) -> Option> { + config + .active_profile + .as_ref() + .map(|profile| HashMap::from([("profile".to_string(), Value::String(profile.clone()))])) +} + +fn sandbox_mode_from_policy(sandbox_policy: &SandboxPolicy) -> Option { + match sandbox_policy { + SandboxPolicy::DangerFullAccess => Some(SandboxMode::DangerFullAccess), + SandboxPolicy::ReadOnly { .. } => Some(SandboxMode::ReadOnly), + SandboxPolicy::WorkspaceWrite { .. } => Some(SandboxMode::WorkspaceWrite), + SandboxPolicy::ExternalSandbox { .. } => None, + } +} + +fn session_configured_from_thread_start_response( + response: &ThreadStartResponse, +) -> Result { + session_configured_from_thread_response( + &response.thread.id, + response.thread.name.clone(), + response.thread.path.clone(), + response.model.clone(), + response.model_provider.clone(), + response.service_tier, + response.approval_policy.to_core(), + response.approvals_reviewer.to_core(), + response.sandbox.to_core(), + response.cwd.clone(), + response.reasoning_effort, + ) +} + +fn session_configured_from_thread_resume_response( + response: &ThreadResumeResponse, +) -> Result { + session_configured_from_thread_response( + &response.thread.id, + response.thread.name.clone(), + response.thread.path.clone(), + response.model.clone(), + response.model_provider.clone(), + response.service_tier, + response.approval_policy.to_core(), + response.approvals_reviewer.to_core(), + response.sandbox.to_core(), + response.cwd.clone(), + response.reasoning_effort, + ) +} + +fn session_configured_from_thread_fork_response( + response: &ThreadForkResponse, +) -> Result { + session_configured_from_thread_response( + &response.thread.id, + response.thread.name.clone(), + response.thread.path.clone(), + response.model.clone(), + response.model_provider.clone(), + response.service_tier, + response.approval_policy.to_core(), + response.approvals_reviewer.to_core(), + response.sandbox.to_core(), + response.cwd.clone(), + response.reasoning_effort, + ) +} + +#[expect( + clippy::too_many_arguments, + reason = "session mapping keeps explicit fields" +)] +fn session_configured_from_thread_response( + thread_id: &str, + thread_name: Option, + rollout_path: Option, + model: String, + model_provider_id: String, + service_tier: Option, + approval_policy: AskForApproval, + approvals_reviewer: codex_core::config::types::ApprovalsReviewer, + sandbox_policy: SandboxPolicy, + cwd: PathBuf, + reasoning_effort: Option, +) -> Result { + let session_id = ThreadId::from_string(thread_id) + .map_err(|err| format!("thread id `{thread_id}` is invalid: {err}"))?; + + Ok(SessionConfiguredEvent { + session_id, + forked_from_id: None, + thread_name, + model, + model_provider_id, + service_tier, + approval_policy, + approvals_reviewer, + sandbox_policy, + cwd, + reasoning_effort, + history_log_id: 0, + history_entry_count: 0, + initial_messages: None, + network_proxy: None, + rollout_path, + }) +} + +fn review_target_to_api(target: ReviewTarget) -> ApiReviewTarget { + match target { + ReviewTarget::UncommittedChanges => ApiReviewTarget::UncommittedChanges, + ReviewTarget::BaseBranch { branch } => ApiReviewTarget::BaseBranch { branch }, + ReviewTarget::Commit { sha, title } => ApiReviewTarget::Commit { sha, title }, + ReviewTarget::Custom { instructions } => ApiReviewTarget::Custom { instructions }, + } +} + +fn review_decision_to_file_change_decision( + decision: codex_protocol::protocol::ReviewDecision, +) -> FileChangeApprovalDecision { + match decision { + codex_protocol::protocol::ReviewDecision::Approved => FileChangeApprovalDecision::Accept, + codex_protocol::protocol::ReviewDecision::ApprovedForSession => { + FileChangeApprovalDecision::AcceptForSession + } + codex_protocol::protocol::ReviewDecision::Abort => FileChangeApprovalDecision::Cancel, + codex_protocol::protocol::ReviewDecision::Denied => FileChangeApprovalDecision::Decline, + codex_protocol::protocol::ReviewDecision::ApprovedExecpolicyAmendment { .. } + | codex_protocol::protocol::ReviewDecision::NetworkPolicyAmendment { .. } => { + FileChangeApprovalDecision::Accept + } + } +} + +fn permission_grant_scope_to_api( + scope: PermissionGrantScope, +) -> codex_app_server_protocol::PermissionGrantScope { + match scope { + PermissionGrantScope::Turn => codex_app_server_protocol::PermissionGrantScope::Turn, + PermissionGrantScope::Session => codex_app_server_protocol::PermissionGrantScope::Session, + } +} + +fn granted_permission_profile_from_request( + permissions: codex_protocol::request_permissions::RequestPermissionProfile, +) -> codex_app_server_protocol::GrantedPermissionProfile { + codex_app_server_protocol::GrantedPermissionProfile { + network: permissions.network.map(Into::into), + file_system: permissions.file_system.map(Into::into), + macos: None, + } +} + +fn elicitation_action_to_api(action: ElicitationAction) -> McpServerElicitationAction { + match action { + ElicitationAction::Accept => McpServerElicitationAction::Accept, + ElicitationAction::Decline => McpServerElicitationAction::Decline, + ElicitationAction::Cancel => McpServerElicitationAction::Cancel, + } +} + +fn normalize_legacy_notification_method(method: &str) -> &str { + method.strip_prefix("codex/event/").unwrap_or(method) +} + +fn lagged_event_warning_message(skipped: usize) -> String { + format!("in-process app-server event stream lagged; dropped {skipped} events") +} + +fn model_preset_from_api(model: codex_app_server_protocol::Model) -> ModelPreset { + ModelPreset { + id: model.id, + model: model.model.clone(), + display_name: model.display_name, + description: model.description, + default_reasoning_effort: model.default_reasoning_effort, + supported_reasoning_efforts: model + .supported_reasoning_efforts + .into_iter() + .map(|option| ReasoningEffortPreset { + effort: option.reasoning_effort, + description: option.description, + }) + .collect(), + supports_personality: model.supports_personality, + is_default: model.is_default, + upgrade: model.upgrade_info.map(|upgrade| ModelUpgrade { + id: upgrade.model, + reasoning_effort_mapping: None, + migration_config_key: model.model.clone(), + model_link: upgrade.model_link, + upgrade_copy: upgrade.upgrade_copy, + migration_markdown: upgrade.migration_markdown, + }), + show_in_picker: !model.hidden, + availability_nux: model.availability_nux.map(|nux| ModelAvailabilityNux { + message: nux.message, + }), + supported_in_api: true, + input_modalities: model.input_modalities, + } +} + +fn collaboration_mode_mask_from_api( + mask: codex_app_server_protocol::CollaborationModeMask, +) -> CollaborationModeMask { + CollaborationModeMask { + name: mask.name, + mode: mask.mode, + model: mask.model, + reasoning_effort: mask.reasoning_effort, + developer_instructions: None, + } +} + +struct DecodedLegacyNotification { + conversation_id: Option, + event: Event, +} + +fn decode_legacy_notification( + notification: JSONRPCNotification, +) -> Result { + let value = notification + .params + .unwrap_or_else(|| serde_json::Value::Object(serde_json::Map::new())); + let method = notification.method; + let normalized_method = normalize_legacy_notification_method(&method).to_string(); + let serde_json::Value::Object(mut object) = value else { + return Err(format!( + "legacy notification `{method}` params were not an object" + )); + }; + let conversation_id = object + .get("conversationId") + .and_then(serde_json::Value::as_str) + .map(str::to_owned); + let mut event_payload = if let Some(serde_json::Value::Object(msg_payload)) = object.get("msg") + { + serde_json::Value::Object(msg_payload.clone()) + } else { + object.remove("conversationId"); + serde_json::Value::Object(object) + }; + let serde_json::Value::Object(ref mut object) = event_payload else { + return Err(format!( + "legacy notification `{method}` event payload was not an object" + )); + }; + object.insert( + "type".to_string(), + serde_json::Value::String(normalized_method), + ); + + let msg: EventMsg = serde_json::from_value(event_payload) + .map_err(|err| format!("failed to decode event: {err}"))?; + Ok(DecodedLegacyNotification { + conversation_id, + event: Event { + id: String::new(), + msg, + }, + }) +} + +async fn resolve_server_request( + client: &InProcessAppServerClient, + request_id: RequestId, + value: T, + method: &str, +) -> Result<(), String> +where + T: serde::Serialize, +{ + let value = serde_json::to_value(value) + .map_err(|err| format!("failed to encode `{method}` server request response: {err}"))?; + client + .resolve_server_request(request_id, value) + .await + .map_err(|err| format!("failed to resolve `{method}` server request: {err}")) +} + +fn legacy_op_name(op: &Op) -> &'static str { + match op { + Op::Interrupt => "interrupt", + Op::CleanBackgroundTerminals => "clean_background_terminals", + Op::RealtimeConversationStart(_) => "realtime_conversation_start", + Op::RealtimeConversationAudio(_) => "realtime_conversation_audio", + Op::RealtimeConversationText(_) => "realtime_conversation_text", + Op::RealtimeConversationClose => "realtime_conversation_close", + Op::UserInput { .. } => "user_input", + Op::UserTurn { .. } => "user_turn", + Op::OverrideTurnContext { .. } => "override_turn_context", + Op::ExecApproval { .. } => "exec_approval", + Op::PatchApproval { .. } => "patch_approval", + Op::ResolveElicitation { .. } => "resolve_elicitation", + Op::UserInputAnswer { .. } => "user_input_answer", + Op::RequestPermissionsResponse { .. } => "request_permissions_response", + Op::DynamicToolResponse { .. } => "dynamic_tool_response", + Op::AddToHistory { .. } => "add_to_history", + Op::GetHistoryEntryRequest { .. } => "get_history_entry_request", + Op::ListMcpTools => "list_mcp_tools", + Op::RefreshMcpServers { .. } => "refresh_mcp_servers", + Op::ReloadUserConfig => "reload_user_config", + Op::ListCustomPrompts => "list_custom_prompts", + Op::ListSkills { .. } => "list_skills", + Op::ListRemoteSkills { .. } => "list_remote_skills", + Op::DownloadRemoteSkill { .. } => "download_remote_skill", + Op::Compact => "compact", + Op::DropMemories => "drop_memories", + Op::UpdateMemories => "update_memories", + Op::SetThreadName { .. } => "set_thread_name", + Op::Undo => "undo", + Op::ThreadRollback { .. } => "thread_rollback", + Op::Review { .. } => "review", + Op::Shutdown => "shutdown", + Op::RunUserShellCommand { .. } => "run_user_shell_command", + Op::ListModels => "list_models", + _ => "unknown", + } +} diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index 4c8e76874c..2b0d52837e 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -67,7 +67,6 @@ use codex_core::git_info::current_branch_name; use codex_core::git_info::get_git_repo_root; use codex_core::git_info::local_git_branches; use codex_core::mcp::McpManager; -use codex_core::models_manager::manager::ModelsManager; use codex_core::plugins::PluginsManager; use codex_core::project_doc::DEFAULT_PROJECT_DOC_FILENAME; use codex_core::skills::model::SkillMetadata; @@ -274,7 +273,7 @@ use self::interrupts::InterruptManager; mod agent; use self::agent::spawn_agent; use self::agent::spawn_agent_from_existing; -pub(crate) use self::agent::spawn_op_forwarder; +pub(crate) use self::agent::spawn_app_event_forwarder; mod session_header; use self::session_header::SessionHeader; mod skills; @@ -476,7 +475,8 @@ pub(crate) struct ChatWidgetInit { pub(crate) initial_user_message: Option, pub(crate) enhanced_keys_supported: bool, pub(crate) auth_manager: Arc, - pub(crate) models_manager: Arc, + pub(crate) available_models: Vec, + pub(crate) available_collaboration_modes: Vec, pub(crate) feedback: codex_feedback::CodexFeedback, pub(crate) is_first_run: bool, pub(crate) feedback_audience: FeedbackAudience, @@ -654,7 +654,8 @@ pub(crate) struct ChatWidget { /// The currently active collaboration mask, if any. active_collaboration_mask: Option, auth_manager: Arc, - models_manager: Arc, + available_models: Vec, + available_collaboration_modes: Vec, session_telemetry: SessionTelemetry, session_header: SessionHeader, initial_user_message: Option, @@ -1813,7 +1814,8 @@ impl ChatWidget { } fn open_plan_implementation_prompt(&mut self) { - let default_mask = collaboration_modes::default_mode_mask(self.models_manager.as_ref()); + let default_mask = + collaboration_modes::default_mode_mask(&self.available_collaboration_modes); let (implement_actions, implement_disabled_reason) = match default_mask { Some(mask) => { let user_text = PLAN_IMPLEMENTATION_CODING_MESSAGE.to_string(); @@ -3519,7 +3521,8 @@ impl ChatWidget { initial_user_message, enhanced_keys_supported, auth_manager, - models_manager, + available_models, + available_collaboration_modes, feedback, is_first_run, feedback_audience, @@ -3540,8 +3543,11 @@ impl ChatWidget { let model_for_header = model .clone() .unwrap_or_else(|| DEFAULT_MODEL_DISPLAY_NAME.to_string()); - let active_collaboration_mask = - Self::initial_collaboration_mask(&config, models_manager.as_ref(), model_override); + let active_collaboration_mask = Self::initial_collaboration_mask( + &config, + &available_collaboration_modes, + model_override, + ); let header_model = active_collaboration_mask .as_ref() .and_then(|mask| mask.model.clone()) @@ -3584,7 +3590,8 @@ impl ChatWidget { current_collaboration_mode, active_collaboration_mask, auth_manager, - models_manager, + available_models, + available_collaboration_modes, session_telemetry, session_header: SessionHeader::new(header_model), initial_user_message, @@ -3706,7 +3713,8 @@ impl ChatWidget { initial_user_message, enhanced_keys_supported, auth_manager, - models_manager, + available_models, + available_collaboration_modes, feedback, is_first_run, feedback_audience, @@ -3726,8 +3734,11 @@ impl ChatWidget { let model_for_header = model .clone() .unwrap_or_else(|| DEFAULT_MODEL_DISPLAY_NAME.to_string()); - let active_collaboration_mask = - Self::initial_collaboration_mask(&config, models_manager.as_ref(), model_override); + let active_collaboration_mask = Self::initial_collaboration_mask( + &config, + &available_collaboration_modes, + model_override, + ); let header_model = active_collaboration_mask .as_ref() .and_then(|mask| mask.model.clone()) @@ -3770,7 +3781,8 @@ impl ChatWidget { current_collaboration_mode, active_collaboration_mask, auth_manager, - models_manager, + available_models, + available_collaboration_modes, session_telemetry, session_header: SessionHeader::new(header_model), initial_user_message, @@ -3884,7 +3896,8 @@ impl ChatWidget { initial_user_message, enhanced_keys_supported, auth_manager, - models_manager, + available_models, + available_collaboration_modes, feedback, is_first_run: _, feedback_audience, @@ -3902,8 +3915,11 @@ impl ChatWidget { let header_model = model .clone() .unwrap_or_else(|| session_configured.model.clone()); - let active_collaboration_mask = - Self::initial_collaboration_mask(&config, models_manager.as_ref(), model_override); + let active_collaboration_mask = Self::initial_collaboration_mask( + &config, + &available_collaboration_modes, + model_override, + ); let header_model = active_collaboration_mask .as_ref() .and_then(|mask| mask.model.clone()) @@ -3948,7 +3964,8 @@ impl ChatWidget { current_collaboration_mode, active_collaboration_mask, auth_manager, - models_manager, + available_models, + available_collaboration_modes, session_telemetry, session_header: SessionHeader::new(header_model), initial_user_message, @@ -4393,7 +4410,9 @@ impl ChatWidget { ); return; } - if let Some(mask) = collaboration_modes::plan_mask(self.models_manager.as_ref()) { + if let Some(mask) = + collaboration_modes::plan_mask(&self.available_collaboration_modes) + { self.set_collaboration_mask(mask); } else { self.add_info_message("Plan mode unavailable right now.".to_string(), None); @@ -6106,8 +6125,7 @@ impl ChatWidget { } fn lower_cost_preset(&self) -> Option { - let models = self.models_manager.try_list_models().ok()?; - models + self.available_models .iter() .find(|preset| preset.show_in_picker && preset.model == NUDGE_MODEL_SLUG) .cloned() @@ -6225,16 +6243,11 @@ impl ChatWidget { return; } - let presets: Vec = match self.models_manager.try_list_models() { - Ok(models) => models, - Err(_) => { - self.add_info_message( - "Models are being updated; please try /model again in a moment.".to_string(), - None, - ); - return; - } - }; + let presets = self.available_models.clone(); + if presets.is_empty() { + self.add_info_message("No models are available right now.".to_string(), None); + return; + } self.open_model_popup_with_presets(presets); } @@ -6645,7 +6658,7 @@ impl ChatWidget { } pub(crate) fn open_collaboration_modes_popup(&mut self) { - let presets = collaboration_modes::presets_for_tui(self.models_manager.as_ref()); + let presets = collaboration_modes::presets_for_tui(&self.available_collaboration_modes); if presets.is_empty() { self.add_info_message( "No collaboration modes are available right now.".to_string(), @@ -6659,7 +6672,7 @@ impl ChatWidget { .as_ref() .and_then(|mask| mask.mode) .or_else(|| { - collaboration_modes::default_mask(self.models_manager.as_ref()) + collaboration_modes::default_mask(&self.available_collaboration_modes) .and_then(|mask| mask.mode) }); let items: Vec = presets @@ -6748,25 +6761,25 @@ impl ChatWidget { None => "the selected reasoning".to_string(), }; let plan_only_description = format!("Always use {reasoning_phrase} in Plan mode."); - let plan_reasoning_source = if let Some(plan_override) = - self.config.plan_mode_reasoning_effort - { - format!( - "user-chosen Plan override ({})", - Self::reasoning_effort_label(plan_override).to_lowercase() - ) - } else if let Some(plan_mask) = collaboration_modes::plan_mask(self.models_manager.as_ref()) - { - match plan_mask.reasoning_effort.flatten() { - Some(plan_effort) => format!( - "built-in Plan default ({})", - Self::reasoning_effort_label(plan_effort).to_lowercase() - ), - None => "built-in Plan default (no reasoning)".to_string(), - } - } else { - "built-in Plan default".to_string() - }; + let plan_reasoning_source = + if let Some(plan_override) = self.config.plan_mode_reasoning_effort { + format!( + "user-chosen Plan override ({})", + Self::reasoning_effort_label(plan_override).to_lowercase() + ) + } else if let Some(plan_mask) = + collaboration_modes::plan_mask(&self.available_collaboration_modes) + { + match plan_mask.reasoning_effort.flatten() { + Some(plan_effort) => format!( + "built-in Plan default ({})", + Self::reasoning_effort_label(plan_effort).to_lowercase() + ), + None => "built-in Plan default (no reasoning)".to_string(), + } + } else { + "built-in Plan default".to_string() + }; let all_modes_description = format!( "Set the global default reasoning level and the Plan mode override. This replaces the current {plan_reasoning_source}." ); @@ -7901,7 +7914,7 @@ impl ChatWidget { if let Some(effort) = effort { mask.reasoning_effort = Some(Some(effort)); } else if let Some(plan_mask) = - collaboration_modes::plan_mask(self.models_manager.as_ref()) + collaboration_modes::plan_mask(&self.available_collaboration_modes) { mask.reasoning_effort = plan_mask.reasoning_effort; } @@ -8042,15 +8055,10 @@ impl ChatWidget { fn current_model_supports_personality(&self) -> bool { let model = self.current_model(); - self.models_manager - .try_list_models() - .ok() - .and_then(|models| { - models - .into_iter() - .find(|preset| preset.model == model) - .map(|preset| preset.supports_personality) - }) + self.available_models + .iter() + .find(|preset| preset.model == model) + .map(|preset| preset.supports_personality) .unwrap_or(false) } @@ -8060,15 +8068,10 @@ impl ChatWidget { /// failures do not hard-block user input in the UI. fn current_model_supports_images(&self) -> bool { let model = self.current_model(); - self.models_manager - .try_list_models() - .ok() - .and_then(|models| { - models - .into_iter() - .find(|preset| preset.model == model) - .map(|preset| preset.input_modalities.contains(&InputModality::Image)) - }) + self.available_models + .iter() + .find(|preset| preset.model == model) + .map(|preset| preset.input_modalities.contains(&InputModality::Image)) .unwrap_or(true) } @@ -8108,10 +8111,10 @@ impl ChatWidget { fn initial_collaboration_mask( _config: &Config, - models_manager: &ModelsManager, + available_collaboration_modes: &[CollaborationModeMask], model_override: Option<&str>, ) -> Option { - let mut mask = collaboration_modes::default_mask(models_manager)?; + let mut mask = collaboration_modes::default_mask(available_collaboration_modes)?; if let Some(model_override) = model_override { mask.model = Some(model_override.to_string()); } @@ -8211,7 +8214,7 @@ impl ChatWidget { } if let Some(next_mask) = collaboration_modes::next_mask( - self.models_manager.as_ref(), + &self.available_collaboration_modes, self.active_collaboration_mask.as_ref(), ) { self.set_collaboration_mask(next_mask); diff --git a/codex-rs/tui/src/chatwidget/agent.rs b/codex-rs/tui/src/chatwidget/agent.rs index e14a9e3628..385bc41091 100644 --- a/codex-rs/tui/src/chatwidget/agent.rs +++ b/codex-rs/tui/src/chatwidget/agent.rs @@ -147,3 +147,19 @@ pub(crate) fn spawn_op_forwarder(thread: std::sync::Arc) -> Unbound codex_op_tx } + +/// Spawn an op forwarder that routes all widget submissions back onto the app event bus. +/// +/// The app layer owns app-server request translation during the migration, so the chat widget +/// should no longer need a direct thread handle for normal runtime operation. +pub(crate) fn spawn_app_event_forwarder(app_event_tx: AppEventSender) -> UnboundedSender { + let (codex_op_tx, mut codex_op_rx) = unbounded_channel::(); + + tokio::spawn(async move { + while let Some(op) = codex_op_rx.recv().await { + app_event_tx.send(AppEvent::CodexOp(op)); + } + }); + + codex_op_tx +} diff --git a/codex-rs/tui/src/chatwidget/tests.rs b/codex-rs/tui/src/chatwidget/tests.rs index 730aa99004..2093b92463 100644 --- a/codex-rs/tui/src/chatwidget/tests.rs +++ b/codex-rs/tui/src/chatwidget/tests.rs @@ -1310,7 +1310,7 @@ async fn interrupted_turn_restore_keeps_active_mode_for_resubmission() { chat.thread_id = Some(ThreadId::new()); chat.set_feature_enabled(Feature::CollaborationModes, true); - let plan_mask = collaboration_modes::plan_mask(chat.models_manager.as_ref()) + let plan_mask = collaboration_modes::plan_mask(&chat.available_collaboration_modes) .expect("expected plan collaboration mode"); let expected_mode = plan_mask .mode @@ -1761,7 +1761,13 @@ async fn helpers_are_available_and_do_not_panic() { initial_user_message: None, enhanced_keys_supported: false, auth_manager, - models_manager: thread_manager.get_models_manager(), + available_models: thread_manager + .get_models_manager() + .try_list_models() + .expect("test models should be available"), + available_collaboration_modes: thread_manager + .get_models_manager() + .list_collaboration_modes(), feedback: codex_feedback::CodexFeedback::new(), is_first_run: true, feedback_audience: FeedbackAudience::External, @@ -1791,6 +1797,17 @@ fn test_session_telemetry(config: &Config, model: &str) -> SessionTelemetry { ) } +fn available_chatwidget_catalog( + models_manager: &ModelsManager, +) -> (Vec, Vec) { + ( + models_manager + .try_list_models() + .expect("test models should be available"), + models_manager.list_collaboration_modes(), + ) +} + // --- Helpers for tests that need direct construction and event draining --- async fn make_chatwidget_manual( model_override: Option<&str>, @@ -1831,6 +1848,8 @@ async fn make_chatwidget_manual( None, CollaborationModesConfig::default(), )); + let (available_models, available_collaboration_modes) = + available_chatwidget_catalog(models_manager.as_ref()); let reasoning_effort = None; let base_mode = CollaborationMode { mode: ModeKind::Default, @@ -1841,7 +1860,8 @@ async fn make_chatwidget_manual( }, }; let current_collaboration_mode = base_mode; - let active_collaboration_mask = collaboration_modes::default_mask(models_manager.as_ref()); + let active_collaboration_mask = + collaboration_modes::default_mask(&available_collaboration_modes); let mut widget = ChatWidget { app_event_tx, codex_op_tx: op_tx, @@ -1852,7 +1872,8 @@ async fn make_chatwidget_manual( current_collaboration_mode, active_collaboration_mask, auth_manager, - models_manager, + available_models, + available_collaboration_modes, session_telemetry, session_header: SessionHeader::new(resolved_model.clone()), initial_user_message: None, @@ -1970,12 +1991,16 @@ pub(crate) fn set_chatgpt_auth(chat: &mut ChatWidget) { chat.auth_manager = codex_core::test_support::auth_manager_from_auth( CodexAuth::create_dummy_chatgpt_auth_for_testing(), ); - chat.models_manager = Arc::new(ModelsManager::new( + let models_manager = Arc::new(ModelsManager::new( chat.config.codex_home.clone(), chat.auth_manager.clone(), None, CollaborationModesConfig::default(), )); + let (available_models, available_collaboration_modes) = + available_chatwidget_catalog(models_manager.as_ref()); + chat.available_models = available_models; + chat.available_collaboration_modes = available_collaboration_modes; } #[tokio::test] @@ -2477,7 +2502,7 @@ async fn submit_user_message_with_mode_sets_coding_collaboration_mode() { chat.thread_id = Some(ThreadId::new()); chat.set_feature_enabled(Feature::CollaborationModes, true); - let default_mode = collaboration_modes::default_mode_mask(chat.models_manager.as_ref()) + let default_mode = collaboration_modes::default_mode_mask(&chat.available_collaboration_modes) .expect("expected default collaboration mode"); chat.submit_user_message_with_mode("Implement the plan.".to_string(), default_mode); @@ -2502,7 +2527,7 @@ async fn reasoning_selection_in_plan_mode_opens_scope_prompt_event() { let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(Some("gpt-5.1-codex-max")).await; chat.thread_id = Some(ThreadId::new()); chat.set_feature_enabled(Feature::CollaborationModes, true); - let plan_mask = collaboration_modes::plan_mask(chat.models_manager.as_ref()) + let plan_mask = collaboration_modes::plan_mask(&chat.available_collaboration_modes) .expect("expected plan collaboration mode"); chat.set_collaboration_mask(plan_mask); let _ = drain_insert_history(&mut rx); @@ -2529,7 +2554,7 @@ async fn reasoning_selection_in_plan_mode_without_effort_change_does_not_open_sc let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(Some("gpt-5.1-codex-max")).await; chat.thread_id = Some(ThreadId::new()); chat.set_feature_enabled(Feature::CollaborationModes, true); - let plan_mask = collaboration_modes::plan_mask(chat.models_manager.as_ref()) + let plan_mask = collaboration_modes::plan_mask(&chat.available_collaboration_modes) .expect("expected plan collaboration mode"); chat.set_collaboration_mask(plan_mask); let _ = drain_insert_history(&mut rx); @@ -2564,7 +2589,7 @@ async fn reasoning_selection_in_plan_mode_matching_plan_effort_but_different_glo let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(Some("gpt-5.1-codex-max")).await; chat.thread_id = Some(ThreadId::new()); chat.set_feature_enabled(Feature::CollaborationModes, true); - let plan_mask = collaboration_modes::plan_mask(chat.models_manager.as_ref()) + let plan_mask = collaboration_modes::plan_mask(&chat.available_collaboration_modes) .expect("expected plan collaboration mode"); chat.set_collaboration_mask(plan_mask); let _ = drain_insert_history(&mut rx); @@ -2597,7 +2622,7 @@ async fn plan_mode_reasoning_override_is_marked_current_in_reasoning_popup() { chat.set_reasoning_effort(Some(ReasoningEffortConfig::High)); chat.set_plan_mode_reasoning_effort(Some(ReasoningEffortConfig::Low)); - let plan_mask = collaboration_modes::plan_mask(chat.models_manager.as_ref()) + let plan_mask = collaboration_modes::plan_mask(&chat.available_collaboration_modes) .expect("expected plan collaboration mode"); chat.set_collaboration_mask(plan_mask); @@ -2617,7 +2642,7 @@ async fn reasoning_selection_in_plan_mode_model_switch_does_not_open_scope_promp let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(Some("gpt-5.1-codex-max")).await; chat.thread_id = Some(ThreadId::new()); chat.set_feature_enabled(Feature::CollaborationModes, true); - let plan_mask = collaboration_modes::plan_mask(chat.models_manager.as_ref()) + let plan_mask = collaboration_modes::plan_mask(&chat.available_collaboration_modes) .expect("expected plan collaboration mode"); chat.set_collaboration_mask(plan_mask); let _ = drain_insert_history(&mut rx); @@ -2883,12 +2908,12 @@ async fn submit_user_message_with_mode_errors_when_mode_changes_during_running_t chat.thread_id = Some(ThreadId::new()); chat.set_feature_enabled(Feature::CollaborationModes, true); let plan_mask = - collaboration_modes::mask_for_kind(chat.models_manager.as_ref(), ModeKind::Plan) + collaboration_modes::mask_for_kind(&chat.available_collaboration_modes, ModeKind::Plan) .expect("expected plan collaboration mask"); chat.set_collaboration_mask(plan_mask); chat.on_task_started(); - let default_mode = collaboration_modes::default_mask(chat.models_manager.as_ref()) + let default_mode = collaboration_modes::default_mask(&chat.available_collaboration_modes) .expect("expected default collaboration mode"); chat.submit_user_message_with_mode("Implement the plan.".to_string(), default_mode); @@ -2912,7 +2937,7 @@ async fn submit_user_message_with_mode_allows_same_mode_during_running_turn() { chat.thread_id = Some(ThreadId::new()); chat.set_feature_enabled(Feature::CollaborationModes, true); let plan_mask = - collaboration_modes::mask_for_kind(chat.models_manager.as_ref(), ModeKind::Plan) + collaboration_modes::mask_for_kind(&chat.available_collaboration_modes, ModeKind::Plan) .expect("expected plan collaboration mask"); chat.set_collaboration_mask(plan_mask.clone()); chat.on_task_started(); @@ -2943,11 +2968,11 @@ async fn submit_user_message_with_mode_submits_when_plan_stream_is_not_active() chat.thread_id = Some(ThreadId::new()); chat.set_feature_enabled(Feature::CollaborationModes, true); let plan_mask = - collaboration_modes::mask_for_kind(chat.models_manager.as_ref(), ModeKind::Plan) + collaboration_modes::mask_for_kind(&chat.available_collaboration_modes, ModeKind::Plan) .expect("expected plan collaboration mask"); chat.set_collaboration_mask(plan_mask); - let default_mode = collaboration_modes::default_mask(chat.models_manager.as_ref()) + let default_mode = collaboration_modes::default_mask(&chat.available_collaboration_modes) .expect("expected default collaboration mode"); let expected_mode = default_mode .mode @@ -2973,7 +2998,7 @@ async fn plan_implementation_popup_skips_replayed_turn_complete() { let (mut chat, _rx, _op_rx) = make_chatwidget_manual(Some("gpt-5")).await; chat.set_feature_enabled(Feature::CollaborationModes, true); let plan_mask = - collaboration_modes::mask_for_kind(chat.models_manager.as_ref(), ModeKind::Plan) + collaboration_modes::mask_for_kind(&chat.available_collaboration_modes, ModeKind::Plan) .expect("expected plan collaboration mask"); chat.set_collaboration_mask(plan_mask); @@ -2994,7 +3019,7 @@ async fn plan_implementation_popup_shows_once_when_replay_precedes_live_turn_com let (mut chat, _rx, _op_rx) = make_chatwidget_manual(Some("gpt-5")).await; chat.set_feature_enabled(Feature::CollaborationModes, true); let plan_mask = - collaboration_modes::mask_for_kind(chat.models_manager.as_ref(), ModeKind::Plan) + collaboration_modes::mask_for_kind(&chat.available_collaboration_modes, ModeKind::Plan) .expect("expected plan collaboration mask"); chat.set_collaboration_mask(plan_mask); @@ -3072,7 +3097,7 @@ async fn plan_implementation_popup_skips_when_messages_queued() { let (mut chat, _rx, _op_rx) = make_chatwidget_manual(Some("gpt-5")).await; chat.set_feature_enabled(Feature::CollaborationModes, true); let plan_mask = - collaboration_modes::mask_for_kind(chat.models_manager.as_ref(), ModeKind::Plan) + collaboration_modes::mask_for_kind(&chat.available_collaboration_modes, ModeKind::Plan) .expect("expected plan collaboration mask"); chat.set_collaboration_mask(plan_mask); chat.bottom_pane.set_task_running(true); @@ -3092,7 +3117,7 @@ async fn plan_implementation_popup_skips_without_proposed_plan() { let (mut chat, _rx, _op_rx) = make_chatwidget_manual(Some("gpt-5")).await; chat.set_feature_enabled(Feature::CollaborationModes, true); let plan_mask = - collaboration_modes::mask_for_kind(chat.models_manager.as_ref(), ModeKind::Plan) + collaboration_modes::mask_for_kind(&chat.available_collaboration_modes, ModeKind::Plan) .expect("expected plan collaboration mask"); chat.set_collaboration_mask(plan_mask); @@ -3118,7 +3143,7 @@ async fn plan_implementation_popup_shows_after_proposed_plan_output() { let (mut chat, _rx, _op_rx) = make_chatwidget_manual(Some("gpt-5")).await; chat.set_feature_enabled(Feature::CollaborationModes, true); let plan_mask = - collaboration_modes::mask_for_kind(chat.models_manager.as_ref(), ModeKind::Plan) + collaboration_modes::mask_for_kind(&chat.available_collaboration_modes, ModeKind::Plan) .expect("expected plan collaboration mask"); chat.set_collaboration_mask(plan_mask); @@ -3139,7 +3164,7 @@ async fn plan_implementation_popup_skips_when_steer_follows_proposed_plan() { let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(Some("gpt-5")).await; chat.set_feature_enabled(Feature::CollaborationModes, true); let plan_mask = - collaboration_modes::mask_for_kind(chat.models_manager.as_ref(), ModeKind::Plan) + collaboration_modes::mask_for_kind(&chat.available_collaboration_modes, ModeKind::Plan) .expect("expected plan collaboration mask"); chat.set_collaboration_mask(plan_mask); chat.thread_id = Some(ThreadId::new()); @@ -3181,7 +3206,7 @@ async fn plan_implementation_popup_shows_after_new_plan_follows_steer() { let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(Some("gpt-5")).await; chat.set_feature_enabled(Feature::CollaborationModes, true); let plan_mask = - collaboration_modes::mask_for_kind(chat.models_manager.as_ref(), ModeKind::Plan) + collaboration_modes::mask_for_kind(&chat.available_collaboration_modes, ModeKind::Plan) .expect("expected plan collaboration mask"); chat.set_collaboration_mask(plan_mask); chat.thread_id = Some(ThreadId::new()); @@ -3230,7 +3255,7 @@ async fn plan_implementation_popup_skips_when_rate_limit_prompt_pending() { ); chat.set_feature_enabled(Feature::CollaborationModes, true); let plan_mask = - collaboration_modes::mask_for_kind(chat.models_manager.as_ref(), ModeKind::Plan) + collaboration_modes::mask_for_kind(&chat.available_collaboration_modes, ModeKind::Plan) .expect("expected plan collaboration mask"); chat.set_collaboration_mask(plan_mask); @@ -3634,11 +3659,7 @@ fn active_blob(chat: &ChatWidget) -> String { } fn get_available_model(chat: &ChatWidget, model: &str) -> ModelPreset { - let models = chat - .models_manager - .try_list_models() - .expect("models lock available"); - models + chat.available_models .iter() .find(|&preset| preset.model == model) .cloned() @@ -3902,7 +3923,7 @@ async fn plan_completion_restores_status_indicator_after_streaming_plan_output() let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await; chat.set_feature_enabled(Feature::CollaborationModes, true); let plan_mask = - collaboration_modes::mask_for_kind(chat.models_manager.as_ref(), ModeKind::Plan) + collaboration_modes::mask_for_kind(&chat.available_collaboration_modes, ModeKind::Plan) .expect("expected plan collaboration mask"); chat.set_collaboration_mask(plan_mask); @@ -3997,7 +4018,7 @@ async fn steer_enter_queues_while_plan_stream_is_active() { chat.thread_id = Some(ThreadId::new()); chat.set_feature_enabled(Feature::CollaborationModes, true); let plan_mask = - collaboration_modes::mask_for_kind(chat.models_manager.as_ref(), ModeKind::Plan) + collaboration_modes::mask_for_kind(&chat.available_collaboration_modes, ModeKind::Plan) .expect("expected plan collaboration mask"); chat.set_collaboration_mask(plan_mask); chat.on_task_started(); @@ -4712,7 +4733,7 @@ async fn enter_submits_when_plan_stream_is_not_active() { chat.thread_id = Some(ThreadId::new()); chat.set_feature_enabled(Feature::CollaborationModes, true); let plan_mask = - collaboration_modes::mask_for_kind(chat.models_manager.as_ref(), ModeKind::Plan) + collaboration_modes::mask_for_kind(&chat.available_collaboration_modes, ModeKind::Plan) .expect("expected plan collaboration mask"); chat.set_collaboration_mask(plan_mask); chat.on_task_started(); @@ -5417,7 +5438,7 @@ async fn mode_switch_surfaces_model_change_notification_when_effective_model_cha let default_model = chat.current_model().to_string(); let mut plan_mask = - collaboration_modes::mask_for_kind(chat.models_manager.as_ref(), ModeKind::Plan) + collaboration_modes::mask_for_kind(&chat.available_collaboration_modes, ModeKind::Plan) .expect("expected plan collaboration mode"); plan_mask.model = Some("gpt-5.1-codex-mini".to_string()); chat.set_collaboration_mask(plan_mask); @@ -5432,7 +5453,7 @@ async fn mode_switch_surfaces_model_change_notification_when_effective_model_cha "expected Plan-mode model switch notice, got: {plan_messages:?}" ); - let default_mask = collaboration_modes::default_mask(chat.models_manager.as_ref()) + let default_mask = collaboration_modes::default_mask(&chat.available_collaboration_modes) .expect("expected default collaboration mode"); chat.set_collaboration_mask(default_mask); @@ -5455,7 +5476,7 @@ async fn mode_switch_surfaces_reasoning_change_notification_when_model_stays_sam chat.set_feature_enabled(Feature::CollaborationModes, true); chat.set_reasoning_effort(Some(ReasoningEffortConfig::High)); - let plan_mask = collaboration_modes::plan_mask(chat.models_manager.as_ref()) + let plan_mask = collaboration_modes::plan_mask(&chat.available_collaboration_modes) .expect("expected plan collaboration mode"); chat.set_collaboration_mask(plan_mask); @@ -5621,7 +5642,13 @@ async fn collaboration_modes_defaults_to_code_on_startup() { initial_user_message: None, enhanced_keys_supported: false, auth_manager, - models_manager: thread_manager.get_models_manager(), + available_models: thread_manager + .get_models_manager() + .try_list_models() + .expect("test models should be available"), + available_collaboration_modes: thread_manager + .get_models_manager() + .list_collaboration_modes(), feedback: codex_feedback::CodexFeedback::new(), is_first_run: true, feedback_audience: FeedbackAudience::External, @@ -5671,7 +5698,13 @@ async fn experimental_mode_plan_is_ignored_on_startup() { initial_user_message: None, enhanced_keys_supported: false, auth_manager, - models_manager: thread_manager.get_models_manager(), + available_models: thread_manager + .get_models_manager() + .try_list_models() + .expect("test models should be available"), + available_collaboration_modes: thread_manager + .get_models_manager() + .list_collaboration_modes(), feedback: codex_feedback::CodexFeedback::new(), is_first_run: true, feedback_audience: FeedbackAudience::External, @@ -5691,7 +5724,7 @@ async fn set_model_updates_active_collaboration_mask() { let (mut chat, _rx, _op_rx) = make_chatwidget_manual(Some("gpt-5.1")).await; chat.set_feature_enabled(Feature::CollaborationModes, true); let plan_mask = - collaboration_modes::mask_for_kind(chat.models_manager.as_ref(), ModeKind::Plan) + collaboration_modes::mask_for_kind(&chat.available_collaboration_modes, ModeKind::Plan) .expect("expected plan collaboration mask"); chat.set_collaboration_mask(plan_mask); @@ -5706,7 +5739,7 @@ async fn set_reasoning_effort_updates_active_collaboration_mask() { let (mut chat, _rx, _op_rx) = make_chatwidget_manual(Some("gpt-5.1")).await; chat.set_feature_enabled(Feature::CollaborationModes, true); let plan_mask = - collaboration_modes::mask_for_kind(chat.models_manager.as_ref(), ModeKind::Plan) + collaboration_modes::mask_for_kind(&chat.available_collaboration_modes, ModeKind::Plan) .expect("expected plan collaboration mask"); chat.set_collaboration_mask(plan_mask); @@ -5725,7 +5758,7 @@ async fn set_reasoning_effort_does_not_override_active_plan_override() { chat.set_feature_enabled(Feature::CollaborationModes, true); chat.set_plan_mode_reasoning_effort(Some(ReasoningEffortConfig::High)); let plan_mask = - collaboration_modes::mask_for_kind(chat.models_manager.as_ref(), ModeKind::Plan) + collaboration_modes::mask_for_kind(&chat.available_collaboration_modes, ModeKind::Plan) .expect("expected plan collaboration mask"); chat.set_collaboration_mask(plan_mask); diff --git a/codex-rs/tui/src/collaboration_modes.rs b/codex-rs/tui/src/collaboration_modes.rs index 1e5676ccde..d9ac031eea 100644 --- a/codex-rs/tui/src/collaboration_modes.rs +++ b/codex-rs/tui/src/collaboration_modes.rs @@ -1,21 +1,24 @@ -use codex_core::models_manager::manager::ModelsManager; use codex_protocol::config_types::CollaborationModeMask; use codex_protocol::config_types::ModeKind; -fn filtered_presets(models_manager: &ModelsManager) -> Vec { - models_manager - .list_collaboration_modes() - .into_iter() - .filter(|mask| mask.mode.is_some_and(ModeKind::is_tui_visible)) +fn filtered_presets(collaboration_modes: &[CollaborationModeMask]) -> Vec { + collaboration_modes + .iter() + .filter(|&mask| mask.mode.is_some_and(ModeKind::is_tui_visible)) + .cloned() .collect() } -pub(crate) fn presets_for_tui(models_manager: &ModelsManager) -> Vec { - filtered_presets(models_manager) +pub(crate) fn presets_for_tui( + collaboration_modes: &[CollaborationModeMask], +) -> Vec { + filtered_presets(collaboration_modes) } -pub(crate) fn default_mask(models_manager: &ModelsManager) -> Option { - let presets = filtered_presets(models_manager); +pub(crate) fn default_mask( + collaboration_modes: &[CollaborationModeMask], +) -> Option { + let presets = filtered_presets(collaboration_modes); presets .iter() .find(|mask| mask.mode == Some(ModeKind::Default)) @@ -24,23 +27,23 @@ pub(crate) fn default_mask(models_manager: &ModelsManager) -> Option Option { if !kind.is_tui_visible() { return None; } - filtered_presets(models_manager) + filtered_presets(collaboration_modes) .into_iter() .find(|mask| mask.mode == Some(kind)) } /// Cycle to the next collaboration mode preset in list order. pub(crate) fn next_mask( - models_manager: &ModelsManager, + collaboration_modes: &[CollaborationModeMask], current: Option<&CollaborationModeMask>, ) -> Option { - let presets = filtered_presets(models_manager); + let presets = filtered_presets(collaboration_modes); if presets.is_empty() { return None; } @@ -52,10 +55,14 @@ pub(crate) fn next_mask( presets.get(next_index).cloned() } -pub(crate) fn default_mode_mask(models_manager: &ModelsManager) -> Option { - mask_for_kind(models_manager, ModeKind::Default) +pub(crate) fn default_mode_mask( + collaboration_modes: &[CollaborationModeMask], +) -> Option { + mask_for_kind(collaboration_modes, ModeKind::Default) } -pub(crate) fn plan_mask(models_manager: &ModelsManager) -> Option { - mask_for_kind(models_manager, ModeKind::Plan) +pub(crate) fn plan_mask( + collaboration_modes: &[CollaborationModeMask], +) -> Option { + mask_for_kind(collaboration_modes, ModeKind::Plan) }