openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-07 15:40:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use direct remote plugin config sync

Browse Source
This commit is contained in:
xli-oai
2026-04-28 22:05:27 -07:00
parent 559cfce5a5
commit bdcf098809
5 changed files with 181 additions and 222 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
codex-rs/app-server/src/codex_message_processor/plugins.rs
View File

@@ -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<PluginUninstallResponse, JSONRPCErrorError> {
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,

165
codex-rs/app-server/src/config_write_router.rs
View File

@@ -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<dyn RemotePluginEnablementWriter>,
}
impl ConfigWriteRouter {
pub(crate) fn new(
config_api: ConfigApi,
remote_plugin_enablement_writer: Arc<dyn RemotePluginEnablementWriter>,
) -> Self {
Self {
config_api,
remote_plugin_enablement_writer,
}
}
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 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<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(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<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()
&& codex_core_plugins::remote::is_supported_remote_plugin_id(plugin_id)
{
return Some(RemotePluginEnabledConfigEdit {
plugin_id: plugin_id.to_string(),
enabled,
});
}
None
}

1
codex-rs/app-server/src/lib.rs
View File

@@ -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;

182
codex-rs/app-server/src/message_processor.rs
View File

@@ -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<OutgoingMessageSender>,
codex_message_processor: Arc<CodexMessageProcessor>,
codex_message_processor: CodexMessageProcessor,
thread_manager: Arc<ThreadManager>,
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<dyn RemotePluginEnablementWriter> =
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<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,
@@ -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;

15
codex-rs/core-plugins/src/remote.rs
View File

@@ -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>,