openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-27 18:48:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Keep remote plugin config sync out of message dispatcher

Browse Source
This commit is contained in:
xli-oai
2026-04-28 20:27:41 -07:00
parent 010064afba
commit 656f6f5a39
4 changed files with 204 additions and 182 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/codex_message_processor/plugins.rs
+18 -22
View File
@@ -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<PluginUninstallResponse, JSONRPCErrorError> {
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,
codex-rs/app-server/src/config_api.rs
+125
View File
@@ -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<dyn UserConfigReloader>,
remote_plugin_config_writer: Option<Arc<dyn RemotePluginConfigWriter>>,
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<dyn RemotePluginConfigWriter>,
) -> Self {
self.remote_plugin_config_writer = Some(remote_plugin_config_writer);
self
}
pub(crate) async fn load_latest_config(
&self,
fallback_cwd: Option<PathBuf>,
@@ -153,6 +177,34 @@ impl ConfigApi {
pub(crate) async fn write_value(
&self,
params: ConfigValueWriteParams,
) -> Result<ConfigWriteResponse, JSONRPCErrorError> {
if let Some(remote_plugin_edit) =
remote_plugin_enabled_config_edit(&params.key_path, &params.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<ConfigWriteResponse, JSONRPCErrorError> {
let pending_changes =
collect_plugin_enabled_candidates([(&params.key_path, &params.value)].into_iter());
@@ -168,6 +220,79 @@ impl ConfigApi {
pub(crate) async fn batch_write(
&self,
params: ConfigBatchWriteParams,
) -> Result<ConfigWriteResponse, JSONRPCErrorError> {
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::<ConfigEdit>::new();
let mut remote_plugin_toggles = BTreeMap::<String, bool>::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<ConfigWriteResponse, JSONRPCErrorError> {
let reload_user_config = params.reload_user_config;
let pending_changes = collect_plugin_enabled_candidates(
codex-rs/app-server/src/message_processor.rs
+17 -160
View File
@@ -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<OutgoingMessageSender>,
codex_message_processor: CodexMessageProcessor,
codex_message_processor: Arc<CodexMessageProcessor>,
thread_manager: Arc<ThreadManager>,
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<ConfigWriteResponse, JSONRPCErrorError> {
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<ConfigWriteResponse, JSONRPCErrorError> {
let ConfigBatchWriteParams {
edits,
file_path,
expected_version,
reload_user_config,
} = params;
let mut local_edits = Vec::<ConfigEdit>::new();
let mut remote_plugin_toggles = BTreeMap::<String, bool>::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;
codex-rs/core-plugins/src/remote.rs
+44
View File
@@ -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<RemotePluginEnabledConfigEdit> {
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>,