From bdcf098809d7efb271db6825a95746d4691f7dec Mon Sep 17 00:00:00 2001 From: xli-oai Date: Tue, 28 Apr 2026 22:05:27 -0700 Subject: [PATCH] Use direct remote plugin config sync --- .../src/codex_message_processor/plugins.rs | 40 ++-- .../app-server/src/config_write_router.rs | 165 ---------------- codex-rs/app-server/src/lib.rs | 1 - codex-rs/app-server/src/message_processor.rs | 182 +++++++++++++++--- codex-rs/core-plugins/src/remote.rs | 15 -- 5 files changed, 181 insertions(+), 222 deletions(-) delete mode 100644 codex-rs/app-server/src/config_write_router.rs 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 4772f98746..f749c69b7e 100644 --- a/codex-rs/app-server/src/codex_message_processor/plugins.rs +++ b/codex-rs/app-server/src/codex_message_processor/plugins.rs @@ -1,13 +1,10 @@ use super::*; -use crate::config_write_router::RemotePluginEnablementWriter; use crate::error_code::internal_error; use crate::error_code::invalid_request; -use async_trait::async_trait; use codex_app_server_protocol::PluginInstallPolicy; -#[async_trait] -impl RemotePluginEnablementWriter for CodexMessageProcessor { - async fn set_remote_plugin_enabled( +impl CodexMessageProcessor { + pub(crate) async fn sync_remote_plugin_enabled_config_write( &self, plugin_id: String, enabled: bool, @@ -18,9 +15,7 @@ impl RemotePluginEnablementWriter for CodexMessageProcessor { { return Err(invalid_request("remote plugin enablement is not enabled")); } - if plugin_id.is_empty() - || !codex_core_plugins::remote::is_valid_remote_plugin_id(&plugin_id) - { + if plugin_id.is_empty() || !is_valid_remote_plugin_id(&plugin_id) { return Err(invalid_request( "invalid remote plugin id: only ASCII letters, digits, `_`, `-`, and `~` are allowed", )); @@ -49,9 +44,7 @@ impl RemotePluginEnablementWriter for CodexMessageProcessor { self.clear_plugin_related_caches(); Ok(()) } -} -impl CodexMessageProcessor { pub(super) async fn plugin_list( &self, request_id: ConnectionRequestId, @@ -455,9 +448,7 @@ impl CodexMessageProcessor { "remote plugin install is not enabled for marketplace {remote_marketplace_name}" ))); } - if plugin_name.is_empty() - || !codex_core_plugins::remote::is_valid_remote_plugin_id(&plugin_name) - { + if plugin_name.is_empty() || !is_valid_remote_plugin_id(&plugin_name) { return Err(invalid_request( "invalid remote plugin id: only ASCII letters, digits, `_`, `-`, and `~` are allowed", )); @@ -626,13 +617,13 @@ impl CodexMessageProcessor { ) -> Result { let PluginUninstallParams { plugin_id } = params; if codex_core::plugins::PluginId::parse(&plugin_id).is_err() - && !codex_core_plugins::remote::is_supported_remote_plugin_id(&plugin_id) + && !is_valid_remote_uninstall_plugin_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 codex_core_plugins::remote::is_supported_remote_plugin_id(&plugin_id) { + if is_valid_remote_uninstall_plugin_id(&plugin_id) { return self.remote_plugin_uninstall_response(plugin_id).await; } let plugins_manager = self.thread_manager.plugins_manager(); @@ -715,9 +706,7 @@ impl CodexMessageProcessor { { return Err(invalid_request("remote plugin uninstall is not enabled")); } - if plugin_id.is_empty() - || !codex_core_plugins::remote::is_valid_remote_plugin_id(&plugin_id) - { + if plugin_id.is_empty() || !is_valid_remote_plugin_id(&plugin_id) { return Err(invalid_request( "invalid remote plugin id: only ASCII letters, digits, `_`, `-`, and `~` are allowed", )); @@ -741,6 +730,21 @@ 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_write_router.rs b/codex-rs/app-server/src/config_write_router.rs deleted file mode 100644 index 7ee86f23ec..0000000000 --- a/codex-rs/app-server/src/config_write_router.rs +++ /dev/null @@ -1,165 +0,0 @@ -use crate::config_api::ConfigApi; -use crate::error_code::invalid_request; -use async_trait::async_trait; -use codex_app_server_protocol::ConfigBatchWriteParams; -use codex_app_server_protocol::ConfigEdit; -use codex_app_server_protocol::ConfigValueWriteParams; -use codex_app_server_protocol::ConfigWriteResponse; -use codex_app_server_protocol::JSONRPCErrorError; -use serde_json::Value as JsonValue; -use std::collections::BTreeMap; -use std::sync::Arc; - -/// Applies remote plugin enablement changes whose current UI entry point is a -/// config-shaped RPC. -#[async_trait] -pub(crate) trait RemotePluginEnablementWriter: Send + Sync { - async fn set_remote_plugin_enabled( - &self, - plugin_id: String, - enabled: bool, - ) -> Result<(), JSONRPCErrorError>; -} - -#[derive(Clone)] -pub(crate) struct ConfigWriteRouter { - config_api: ConfigApi, - remote_plugin_enablement_writer: Arc, -} - -impl ConfigWriteRouter { - pub(crate) fn new( - config_api: ConfigApi, - remote_plugin_enablement_writer: Arc, - ) -> Self { - Self { - config_api, - remote_plugin_enablement_writer, - } - } - - 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 response = self - .config_api - .batch_write(ConfigBatchWriteParams { - edits: Vec::new(), - file_path: params.file_path, - expected_version: params.expected_version, - reload_user_config: false, - }) - .await?; - self.remote_plugin_enablement_writer - .set_remote_plugin_enabled(remote_plugin_edit.plugin_id, remote_plugin_edit.enabled) - .await?; - return Ok(response); - } - - self.config_api.write_value(params).await - } - - pub(crate) async fn batch_write( - &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(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 - .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.remote_plugin_enablement_writer - .set_remote_plugin_enabled(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) - } -} - -#[derive(Debug, Clone, PartialEq, Eq)] -struct RemotePluginEnabledConfigEdit { - plugin_id: String, - enabled: bool, -} - -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() - && codex_core_plugins::remote::is_supported_remote_plugin_id(plugin_id) - { - return Some(RemotePluginEnabledConfigEdit { - plugin_id: plugin_id.to_string(), - enabled, - }); - } - None -} diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index e183132724..b80510b0ff 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -77,7 +77,6 @@ mod config; mod config_api; mod config_manager; mod config_manager_service; -mod config_write_router; mod connection_rpc_gate; mod device_key_api; mod dynamic_tools; diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index c8886dc711..702d4a2121 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -1,3 +1,4 @@ +use std::collections::BTreeMap; use std::collections::HashSet; use std::future::Future; use std::sync::Arc; @@ -9,8 +10,6 @@ use crate::codex_message_processor::CodexMessageProcessor; use crate::codex_message_processor::CodexMessageProcessorArgs; use crate::config_api::ConfigApi; use crate::config_manager::ConfigManager; -use crate::config_write_router::ConfigWriteRouter; -use crate::config_write_router::RemotePluginEnablementWriter; use crate::connection_rpc_gate::ConnectionRpcGate; use crate::device_key_api::DeviceKeyApi; use crate::error_code::invalid_request; @@ -40,8 +39,10 @@ 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; @@ -86,6 +87,7 @@ 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; @@ -163,10 +165,9 @@ impl ExternalAuth for ExternalAuthRefreshBridge { pub(crate) struct MessageProcessor { outgoing: Arc, - codex_message_processor: Arc, + codex_message_processor: CodexMessageProcessor, thread_manager: Arc, config_api: ConfigApi, - config_write_router: ConfigWriteRouter, device_key_api: DeviceKeyApi, external_agent_config_api: ExternalAgentConfigApi, fs_api: FsApi, @@ -307,18 +308,17 @@ impl MessageProcessor { .plugins_manager() .set_analytics_events_client(analytics_events_client.clone()); - 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, - })); + 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, + }); 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. @@ -331,10 +331,6 @@ impl MessageProcessor { thread_manager.clone(), analytics_events_client.clone(), ); - let remote_plugin_enablement_writer: Arc = - codex_message_processor.clone(); - let config_write_router = - ConfigWriteRouter::new(config_api.clone(), remote_plugin_enablement_writer); let device_key_api = DeviceKeyApi::new(config.sqlite_home.clone(), config.model_provider_id.clone()); let external_agent_config_api = @@ -352,7 +348,6 @@ impl MessageProcessor { codex_message_processor, thread_manager: Arc::clone(&thread_manager), config_api, - config_write_router, device_key_api, external_agent_config_api, fs_api, @@ -1000,7 +995,9 @@ impl MessageProcessor { request_id: ConnectionRequestId, params: ConfigValueWriteParams, ) { - let result = self.config_write_router.write_value(params).await; + let result = self + .write_config_value_with_remote_plugin_sync(params) + .await; self.handle_config_mutation_result(request_id, result).await } @@ -1009,10 +1006,122 @@ impl MessageProcessor { request_id: ConnectionRequestId, params: ConfigBatchWriteParams, ) { - let result = self.config_write_router.batch_write(params).await; + let result = self + .batch_write_config_with_remote_plugin_sync(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, @@ -1295,6 +1404,33 @@ 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 f500edd56c..753bc86f52 100644 --- a/codex-rs/core-plugins/src/remote.rs +++ b/codex-rs/core-plugins/src/remote.rs @@ -546,21 +546,6 @@ pub async fn set_remote_plugin_enabled( Ok(()) } -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_supported_remote_plugin_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>,