From 2d3de3783460a57f1e089b383f8d5d4870a41779 Mon Sep 17 00:00:00 2001 From: Dmitriy Kovalenko Date: Fri, 23 Oct 2026 18:40:55 +0000 Subject: [PATCH] Surface automatic marketplace update failures as startup warnings (#57166) ## Why Automatic marketplace update failures were only logged, leaving users without a visible startup warning. ## What changed Send marketplace update errors and task startup failures through app-server warning notifications. Queue background warnings until a client initializes in both standalone and in-process servers, and centralize startup configuration warning delivery. ## Testing Add a TUI integration test and snapshot covering two marketplace name mismatch failures, the warning count, and navigation between their details with `F2` and the arrow keys. GitOrigin-RevId: af55a5555affd9d89ac15a546163daeaafd3b3e8 --- codex-rs/app-server/src/in_process.rs | 5 + codex-rs/app-server/src/lib.rs | 6 + codex-rs/app-server/src/message_processor.rs | 36 +++--- .../src/message_processor_tracing_tests.rs | 1 + ...ssage_processor_workspace_routing_tests.rs | 1 + .../initialize_processor.rs | 25 ---- codex-rs/app-server/src/startup_warnings.rs | 39 +++++++ .../src/user_verification_test_support.rs | 1 + codex-rs/core-plugins/src/manager.rs | 12 ++ codex-rs/tui/src/app/tests.rs | 2 + .../app/tests/marketplace_upgrade_tests.rs | 110 ++++++++++++++++++ ...de_tests__marketplace_upgrade_warning.snap | 57 +++++++++ 12 files changed, 256 insertions(+), 39 deletions(-) create mode 100644 codex-rs/app-server/src/startup_warnings.rs create mode 100644 codex-rs/tui/src/app/tests/marketplace_upgrade_tests.rs create mode 100644 codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__marketplace_upgrade_tests__marketplace_upgrade_warning.snap diff --git a/codex-rs/app-server/src/in_process.rs b/codex-rs/app-server/src/in_process.rs index b61c78e8e1..c9576732b0 100644 --- a/codex-rs/app-server/src/in_process.rs +++ b/codex-rs/app-server/src/in_process.rs @@ -500,6 +500,7 @@ async fn start_uninitialized(mut args: InProcessStartArgs) -> IoResult(channel_capacity); let mut processor_handle = tokio::spawn(async move { + let (startup_warnings_tx, mut startup_warnings_rx) = mpsc::unbounded_channel(); let processor = Arc::new(MessageProcessor::new(MessageProcessorArgs { cloud_skill_provider: None, outgoing: Arc::clone(&processor_outgoing), @@ -513,6 +514,7 @@ async fn start_uninitialized(mut args: InProcessStartArgs) -> IoResult IoResult { + crate::startup_warnings::send_warning(&processor_outgoing, message).await; + } created = thread_created_rx.recv(), if listen_for_threads => { match created { Ok(thread_id) => { diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index ae5941119f..8abfa4aa5d 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -143,6 +143,7 @@ mod request_processors; mod request_serialization; mod server_request_error; mod skills_watcher; +mod startup_warnings; mod thread_sample; mod thread_state; mod thread_status; @@ -1007,6 +1008,7 @@ pub async fn run_main_with_transport_options( let auth_manager = Arc::clone(&auth_manager); let initialize_notification_sender = outgoing_message_sender.clone(); let outbound_control_tx = outbound_control_tx; + let (startup_warnings_tx, mut startup_warnings_rx) = mpsc::unbounded_channel(); let mut processor = MessageProcessor::new(MessageProcessorArgs { cloud_skill_provider, outgoing: outgoing_message_sender, @@ -1020,6 +1022,7 @@ pub async fn run_main_with_transport_options( log_db, state_db: state_db.clone(), config_warnings, + startup_warnings_tx, session_source, user_verification: Arc::new(crate::user_verification::Service::new(Arc::clone( &auth_manager, @@ -1301,6 +1304,9 @@ pub async fn run_main_with_transport_options( } } } + Some(message) = startup_warnings_rx.recv(), if connections.values().any(|state| state.session.initialized()) => { + startup_warnings::send_warning(&initialize_notification_sender, message).await; + } _ = connection_cleanup_tasks.reap_next() => {} result = thread_listener_tasks.join_next(), if !thread_listener_tasks.is_empty() => { if let Some(Err(err)) = result { diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index a1b1133119..3162839f08 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -55,6 +55,7 @@ use crate::request_serialization::QueuedInitializedRequest; use crate::request_serialization::RequestSerializationQueueKey; use crate::request_serialization::RequestSerializationQueues; use crate::skills_watcher::SkillsWatcher; +use crate::startup_warnings::StartupWarnings; use crate::thread_state::ConnectionCapabilities; use crate::thread_state::ThreadStateManager; use crate::transport::AppServerTransport; @@ -99,6 +100,7 @@ use codex_thread_store::QueueStore; use tokio::sync::Mutex; use tokio::sync::Semaphore; use tokio::sync::broadcast; +use tokio::sync::mpsc; use tokio::sync::watch; use tokio::time::Duration; use tokio::time::timeout; @@ -145,6 +147,7 @@ pub(crate) struct MessageProcessor { pub(crate) turn_admission: TurnAdmission, user_verification: Arc, outgoing: Arc, + startup_warnings: StartupWarnings, models_refresh_worker: ModelsRefreshWorker, turn_cost_worker: Option, skills_watcher: Arc, @@ -262,6 +265,7 @@ pub(crate) struct MessageProcessorArgs { pub(crate) log_db: Option, pub(crate) state_db: Option, pub(crate) config_warnings: Vec, + pub(crate) startup_warnings_tx: mpsc::UnboundedSender, pub(crate) session_source: SessionSource, pub(crate) auth_manager: Arc, pub(crate) user_verification: Arc, @@ -290,6 +294,7 @@ impl MessageProcessor { log_db, state_db, config_warnings, + startup_warnings_tx, session_source, auth_manager, user_verification, @@ -473,15 +478,6 @@ impl MessageProcessor { state_db.clone(), ); let git_processor = GitRequestProcessor::new(); - let initialize_processor = InitializeRequestProcessor::new( - gateway_login_control, - outgoing.clone(), - analytics_events_client.clone(), - Arc::clone(&config), - config_warnings.clone(), - rpc_transport, - Arc::clone(&user_verification), - ); let marketplace_processor = MarketplaceRequestProcessor::new(config_manager.clone(), Arc::clone(&thread_manager)); let mcp_processor = McpRequestProcessor::new( @@ -540,7 +536,7 @@ impl MessageProcessor { log_db, Arc::clone(&skills_watcher), turn_cost_worker.as_ref().map(TurnCostWorker::handle), - config_warnings, + config_warnings.clone(), sample_responses, ); let turn_processor = TurnRequestProcessor::new( @@ -574,8 +570,17 @@ impl MessageProcessor { &config.plugins_config_input(), reload_config, Some(on_effective_plugins_changed), + startup_warnings_tx, ); } + let initialize_processor = InitializeRequestProcessor::new( + gateway_login_control, + outgoing.clone(), + analytics_events_client.clone(), + Arc::clone(&config), + rpc_transport, + Arc::clone(&user_verification), + ); let external_agent_config_processor = ExternalAgentConfigRequestProcessor::new(ExternalAgentConfigRequestProcessorArgs { outgoing: outgoing.clone(), @@ -604,6 +609,9 @@ impl MessageProcessor { turn_admission, user_verification, outgoing, + startup_warnings: StartupWarnings { + config: config_warnings, + }, models_refresh_worker, turn_cost_worker, skills_watcher, @@ -822,8 +830,8 @@ impl MessageProcessor { &self, connection_id: ConnectionId, ) { - self.initialize_processor - .send_initialize_notifications_to_connection(connection_id) + self.startup_warnings + .send_config_to_connections(&self.outgoing, &[connection_id]) .await; } @@ -845,8 +853,8 @@ impl MessageProcessor { } pub(crate) async fn send_initialize_notifications(&self) { - self.initialize_processor - .send_initialize_notifications() + self.startup_warnings + .send_config_to_connections(&self.outgoing, &[]) .await; } diff --git a/codex-rs/app-server/src/message_processor_tracing_tests.rs b/codex-rs/app-server/src/message_processor_tracing_tests.rs index 2bb77bf6fb..aeff494f09 100644 --- a/codex-rs/app-server/src/message_processor_tracing_tests.rs +++ b/codex-rs/app-server/src/message_processor_tracing_tests.rs @@ -273,6 +273,7 @@ pub(super) async fn build_test_processor( log_db: None, state_db: None, config_warnings: Vec::new(), + startup_warnings_tx: mpsc::unbounded_channel().0, session_source: SessionSource::VSCode, user_verification: Arc::new(crate::user_verification::Service::new(Arc::clone( &auth_manager, diff --git a/codex-rs/app-server/src/message_processor_workspace_routing_tests.rs b/codex-rs/app-server/src/message_processor_workspace_routing_tests.rs index cfb49bea5b..e3f911f6db 100644 --- a/codex-rs/app-server/src/message_processor_workspace_routing_tests.rs +++ b/codex-rs/app-server/src/message_processor_workspace_routing_tests.rs @@ -76,6 +76,7 @@ async fn background_routing_call_sites_skip_config(api_key: Option<&str>) -> any log_db: None, state_db: None, config_warnings: Vec::new(), + startup_warnings_tx: mpsc::unbounded_channel().0, session_source: SessionSource::VSCode, user_verification: Arc::new(crate::user_verification::Service::new(Arc::clone( &auth_manager, diff --git a/codex-rs/app-server/src/request_processors/initialize_processor.rs b/codex-rs/app-server/src/request_processors/initialize_processor.rs index d68a29df9f..6259cf7488 100644 --- a/codex-rs/app-server/src/request_processors/initialize_processor.rs +++ b/codex-rs/app-server/src/request_processors/initialize_processor.rs @@ -25,7 +25,6 @@ pub(crate) struct InitializeRequestProcessor { outgoing: Arc, analytics_events_client: AnalyticsEventsClient, config: Arc, - config_warnings: Arc>, rpc_transport: AppServerRpcTransport, user_verification: Arc, } @@ -37,7 +36,6 @@ impl InitializeRequestProcessor { outgoing: Arc, analytics_events_client: AnalyticsEventsClient, config: Arc, - config_warnings: Vec, rpc_transport: AppServerRpcTransport, user_verification: Arc, ) -> Self { @@ -47,7 +45,6 @@ impl InitializeRequestProcessor { outgoing, analytics_events_client, config, - config_warnings: Arc::new(config_warnings), rpc_transport, user_verification, } @@ -229,28 +226,6 @@ impl InitializeRequestProcessor { Ok(false) } - pub(crate) async fn send_initialize_notifications_to_connection( - &self, - connection_id: ConnectionId, - ) { - for notification in self.config_warnings.iter().cloned() { - self.outgoing - .send_server_notification_to_connections( - &[connection_id], - ServerNotification::ConfigWarning(notification), - ) - .await; - } - } - - pub(crate) async fn send_initialize_notifications(&self) { - for notification in self.config_warnings.iter().cloned() { - self.outgoing - .send_server_notification(ServerNotification::ConfigWarning(notification)) - .await; - } - } - pub(crate) fn track_initialized_request( &self, connection_id: ConnectionId, diff --git a/codex-rs/app-server/src/startup_warnings.rs b/codex-rs/app-server/src/startup_warnings.rs new file mode 100644 index 0000000000..b2eec2f6b2 --- /dev/null +++ b/codex-rs/app-server/src/startup_warnings.rs @@ -0,0 +1,39 @@ +//! Delivers configuration and background task warnings after a client initializes. +//! The runtime event loop owns the receiver and consumes background warnings once. + +use codex_app_server_protocol::ConfigWarningNotification; +use codex_app_server_protocol::ServerNotification; +use codex_app_server_protocol::WarningNotification; + +use crate::outgoing_message::ConnectionId; +use crate::outgoing_message::OutgoingMessageSender; + +pub(crate) struct StartupWarnings { + pub(crate) config: Vec, +} + +impl StartupWarnings { + pub(crate) async fn send_config_to_connections( + &self, + outgoing: &OutgoingMessageSender, + connection_ids: &[ConnectionId], + ) { + for notification in self.config.iter().cloned() { + outgoing + .send_server_notification_to_connections( + connection_ids, + ServerNotification::ConfigWarning(notification), + ) + .await; + } + } +} + +pub(crate) async fn send_warning(outgoing: &OutgoingMessageSender, message: String) { + outgoing + .send_server_notification(ServerNotification::Warning(WarningNotification { + thread_id: None, + message, + })) + .await; +} diff --git a/codex-rs/app-server/src/user_verification_test_support.rs b/codex-rs/app-server/src/user_verification_test_support.rs index 28a35ca132..1a392d102c 100644 --- a/codex-rs/app-server/src/user_verification_test_support.rs +++ b/codex-rs/app-server/src/user_verification_test_support.rs @@ -166,6 +166,7 @@ impl Harness { log_db: None, state_db: None, config_warnings: Vec::new(), + startup_warnings_tx: mpsc::unbounded_channel().0, session_source: SessionSource::Cli, auth_manager: Arc::clone(&auth), user_verification: Arc::clone(&service), diff --git a/codex-rs/core-plugins/src/manager.rs b/codex-rs/core-plugins/src/manager.rs index bad8481127..6d9d676258 100644 --- a/codex-rs/core-plugins/src/manager.rs +++ b/codex-rs/core-plugins/src/manager.rs @@ -145,6 +145,7 @@ use std::time::Instant; use tokio::sync::OnceCell; use tokio::sync::OwnedSemaphorePermit; use tokio::sync::Semaphore; +use tokio::sync::mpsc; use tokio::sync::watch; use tracing::instrument; use tracing::warn; @@ -2869,11 +2870,13 @@ impl PluginsManager { }) } + /// Starts background tasks and sends marketplace upgrade warnings through the supplied channel. pub fn maybe_start_plugin_startup_tasks_for_config( self: &Arc, config: &PluginsConfigInput, reload_config: ConfigLayerReload, on_effective_plugins_changed: Option, + upgrade_warnings: mpsc::UnboundedSender, ) { if config.plugins_enabled { self.maybe_start_curated_repo_sync_for_config( @@ -2897,6 +2900,7 @@ impl PluginsManager { let config = config.clone(); let on_effective_plugins_changed = on_effective_plugins_changed.clone(); let runtime = tokio::runtime::Handle::current(); + let task_warnings = upgrade_warnings.clone(); if let Err(err) = std::thread::Builder::new() .name("plugins-marketplace-auto-upgrade".to_string()) .spawn(move || { @@ -2924,10 +2928,16 @@ impl PluginsManager { error = %error.message, "failed to auto-upgrade configured marketplace" ); + let _ = task_warnings.send(format!( + "Codex could not update marketplace `{}`: {}", + error.marketplace_name, error.message + )); } } Err(err) => { warn!("failed to auto-upgrade configured marketplaces: {err}"); + let _ = task_warnings + .send(format!("Codex could not update marketplaces: {err}")); } } @@ -2944,6 +2954,8 @@ impl PluginsManager { }; state.in_flight = false; warn!("failed to start configured marketplace auto-upgrade task: {err}"); + let _ = upgrade_warnings + .send(format!("Codex could not start marketplace updates: {err}")); } } let config_for_remote_sync = config.clone(); diff --git a/codex-rs/tui/src/app/tests.rs b/codex-rs/tui/src/app/tests.rs index ae4532818a..c0c378b16d 100644 --- a/codex-rs/tui/src/app/tests.rs +++ b/codex-rs/tui/src/app/tests.rs @@ -51,6 +51,8 @@ mod key_chords; mod local_command_scroll_tests; #[path = "tests/luna_reserve_recovery_tests.rs"] mod luna_reserve_recovery_tests; +#[path = "tests/marketplace_upgrade_tests.rs"] +mod marketplace_upgrade_tests; #[path = "tests/mcp_startup.rs"] mod mcp_startup; #[path = "tests/misalignment_policy_tests.rs"] diff --git a/codex-rs/tui/src/app/tests/marketplace_upgrade_tests.rs b/codex-rs/tui/src/app/tests/marketplace_upgrade_tests.rs new file mode 100644 index 0000000000..173268cb0d --- /dev/null +++ b/codex-rs/tui/src/app/tests/marketplace_upgrade_tests.rs @@ -0,0 +1,110 @@ +//! Marketplace startup failures reach the warning notice and its details view. + +use super::*; +use crate::app::owned_transcript::tests::buffer_text; +use crate::chatwidget::tests::helpers::normalize_snapshot_paths; +use crate::custom_terminal::test_support::last_rendered_buffer; +use codex_app_server_client::AppServerEvent; +use codex_config::MarketplaceConfigUpdate; +use codex_config::record_user_marketplace; + +#[tokio::test] +async fn automatic_marketplace_upgrade_failure_shows_warning() -> Result<()> { + let (mut app, mut events, _ops) = make_test_app_with_channels().await; + let source = tempdir()?; + std::fs::create_dir_all(source.path().join(".agents/plugins"))?; + std::fs::write( + source.path().join(".agents/plugins/marketplace.json"), + r#"{"name":"other","plugins":[]}"#, + )?; + for args in [ + vec!["init"], + vec!["add", "."], + vec!["commit", "-m", "initial marketplace"], + ] { + let output = std::process::Command::new("git") + .args(["-c", "commit.gpgSign=false"]) + .args(args) + .current_dir(source.path()) + .env("GIT_AUTHOR_NAME", "Test") + .env("GIT_AUTHOR_EMAIL", "test@example.com") + .env("GIT_COMMITTER_NAME", "Test") + .env("GIT_COMMITTER_EMAIL", "test@example.com") + .output()?; + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + } + // Both checkouts succeed, but their names do not match the configured marketplaces. + let marketplace_names = ["team", "work"]; + for marketplace_name in marketplace_names { + record_user_marketplace( + app.config.codex_home.as_path(), + marketplace_name, + &MarketplaceConfigUpdate { + source_type: "git", + source: &source.path().display().to_string(), + ref_name: None, + sparse_paths: &[], + }, + )?; + } + app.config = ConfigBuilder::default() + .codex_home(app.config.codex_home.to_path_buf()) + .loader_overrides(LoaderOverrides::without_managed_config_for_tests()) + .harness_overrides(ConfigOverrides { + cwd: Some(source.path().to_path_buf()), + ..Default::default() + }) + .build() + .await?; + app.local_settings.tui.animations = false; + app.local_settings.tui.show_tooltips = false; + while events.try_recv().is_ok() {} + + let mut server = Box::pin(crate::start_embedded_app_server_for_picker(&app.config)).await?; + let mut warnings_received = 0; + while warnings_received < marketplace_names.len() { + let event = server.next_event().await.expect("app-server notification"); + let is_warning = matches!( + &event.event, + AppServerEvent::ServerNotification(notification) + if matches!(notification.as_ref(), ServerNotification::Warning(_)) + ); + app.handle_backend_event(&server, event).await; + if is_warning { + warnings_received += 1; + } + } + + let mut tui = crate::tui::test_support::make_test_tui()?; + tui.set_owned_screen(/*owned*/ true)?; + while let Ok(event) = events.try_recv() { + Box::pin(app.handle_event(&mut tui, &mut server, event)).await?; + } + let size = Size::new(/*width*/ 80, /*height*/ 16); + tui.screen_size_for_event(&TuiEvent::Resize(size))?; + app.render_owned_transcript(&mut tui, size)?; + let notice = buffer_text(last_rendered_buffer(&tui.terminal)); + + Box::pin(app.handle_tui_event(&mut tui, &mut server, TuiEvent::Key(KeyCode::F(2).into()))) + .await?; + app.render_owned_transcript(&mut tui, size)?; + let first_details = buffer_text(last_rendered_buffer(&tui.terminal)); + + Box::pin(app.handle_tui_event(&mut tui, &mut server, TuiEvent::Key(KeyCode::Right.into()))) + .await?; + app.render_owned_transcript(&mut tui, size)?; + let second_details = buffer_text(last_rendered_buffer(&tui.terminal)); + server.shutdown().await?; + + insta::assert_snapshot!( + "marketplace_upgrade_warning", + normalize_snapshot_paths(format!( + "Warning notice:\n{notice}\n\nFirst warning details:\n{first_details}\n\nSecond warning details:\n{second_details}" + )) + ); + Ok(()) +} diff --git a/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__marketplace_upgrade_tests__marketplace_upgrade_warning.snap b/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__marketplace_upgrade_tests__marketplace_upgrade_warning.snap new file mode 100644 index 0000000000..1cb3812db0 --- /dev/null +++ b/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__marketplace_upgrade_tests__marketplace_upgrade_warning.snap @@ -0,0 +1,57 @@ +--- +source: tui/src/app/tests/marketplace_upgrade_tests.rs +expression: "normalize_snapshot_paths(format!(\"Warning notice:\\n{notice}\\n\\nFirst warning details:\\n{first_details}\\n\\nSecond warning details:\\n{second_details}\"))" +--- +Warning notice: + + + + + + + + + + + + +› Ask Codex to do anything + + GPT-5.6-Sol default · /tmp/project + ? for shortcuts ⚠ 2 warnings · f2 to view + +First warning details: + + + + + Warnings · 1 of 2 · Startup + + Codex could not update marketplace `team`: upgraded marketplace name `other` + does not match configured marketplace `team` + + + + + + + + k keep & next · esc dismiss & close · ⌃o copy · ←/→ warning · ↓ scroll + +Second warning details: + + + + + Warnings · 2 of 2 · Startup + + Codex could not update marketplace `work`: upgraded marketplace name `other` + does not match configured marketplace `work` + + + + + + + + k keep & next · esc dismiss & close · ⌃o copy · ←/→ warning · ↓ scroll