From 656f6f5a39a3ff45d3aa60c865608fb88cb240a1 Mon Sep 17 00:00:00 2001 From: xli-oai Date: Tue, 28 Apr 2026 20:26:43 -0700 Subject: [PATCH] Keep remote plugin config sync out of message dispatcher --- .../src/codex_message_processor/plugins.rs | 40 ++-- codex-rs/app-server/src/config_api.rs | 125 +++++++++++++ codex-rs/app-server/src/message_processor.rs | 177 ++---------------- codex-rs/core-plugins/src/remote.rs | 44 +++++ 4 files changed, 204 insertions(+), 182 deletions(-) diff --git a/codex-rs/app-server/src/codex_message_processor/plugins.rs b/codex-rs/app-server/src/codex_message_processor/plugins.rs index f749c69b7e..61237e7e1f 100644 --- a/codex-rs/app-server/src/codex_message_processor/plugins.rs +++ b/codex-rs/app-server/src/codex_message_processor/plugins.rs @@ -1,10 +1,13 @@ use super::*; +use crate::config_api::RemotePluginConfigWriter; use crate::error_code::internal_error; use crate::error_code::invalid_request; +use async_trait::async_trait; use codex_app_server_protocol::PluginInstallPolicy; -impl CodexMessageProcessor { - pub(crate) async fn sync_remote_plugin_enabled_config_write( +#[async_trait] +impl RemotePluginConfigWriter for CodexMessageProcessor { + async fn write_remote_plugin_enabled_config( &self, plugin_id: String, enabled: bool, @@ -15,7 +18,9 @@ impl CodexMessageProcessor { { return Err(invalid_request("remote plugin enablement is not enabled")); } - if plugin_id.is_empty() || !is_valid_remote_plugin_id(&plugin_id) { + if plugin_id.is_empty() + || !codex_core_plugins::remote::is_valid_remote_plugin_id(&plugin_id) + { return Err(invalid_request( "invalid remote plugin id: only ASCII letters, digits, `_`, `-`, and `~` are allowed", )); @@ -44,7 +49,9 @@ impl CodexMessageProcessor { self.clear_plugin_related_caches(); Ok(()) } +} +impl CodexMessageProcessor { pub(super) async fn plugin_list( &self, request_id: ConnectionRequestId, @@ -448,7 +455,9 @@ impl CodexMessageProcessor { "remote plugin install is not enabled for marketplace {remote_marketplace_name}" ))); } - if plugin_name.is_empty() || !is_valid_remote_plugin_id(&plugin_name) { + if plugin_name.is_empty() + || !codex_core_plugins::remote::is_valid_remote_plugin_id(&plugin_name) + { return Err(invalid_request( "invalid remote plugin id: only ASCII letters, digits, `_`, `-`, and `~` are allowed", )); @@ -617,13 +626,13 @@ impl CodexMessageProcessor { ) -> Result { let PluginUninstallParams { plugin_id } = params; if codex_core::plugins::PluginId::parse(&plugin_id).is_err() - && !is_valid_remote_uninstall_plugin_id(&plugin_id) + && !codex_core_plugins::remote::is_remote_plugin_config_id(&plugin_id) { return Err(invalid_request( "invalid plugin id: expected a local plugin id in the form `plugin@marketplace` or a remote plugin id starting with `plugins~`, `app_`, `asdk_app_`, or `connector_`", )); } - if is_valid_remote_uninstall_plugin_id(&plugin_id) { + if codex_core_plugins::remote::is_remote_plugin_config_id(&plugin_id) { return self.remote_plugin_uninstall_response(plugin_id).await; } let plugins_manager = self.thread_manager.plugins_manager(); @@ -706,7 +715,9 @@ impl CodexMessageProcessor { { return Err(invalid_request("remote plugin uninstall is not enabled")); } - if plugin_id.is_empty() || !is_valid_remote_plugin_id(&plugin_id) { + if plugin_id.is_empty() + || !codex_core_plugins::remote::is_valid_remote_plugin_id(&plugin_id) + { return Err(invalid_request( "invalid remote plugin id: only ASCII letters, digits, `_`, `-`, and `~` are allowed", )); @@ -730,21 +741,6 @@ impl CodexMessageProcessor { } } -fn is_valid_remote_plugin_id(plugin_name: &str) -> bool { - plugin_name - .chars() - .all(|ch| ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' || ch == '~') -} - -fn is_valid_remote_uninstall_plugin_id(plugin_name: &str) -> bool { - !plugin_name.is_empty() - && is_valid_remote_plugin_id(plugin_name) - && (plugin_name.starts_with("plugins~") - || plugin_name.starts_with("app_") - || plugin_name.starts_with("asdk_app_") - || plugin_name.starts_with("connector_")) -} - fn remote_marketplace_to_info(marketplace: RemoteMarketplace) -> PluginMarketplaceEntry { PluginMarketplaceEntry { name: marketplace.name, diff --git a/codex-rs/app-server/src/config_api.rs b/codex-rs/app-server/src/config_api.rs index ac7c1510c8..ec8b78a6e5 100644 --- a/codex-rs/app-server/src/config_api.rs +++ b/codex-rs/app-server/src/config_api.rs @@ -6,6 +6,7 @@ use crate::error_code::invalid_request; use async_trait::async_trait; use codex_analytics::AnalyticsEventsClient; use codex_app_server_protocol::ConfigBatchWriteParams; +use codex_app_server_protocol::ConfigEdit; use codex_app_server_protocol::ConfigReadParams; use codex_app_server_protocol::ConfigReadResponse; use codex_app_server_protocol::ConfigRequirements; @@ -34,12 +35,14 @@ use codex_core::ThreadManager; use codex_core::config::Config; use codex_core::plugins::PluginId; use codex_core_plugins::loader::installed_plugin_telemetry_metadata; +use codex_core_plugins::remote::remote_plugin_enabled_config_edit; use codex_core_plugins::toggles::collect_plugin_enabled_candidates; use codex_features::canonical_feature_for_key; use codex_features::feature_for_key; use codex_protocol::config_types::WebSearchMode; use codex_protocol::protocol::Op; use serde_json::json; +use std::collections::BTreeMap; use std::path::PathBuf; use std::sync::Arc; use tracing::warn; @@ -59,6 +62,17 @@ pub(crate) trait UserConfigReloader: Send + Sync { async fn reload_user_config(&self); } +/// Handles config-write compatibility cases whose source of truth is outside +/// `config.toml`. +#[async_trait] +pub(crate) trait RemotePluginConfigWriter: Send + Sync { + async fn write_remote_plugin_enabled_config( + &self, + plugin_id: String, + enabled: bool, + ) -> Result<(), JSONRPCErrorError>; +} + #[async_trait] impl UserConfigReloader for ThreadManager { async fn reload_user_config(&self) { @@ -78,6 +92,7 @@ impl UserConfigReloader for ThreadManager { pub(crate) struct ConfigApi { config_manager: ConfigManager, user_config_reloader: Arc, + remote_plugin_config_writer: Option>, analytics_events_client: AnalyticsEventsClient, } @@ -90,10 +105,19 @@ impl ConfigApi { Self { config_manager, user_config_reloader, + remote_plugin_config_writer: None, analytics_events_client, } } + pub(crate) fn with_remote_plugin_config_writer( + mut self, + remote_plugin_config_writer: Arc, + ) -> Self { + self.remote_plugin_config_writer = Some(remote_plugin_config_writer); + self + } + pub(crate) async fn load_latest_config( &self, fallback_cwd: Option, @@ -153,6 +177,34 @@ impl ConfigApi { pub(crate) async fn write_value( &self, params: ConfigValueWriteParams, + ) -> Result { + if let Some(remote_plugin_edit) = + remote_plugin_enabled_config_edit(¶ms.key_path, ¶ms.value) + && let Some(remote_plugin_config_writer) = &self.remote_plugin_config_writer + { + let response = self + .batch_write_to_config(ConfigBatchWriteParams { + edits: Vec::new(), + file_path: params.file_path, + expected_version: params.expected_version, + reload_user_config: false, + }) + .await?; + remote_plugin_config_writer + .write_remote_plugin_enabled_config( + remote_plugin_edit.plugin_id, + remote_plugin_edit.enabled, + ) + .await?; + return Ok(response); + } + + self.write_value_to_config(params).await + } + + async fn write_value_to_config( + &self, + params: ConfigValueWriteParams, ) -> Result { let pending_changes = collect_plugin_enabled_candidates([(¶ms.key_path, ¶ms.value)].into_iter()); @@ -168,6 +220,79 @@ impl ConfigApi { pub(crate) async fn batch_write( &self, params: ConfigBatchWriteParams, + ) -> Result { + let Some(remote_plugin_config_writer) = &self.remote_plugin_config_writer else { + return self.batch_write_to_config(params).await; + }; + + let ConfigBatchWriteParams { + edits, + file_path, + expected_version, + reload_user_config, + } = params; + let mut local_edits = Vec::::new(); + let mut remote_plugin_toggles = BTreeMap::::new(); + + for edit in edits { + if let Some(remote_plugin_edit) = + remote_plugin_enabled_config_edit(&edit.key_path, &edit.value) + { + remote_plugin_toggles + .insert(remote_plugin_edit.plugin_id, remote_plugin_edit.enabled); + } else { + local_edits.push(edit); + } + } + + if !remote_plugin_toggles.is_empty() && !local_edits.is_empty() { + return Err(invalid_request( + "remote plugin enablement edits cannot be batched with local config edits", + )); + } + + if remote_plugin_toggles.is_empty() { + return self + .batch_write_to_config(ConfigBatchWriteParams { + edits: local_edits, + file_path, + expected_version, + reload_user_config, + }) + .await; + } + + let response = self + .batch_write_to_config(ConfigBatchWriteParams { + edits: Vec::new(), + file_path: file_path.clone(), + expected_version, + reload_user_config: false, + }) + .await?; + + for (plugin_id, enabled) in remote_plugin_toggles { + remote_plugin_config_writer + .write_remote_plugin_enabled_config(plugin_id, enabled) + .await?; + } + + if reload_user_config { + self.batch_write_to_config(ConfigBatchWriteParams { + edits: Vec::new(), + file_path, + expected_version: None, + reload_user_config: true, + }) + .await?; + } + + Ok(response) + } + + async fn batch_write_to_config( + &self, + params: ConfigBatchWriteParams, ) -> Result { let reload_user_config = params.reload_user_config; let pending_changes = collect_plugin_enabled_candidates( diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 702d4a2121..5bc534286b 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -1,4 +1,3 @@ -use std::collections::BTreeMap; use std::collections::HashSet; use std::future::Future; use std::sync::Arc; @@ -39,10 +38,8 @@ use codex_app_server_protocol::ClientInfo; use codex_app_server_protocol::ClientNotification; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::ConfigBatchWriteParams; -use codex_app_server_protocol::ConfigEdit; use codex_app_server_protocol::ConfigValueWriteParams; use codex_app_server_protocol::ConfigWarningNotification; -use codex_app_server_protocol::ConfigWriteResponse; use codex_app_server_protocol::DeviceKeyCreateParams; use codex_app_server_protocol::DeviceKeyPublicParams; use codex_app_server_protocol::DeviceKeySignParams; @@ -87,7 +84,6 @@ use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::W3cTraceContext; use codex_state::log_db::LogDbLayer; use futures::FutureExt; -use serde_json::Value as JsonValue; use tokio::sync::broadcast; use tokio::sync::watch; use tokio::time::Duration; @@ -165,7 +161,7 @@ impl ExternalAuth for ExternalAuthRefreshBridge { pub(crate) struct MessageProcessor { outgoing: Arc, - codex_message_processor: CodexMessageProcessor, + codex_message_processor: Arc, thread_manager: Arc, config_api: ConfigApi, device_key_api: DeviceKeyApi, @@ -308,17 +304,18 @@ impl MessageProcessor { .plugins_manager() .set_analytics_events_client(analytics_events_client.clone()); - let codex_message_processor = CodexMessageProcessor::new(CodexMessageProcessorArgs { - auth_manager: auth_manager.clone(), - thread_manager: Arc::clone(&thread_manager), - outgoing: outgoing.clone(), - analytics_events_client: analytics_events_client.clone(), - arg0_paths, - config: Arc::clone(&config), - config_manager: config_manager.clone(), - feedback, - log_db, - }); + let codex_message_processor = + Arc::new(CodexMessageProcessor::new(CodexMessageProcessorArgs { + auth_manager: auth_manager.clone(), + thread_manager: Arc::clone(&thread_manager), + outgoing: outgoing.clone(), + analytics_events_client: analytics_events_client.clone(), + arg0_paths, + config: Arc::clone(&config), + config_manager: config_manager.clone(), + feedback, + log_db, + })); if matches!(plugin_startup_tasks, crate::PluginStartupTasks::Start) { // Keep plugin startup warmups aligned at app-server startup. // TODO(xl): Move into PluginManager once this no longer depends on config feature gating. @@ -330,7 +327,8 @@ impl MessageProcessor { config_manager, thread_manager.clone(), analytics_events_client.clone(), - ); + ) + .with_remote_plugin_config_writer(codex_message_processor.clone()); let device_key_api = DeviceKeyApi::new(config.sqlite_home.clone(), config.model_provider_id.clone()); let external_agent_config_api = @@ -995,9 +993,7 @@ impl MessageProcessor { request_id: ConnectionRequestId, params: ConfigValueWriteParams, ) { - let result = self - .write_config_value_with_remote_plugin_sync(params) - .await; + let result = self.config_api.write_value(params).await; self.handle_config_mutation_result(request_id, result).await } @@ -1006,122 +1002,10 @@ impl MessageProcessor { request_id: ConnectionRequestId, params: ConfigBatchWriteParams, ) { - let result = self - .batch_write_config_with_remote_plugin_sync(params) - .await; + let result = self.config_api.batch_write(params).await; self.handle_config_mutation_result(request_id, result).await; } - async fn write_config_value_with_remote_plugin_sync( - &self, - params: ConfigValueWriteParams, - ) -> Result { - let ConfigValueWriteParams { - key_path, - value, - merge_strategy, - file_path, - expected_version, - } = params; - - if let Some((plugin_id, enabled)) = remote_plugin_enabled_config_edit(&key_path, &value) { - let response = self - .config_api - .batch_write(ConfigBatchWriteParams { - edits: Vec::new(), - file_path, - expected_version, - reload_user_config: false, - }) - .await?; - self.codex_message_processor - .sync_remote_plugin_enabled_config_write(plugin_id, enabled) - .await?; - return Ok(response); - } - - self.config_api - .write_value(ConfigValueWriteParams { - key_path, - value, - merge_strategy, - file_path, - expected_version, - }) - .await - } - - async fn batch_write_config_with_remote_plugin_sync( - &self, - params: ConfigBatchWriteParams, - ) -> Result { - let ConfigBatchWriteParams { - edits, - file_path, - expected_version, - reload_user_config, - } = params; - let mut local_edits = Vec::::new(); - let mut remote_plugin_toggles = BTreeMap::::new(); - - for edit in edits { - if let Some((plugin_id, enabled)) = - remote_plugin_enabled_config_edit(&edit.key_path, &edit.value) - { - remote_plugin_toggles.insert(plugin_id, enabled); - } else { - local_edits.push(edit); - } - } - - if !remote_plugin_toggles.is_empty() && !local_edits.is_empty() { - return Err(invalid_request( - "remote plugin enablement edits cannot be batched with local config edits", - )); - } - - if remote_plugin_toggles.is_empty() { - return self - .config_api - .batch_write(ConfigBatchWriteParams { - edits: local_edits, - file_path, - expected_version, - reload_user_config, - }) - .await; - } - - let response = self - .config_api - .batch_write(ConfigBatchWriteParams { - edits: Vec::new(), - file_path: file_path.clone(), - expected_version, - reload_user_config: false, - }) - .await?; - - for (plugin_id, enabled) in remote_plugin_toggles { - self.codex_message_processor - .sync_remote_plugin_enabled_config_write(plugin_id, enabled) - .await?; - } - - if reload_user_config { - self.config_api - .batch_write(ConfigBatchWriteParams { - edits: Vec::new(), - file_path, - expected_version: None, - reload_user_config: true, - }) - .await?; - } - - Ok(response) - } - async fn handle_experimental_feature_enablement_set( &self, request_id: ConnectionRequestId, @@ -1404,33 +1288,6 @@ fn migration_items_need_runtime_refresh(items: &[ExternalAgentConfigMigrationIte }) } -fn remote_plugin_enabled_config_edit(key_path: &str, value: &JsonValue) -> Option<(String, bool)> { - let enabled = value.as_bool()?; - let mut segments = key_path.split('.'); - let table = segments.next()?; - let plugin_id = segments.next()?; - let field = segments.next()?; - if table == "plugins" - && field == "enabled" - && segments.next().is_none() - && is_remote_plugin_config_id(plugin_id) - { - return Some((plugin_id.to_string(), enabled)); - } - None -} - -fn is_remote_plugin_config_id(plugin_id: &str) -> bool { - !plugin_id.is_empty() - && plugin_id - .chars() - .all(|ch| ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' || ch == '~') - && (plugin_id.starts_with("plugins~") - || plugin_id.starts_with("app_") - || plugin_id.starts_with("asdk_app_") - || plugin_id.starts_with("connector_")) -} - #[cfg(test)] mod tracing_tests; diff --git a/codex-rs/core-plugins/src/remote.rs b/codex-rs/core-plugins/src/remote.rs index 753bc86f52..097e168d6a 100644 --- a/codex-rs/core-plugins/src/remote.rs +++ b/codex-rs/core-plugins/src/remote.rs @@ -9,6 +9,7 @@ use codex_login::default_client::build_reqwest_client; use codex_plugin::PluginId; use reqwest::RequestBuilder; use serde::Deserialize; +use serde_json::Value as JsonValue; use std::collections::BTreeMap; use std::collections::BTreeSet; use std::collections::HashSet; @@ -32,6 +33,12 @@ pub struct RemotePluginServiceConfig { pub chatgpt_base_url: String, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RemotePluginEnabledConfigEdit { + pub plugin_id: String, + pub enabled: bool, +} + #[derive(Debug, Clone, PartialEq)] pub struct RemoteMarketplace { pub name: String, @@ -546,6 +553,43 @@ pub async fn set_remote_plugin_enabled( Ok(()) } +pub fn remote_plugin_enabled_config_edit( + key_path: &str, + value: &JsonValue, +) -> Option { + let enabled = value.as_bool()?; + let mut segments = key_path.split('.'); + let table = segments.next()?; + let plugin_id = segments.next()?; + let field = segments.next()?; + if table == "plugins" + && field == "enabled" + && segments.next().is_none() + && is_remote_plugin_config_id(plugin_id) + { + return Some(RemotePluginEnabledConfigEdit { + plugin_id: plugin_id.to_string(), + enabled, + }); + } + None +} + +pub fn is_valid_remote_plugin_id(plugin_id: &str) -> bool { + plugin_id + .chars() + .all(|ch| ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' || ch == '~') +} + +pub fn is_remote_plugin_config_id(plugin_id: &str) -> bool { + !plugin_id.is_empty() + && is_valid_remote_plugin_id(plugin_id) + && (plugin_id.starts_with("plugins~") + || plugin_id.starts_with("app_") + || plugin_id.starts_with("asdk_app_") + || plugin_id.starts_with("connector_")) +} + pub async fn uninstall_remote_plugin( config: &RemotePluginServiceConfig, auth: Option<&CodexAuth>,