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
This commit is contained in:
Dmitriy Kovalenko
2026-10-26 17:56:43 +00:00
committed by copyberry
parent 108cf2975a
commit 2d3de37834
12 changed files with 256 additions and 39 deletions
+5
View File
@@ -500,6 +500,7 @@ async fn start_uninitialized(mut args: InProcessStartArgs) -> IoResult<InProcess
let processor_outgoing = Arc::clone(&outgoing_message_sender);
let (processor_tx, mut processor_rx) = mpsc::channel::<ProcessorCommand>(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<InProcess
log_db: args.log_db,
state_db: args.state_db,
config_warnings: args.config_warnings,
startup_warnings_tx,
session_source: args.session_source,
user_verification: Arc::new(crate::user_verification::Service::new(Arc::clone(
&auth_manager,
@@ -574,6 +576,9 @@ async fn start_uninitialized(mut args: InProcessStartArgs) -> IoResult<InProcess
}
}
}
Some(message) = startup_warnings_rx.recv(), if session.initialized() => {
crate::startup_warnings::send_warning(&processor_outgoing, message).await;
}
created = thread_created_rx.recv(), if listen_for_threads => {
match created {
Ok(thread_id) => {
+6
View File
@@ -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 {
+22 -14
View File
@@ -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<crate::user_verification::Service>,
outgoing: Arc<OutgoingMessageSender>,
startup_warnings: StartupWarnings,
models_refresh_worker: ModelsRefreshWorker,
turn_cost_worker: Option<TurnCostWorker>,
skills_watcher: Arc<SkillsWatcher>,
@@ -262,6 +265,7 @@ pub(crate) struct MessageProcessorArgs {
pub(crate) log_db: Option<LogDbLayer>,
pub(crate) state_db: Option<StateDbHandle>,
pub(crate) config_warnings: Vec<ConfigWarningNotification>,
pub(crate) startup_warnings_tx: mpsc::UnboundedSender<String>,
pub(crate) session_source: SessionSource,
pub(crate) auth_manager: Arc<AuthManager>,
pub(crate) user_verification: Arc<crate::user_verification::Service>,
@@ -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;
}
@@ -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,
@@ -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,
@@ -25,7 +25,6 @@ pub(crate) struct InitializeRequestProcessor {
outgoing: Arc<OutgoingMessageSender>,
analytics_events_client: AnalyticsEventsClient,
config: Arc<Config>,
config_warnings: Arc<Vec<ConfigWarningNotification>>,
rpc_transport: AppServerRpcTransport,
user_verification: Arc<crate::user_verification::Service>,
}
@@ -37,7 +36,6 @@ impl InitializeRequestProcessor {
outgoing: Arc<OutgoingMessageSender>,
analytics_events_client: AnalyticsEventsClient,
config: Arc<Config>,
config_warnings: Vec<ConfigWarningNotification>,
rpc_transport: AppServerRpcTransport,
user_verification: Arc<crate::user_verification::Service>,
) -> 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,
@@ -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<ConfigWarningNotification>,
}
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;
}
@@ -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),
+12
View File
@@ -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<Self>,
config: &PluginsConfigInput,
reload_config: ConfigLayerReload,
on_effective_plugins_changed: Option<EffectivePluginsChangedCallback>,
upgrade_warnings: mpsc::UnboundedSender<String>,
) {
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();
+2
View File
@@ -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"]
@@ -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(())
}
@@ -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