openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-05 20:10:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Emit external agent import progress telemetry

Browse Source
This commit is contained in:
charlesgong-openai
2026-06-12 19:07:10 -07:00
parent b32453ec46
commit b2bceb7904
5 changed files with 385 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/config/external_agent_config.rs
+253 -18
View File
@@ -1,13 +1,17 @@
use codex_config::types::PluginConfig;
use codex_core::config::Config;
use codex_core::config::ConfigBuilder;
use codex_core_plugins::PluginInstallError;
use codex_core_plugins::PluginInstallRequest;
use codex_core_plugins::PluginsManager;
use codex_core_plugins::marketplace::MarketplaceError;
use codex_core_plugins::marketplace::MarketplacePluginInstallPolicy;
use codex_core_plugins::marketplace::find_marketplace_manifest_path;
use codex_core_plugins::marketplace_add::MarketplaceAddError;
use codex_core_plugins::marketplace_add::MarketplaceAddRequest;
use codex_core_plugins::marketplace_add::add_marketplace;
use codex_core_plugins::marketplace_add::is_local_marketplace_source;
use codex_core_plugins::store::PluginStoreError;
use codex_external_agent_migration::build_mcp_config_from_external;
use codex_external_agent_migration::count_missing_commands;
use codex_external_agent_migration::count_missing_subagents;
@@ -34,6 +38,7 @@ use toml::Value as TomlValue;
const EXTERNAL_AGENT_CONFIG_DETECT_METRIC: &str = "codex.external_agent_config.detect";
const EXTERNAL_AGENT_CONFIG_IMPORT_METRIC: &str = "codex.external_agent_config.import";
const EXTERNAL_AGENT_CONFIG_IMPORT_ERROR_METRIC: &str = "codex.external_agent_config.import_error";
const EXTERNAL_AGENT_DIR: &str = ".claude";
const EXTERNAL_AGENT_CONFIG_MD: &str = "CLAUDE.md";
const EXTERNAL_OFFICIAL_MARKETPLACE_NAME: &str = "claude-plugins-official";
@@ -380,6 +385,17 @@ impl ExternalAgentConfigService {
if item_type != ExternalAgentConfigMigrationItemType::Plugins {
record_import_error(&mut item_result, "item_import", err.to_string(), None);
}
let cwd = cwd_for_log
.as_deref()
.map(|cwd| cwd.display().to_string())
.unwrap_or_else(|| "<home>".to_string());
tracing::warn!(
migration_type = migration_item_type_label(item_type),
description = %description,
cwd = %cwd,
error = %err,
"external agent config migration item failed"
);
outcome.item_results.push(item_result);
continue;
}
@@ -825,9 +841,16 @@ impl ExternalAgentConfigService {
details: Option<MigrationDetails>,
) -> io::Result<PluginImportOutcome> {
let Some(MigrationDetails { plugins, .. }) = details else {
return Err(invalid_data_error(
"plugins migration item is missing details".to_string(),
));
let err = invalid_data_error("plugins migration item is missing details".to_string());
emit_migration_error_metric(
ExternalAgentConfigMigrationItemType::Plugins,
"missing_details",
);
tracing::warn!(
error = %err,
"external agent plugin import failed"
);
return Err(err);
};
let mut outcome = PluginImportOutcome::default();
let plugins_manager = PluginsManager::new(self.codex_home.clone());
@@ -843,12 +866,46 @@ impl ExternalAgentConfigService {
|cwd| cwd.join(EXTERNAL_AGENT_DIR).join("settings.json"),
);
let source_root = cwd.unwrap_or(self.external_agent_home.as_path());
let import_source =
effective_external_settings(&source_settings)?.and_then(|settings| {
collect_marketplace_import_sources(&settings, source_root)
.remove(&marketplace_name)
});
let settings = match effective_external_settings(&source_settings) {
Ok(settings) => settings,
Err(err) => {
record_plugin_import_error(
&mut outcome,
cwd,
"settings_load",
err.to_string(),
Some(source_settings.display().to_string()),
);
tracing::warn!(
marketplace = %marketplace_name,
source_settings = %source_settings.display(),
error = %err,
"failed to load external agent marketplace source settings during plugin import"
);
outcome.failed_marketplaces.push(marketplace_name);
outcome.failed_plugin_ids.extend(plugin_ids);
continue;
}
};
let import_source = settings.and_then(|settings| {
collect_marketplace_import_sources(&settings, source_root).remove(&marketplace_name)
});
let Some(import_source) = import_source else {
record_plugin_import_error(
&mut outcome,
cwd,
"marketplace_source_missing",
format!(
"external agent plugin import source was not found for marketplace `{marketplace_name}`"
),
Some(source_settings.display().to_string()),
);
tracing::warn!(
marketplace = %marketplace_name,
source_settings = %source_settings.display(),
plugin_ids = ?plugin_ids,
"external agent plugin import source was not found"
);
outcome.failed_marketplaces.push(marketplace_name);
outcome.failed_plugin_ids.extend(plugin_ids);
continue;
@@ -861,9 +918,23 @@ impl ExternalAgentConfigService {
let add_marketplace_outcome = add_marketplace(self.codex_home.clone(), request).await;
let marketplace_path = match add_marketplace_outcome {
Ok(add_marketplace_outcome) => {
let Some(marketplace_path) = find_marketplace_manifest_path(
add_marketplace_outcome.installed_root.as_path(),
) else {
let installed_root = add_marketplace_outcome.installed_root;
let Some(marketplace_path) =
find_marketplace_manifest_path(installed_root.as_path())
else {
record_plugin_import_error(
&mut outcome,
cwd,
"marketplace_manifest_missing",
"external agent marketplace import did not contain a supported marketplace manifest",
Some(installed_root.as_path().display().to_string()),
);
tracing::warn!(
marketplace = %marketplace_name,
installed_root = %installed_root.as_path().display(),
plugin_ids = ?plugin_ids,
"external agent marketplace import did not contain a supported marketplace manifest"
);
outcome.failed_marketplaces.push(marketplace_name);
outcome.failed_plugin_ids.extend(plugin_ids);
continue;
@@ -873,13 +944,28 @@ impl ExternalAgentConfigService {
.push(marketplace_name.clone());
marketplace_path
}
Err(_) => {
Err(err) => {
let problem = marketplace_add_failure_stage(&err);
record_plugin_import_error(
&mut outcome,
cwd,
problem,
err.to_string(),
Some(marketplace_name.clone()),
);
tracing::warn!(
marketplace = %marketplace_name,
plugin_ids = ?plugin_ids,
error = %err,
"failed to import external agent plugin marketplace"
);
outcome.failed_marketplaces.push(marketplace_name);
outcome.failed_plugin_ids.extend(plugin_ids);
continue;
}
};
for plugin_name in plugin_names {
let plugin_id = format!("{plugin_name}@{marketplace_name}");
match plugins_manager
.install_plugin(PluginInstallRequest {
plugin_name: plugin_name.clone(),
@@ -887,12 +973,26 @@ impl ExternalAgentConfigService {
})
.await
{
Ok(_) => outcome
.succeeded_plugin_ids
.push(format!("{plugin_name}@{marketplace_name}")),
Err(_) => outcome
.failed_plugin_ids
.push(format!("{plugin_name}@{marketplace_name}")),
Ok(_) => outcome.succeeded_plugin_ids.push(plugin_id),
Err(err) => {
let problem = plugin_install_failure_stage(&err);
record_plugin_import_error(
&mut outcome,
cwd,
problem,
err.to_string(),
Some(plugin_id.clone()),
);
tracing::warn!(
marketplace = %marketplace_name,
plugin = %plugin_name,
marketplace_path = %marketplace_path.as_path().display(),
problem,
error = %err,
"failed to install external agent plugin"
);
outcome.failed_plugin_ids.push(plugin_id);
}
}
}
}
@@ -1767,6 +1867,7 @@ pub(crate) fn record_import_error(
message: impl Into<String>,
source: Option<String>,
) {
emit_migration_error_metric(result.item_type, failure_stage);
result.record_error(ExternalAgentConfigImportRawError {
item_type: result.item_type,
failure_stage: failure_stage.to_string(),
@@ -1776,6 +1877,23 @@ pub(crate) fn record_import_error(
});
}
fn record_plugin_import_error(
outcome: &mut PluginImportOutcome,
cwd: Option<&Path>,
failure_stage: &'static str,
message: impl Into<String>,
source: Option<String>,
) {
emit_migration_error_metric(ExternalAgentConfigMigrationItemType::Plugins, failure_stage);
outcome.raw_errors.push(ExternalAgentConfigImportRawError {
item_type: ExternalAgentConfigMigrationItemType::Plugins,
failure_stage: failure_stage.to_string(),
message: message.into(),
cwd: cwd.map(Path::to_path_buf),
source,
});
}
fn migration_metric_tags(
item_type: ExternalAgentConfigMigrationItemType,
skills_count: Option<usize>,
@@ -1795,6 +1913,123 @@ fn migration_metric_tags(
tags
}
fn plugin_install_failure_stage(error: &PluginInstallError) -> &'static str {
match error {
PluginInstallError::Marketplace(MarketplaceError::MarketplaceNotFound { .. }) => {
"marketplace_not_found"
}
PluginInstallError::Marketplace(MarketplaceError::InvalidMarketplaceFile { .. }) => {
"marketplace_invalid_file"
}
PluginInstallError::Marketplace(MarketplaceError::Io { .. }) => "marketplace_io",
PluginInstallError::Marketplace(MarketplaceError::PluginNotFound { .. }) => {
"plugin_not_found"
}
PluginInstallError::Marketplace(MarketplaceError::PluginNotAvailable { .. }) => {
"plugin_not_available"
}
PluginInstallError::Marketplace(MarketplaceError::PluginsDisabled) => "plugins_disabled",
PluginInstallError::Marketplace(MarketplaceError::InvalidPlugin(message)) => {
plugin_invalid_failure_stage(message)
}
PluginInstallError::Remote(_) => "plugin_remote_sync",
PluginInstallError::Store(PluginStoreError::Io { .. }) => "plugin_store_io",
PluginInstallError::Store(PluginStoreError::Invalid(message)) => {
plugin_store_invalid_failure_stage(message)
}
PluginInstallError::Config(_) => "plugin_config",
PluginInstallError::Join(_) => "plugin_install_join",
}
}
fn marketplace_add_failure_stage(error: &MarketplaceAddError) -> &'static str {
match error {
MarketplaceAddError::InvalidRequest(message) => {
if message.contains("path does not exist or is not a directory")
|| message.contains("failed to resolve local marketplace source path")
{
"marketplace_source_missing"
} else if message.contains("invalid marketplace")
|| message.contains("does not expose")
|| message.contains("marketplace file")
{
"marketplace_invalid"
} else {
"marketplace_add_invalid_request"
}
}
MarketplaceAddError::Internal(message) => {
if message.contains(" git ") || message.contains("git clone") {
"git_materialization"
} else {
"marketplace_add_internal"
}
}
}
}
fn plugin_invalid_failure_stage(message: &str) -> &'static str {
if message == "missing or invalid plugin.json" {
"plugin_json_missing_or_invalid"
} else if message == "path does not exist or is not a directory" {
"plugin_source_not_directory"
} else if message.starts_with("failed to materialize plugin source") {
"git_materialization"
} else if message.starts_with("invalid plugin name") {
"plugin_json_invalid_name"
} else {
"plugin_invalid"
}
}
fn plugin_store_invalid_failure_stage(message: &str) -> &'static str {
if message == "missing plugin.json" {
"plugin_json_missing"
} else if message == "missing or invalid plugin.json" {
"plugin_json_missing_or_invalid"
} else if message.starts_with("failed to parse plugin.json") {
"plugin_json_parse"
} else if message.starts_with("invalid plugin version in plugin.json") {
"plugin_json_invalid_version"
} else if message.starts_with("plugin.json name `") {
"plugin_json_name_mismatch"
} else if message.starts_with("invalid plugin name") {
"plugin_json_invalid_name"
} else if message.starts_with("invalid plugin version") {
"plugin_json_invalid_version"
} else if message.starts_with("plugin source path is not a directory") {
"plugin_source_not_directory"
} else if message.starts_with("failed to materialize plugin source") {
"git_materialization"
} else if message == "path does not exist or is not a directory" {
"plugin_source_not_directory"
} else {
"plugin_store_invalid"
}
}
pub(crate) fn emit_migration_error_metric(
item_type: ExternalAgentConfigMigrationItemType,
failure_stage: &'static str,
) {
let Some(metrics) = codex_otel::global() else {
return;
};
let tags = [
("migration_type", migration_item_type_label(item_type)),
("failure_stage", failure_stage),
];
let tag_refs = tags
.iter()
.map(|(key, value)| (*key, *value))
.collect::<Vec<_>>();
let _ = metrics.counter(
EXTERNAL_AGENT_CONFIG_IMPORT_ERROR_METRIC,
/*inc*/ 1,
&tag_refs,
);
}
fn emit_migration_metric(
metric_name: &str,
item_type: ExternalAgentConfigMigrationItemType,
codex-rs/app-server/src/config/external_agent_config_tests.rs
+33
View File
@@ -1632,6 +1632,28 @@ fn migration_metric_tags_for_skills_include_skills_count() {
);
}
#[test]
fn plugin_install_failure_stage_reports_plugin_manifest_problems() {
assert_eq!(
plugin_install_failure_stage(&PluginInstallError::Store(PluginStoreError::Invalid(
"missing plugin.json".to_string()
))),
"plugin_json_missing"
);
assert_eq!(
plugin_install_failure_stage(&PluginInstallError::Store(PluginStoreError::Invalid(
"plugin.json name `wrong` does not match marketplace plugin name `right`".to_string()
))),
"plugin_json_name_mismatch"
);
assert_eq!(
plugin_install_failure_stage(&PluginInstallError::Store(PluginStoreError::Invalid(
"failed to parse plugin.json: expected ident".to_string()
))),
"plugin_json_parse"
);
}
#[tokio::test]
async fn detect_home_lists_enabled_plugins_from_settings() {
let (_root, external_agent_home, codex_home) = fixture_paths();
@@ -2190,6 +2212,11 @@ async fn import_plugins_requires_source_marketplace_details() {
outcome.failed_plugin_ids,
vec!["formatter@other-tools".to_string()]
);
assert_eq!(outcome.raw_errors.len(), 1);
assert_eq!(
outcome.raw_errors[0].failure_stage,
"marketplace_source_missing"
);
}
#[tokio::test]
@@ -2224,6 +2251,11 @@ async fn import_plugins_defers_marketplace_source_validation_to_add_marketplace(
outcome.failed_plugin_ids,
vec!["formatter@acme-tools".to_string()]
);
assert_eq!(outcome.raw_errors.len(), 1);
assert_eq!(
outcome.raw_errors[0].failure_stage,
"marketplace_source_missing"
);
}
#[tokio::test]
@@ -2539,6 +2571,7 @@ async fn import_plugins_infers_external_official_marketplace_when_missing_from_s
outcome.failed_plugin_ids,
vec![format!("sample@{EXTERNAL_OFFICIAL_MARKETPLACE_NAME}")]
);
assert_eq!(outcome.raw_errors.len(), 1);
}
#[tokio::test]
codex-rs/app-server/src/request_processors/external_agent_config_processor.rs
+68 -1
View File
@@ -19,7 +19,9 @@ use codex_app_server_protocol::CommandMigration;
use codex_app_server_protocol::ExternalAgentConfigDetectParams;
use codex_app_server_protocol::ExternalAgentConfigDetectResponse;
use codex_app_server_protocol::ExternalAgentConfigImportCompletedNotification;
use codex_app_server_protocol::ExternalAgentConfigImportItemResult as ProtocolImportItemResult;
use codex_app_server_protocol::ExternalAgentConfigImportParams;
use codex_app_server_protocol::ExternalAgentConfigImportProgressNotification;
use codex_app_server_protocol::ExternalAgentConfigImportRawError as ProtocolImportRawError;
use codex_app_server_protocol::ExternalAgentConfigImportResponse;
use codex_app_server_protocol::ExternalAgentConfigImportTypeResult as ProtocolImportTypeResult;
@@ -187,7 +189,15 @@ impl ExternalAgentConfigRequestProcessor {
});
let (pending_session_imports, session_validation_result) =
self.validate_pending_session_imports(&params);
let import_outcome = self.import_external_agent_config(params).await?;
let import_outcome = self
.import_external_agent_config(params)
.await
.inspect_err(|error| {
tracing::warn!(
error = %error.message.as_str(),
"external agent config import failed"
);
})?;
if needs_runtime_refresh {
self.config_processor.handle_config_mutation().await;
}
@@ -206,9 +216,11 @@ impl ExternalAgentConfigRequestProcessor {
let mut completed_item_results = Vec::new();
if let Some(session_validation_result) = session_validation_result {
send_import_progress(&self.outgoing, &import_id, &session_validation_result).await;
completed_item_results.push(session_validation_result);
}
for item_result in import_outcome.item_results {
send_import_progress(&self.outgoing, &import_id, &item_result).await;
completed_item_results.push(item_result);
}
@@ -236,13 +248,19 @@ impl ExternalAgentConfigRequestProcessor {
});
let pending_plugin_imports = import_outcome.pending_plugin_imports;
tokio::spawn(async move {
let session_progress_outgoing = Arc::clone(&outgoing);
let session_import_id = import_id.clone();
let session_imports = async move {
let session_import_result = session_import_result?;
let item_result = session_importer
.import_sessions(pending_session_imports, session_import_result)
.await;
send_import_progress(&session_progress_outgoing, &session_import_id, &item_result)
.await;
Some(item_result)
};
let plugin_progress_outgoing = Arc::clone(&outgoing);
let plugin_import_id = import_id.clone();
let plugin_imports = async move {
let mut item_results = Vec::new();
for pending_plugin_import in pending_plugin_imports {
@@ -265,8 +283,18 @@ impl ExternalAgentConfigRequestProcessor {
error.message.clone(),
None,
);
tracing::warn!(
error = %error.message,
"external agent config plugin import failed"
);
}
}
send_import_progress(
&plugin_progress_outgoing,
&plugin_import_id,
&item_result,
)
.await;
item_results.push(item_result);
}
item_results
@@ -337,6 +365,10 @@ impl ExternalAgentConfigRequestProcessor {
),
Some(session.path.display().to_string()),
);
tracing::warn!(
path = %session.path.display(),
"skipping external agent session import because the session was not detected"
);
continue;
}
Err(err) => {
@@ -346,6 +378,11 @@ impl ExternalAgentConfigRequestProcessor {
err.to_string(),
Some(session.path.display().to_string()),
);
tracing::warn!(
path = %session.path.display(),
error = %err,
"skipping external agent session import because the source path could not be validated"
);
continue;
}
};
@@ -471,6 +508,21 @@ impl ExternalAgentConfigRequestProcessor {
}
}
async fn send_import_progress(
outgoing: &OutgoingMessageSender,
import_id: &str,
item_result: &CoreImportItemResult,
) {
outgoing
.send_server_notification(ServerNotification::ExternalAgentConfigImportProgress(
ExternalAgentConfigImportProgressNotification {
import_id: import_id.to_string(),
item_result: protocol_import_item_result(item_result),
},
))
.await;
}
fn completed_notification(
import_id: String,
item_results: &[CoreImportItemResult],
@@ -521,6 +573,21 @@ fn completed_notification(
}
}
fn protocol_import_item_result(item_result: &CoreImportItemResult) -> ProtocolImportItemResult {
ProtocolImportItemResult {
item_type: protocol_migration_item_type(item_result.item_type),
description: item_result.description.clone(),
cwd: item_result.cwd.clone(),
success_count: item_result.success_count,
error_count: item_result.error_count,
raw_errors: item_result
.raw_errors
.iter()
.map(protocol_import_raw_error)
.collect(),
}
}
fn protocol_import_raw_error(raw_error: &CoreImportRawError) -> ProtocolImportRawError {
ProtocolImportRawError {
item_type: protocol_migration_item_type(raw_error.item_type),
codex-rs/app-server/src/request_processors/external_agent_session_import.rs
+10
View File
@@ -75,6 +75,7 @@ impl ExternalAgentSessionImporter {
"external agent session import permit could not be acquired",
None,
);
tracing::warn!("external agent session import permit could not be acquired");
return item_result;
};
let import_results = futures::stream::iter(sessions)
@@ -100,6 +101,11 @@ impl ExternalAgentSessionImporter {
failure.message.clone(),
Some(failure.source_path.display().to_string()),
);
tracing::warn!(
error = %failure.message,
path = %failure.source_path.display(),
"external agent session import failed"
);
}
}
}
@@ -110,6 +116,10 @@ impl ExternalAgentSessionImporter {
err.to_string(),
None,
);
tracing::warn!(
error = %err,
"external agent session import ledger update failed"
);
}
item_result
}
codex-rs/app-server/tests/suite/v2/external_agent_config.rs
+21 -4
View File
@@ -7,6 +7,7 @@ use app_test_support::to_response;
use app_test_support::write_mock_responses_config_toml;
use codex_app_server_protocol::ExternalAgentConfigDetectResponse;
use codex_app_server_protocol::ExternalAgentConfigImportCompletedNotification;
use codex_app_server_protocol::ExternalAgentConfigImportProgressNotification;
use codex_app_server_protocol::ExternalAgentConfigImportResponse;
use codex_app_server_protocol::ExternalAgentConfigImportTypeResult;
use codex_app_server_protocol::ExternalAgentConfigMigrationItemType;
@@ -936,6 +937,25 @@ async fn external_agent_config_import_skips_undetected_session_paths() -> Result
.await??;
let response: ExternalAgentConfigImportResponse = to_response(response)?;
let import_id = assert_import_response(response);
let progress = timeout(
DEFAULT_TIMEOUT,
mcp.read_stream_until_notification_message("externalAgentConfig/import/progress"),
)
.await??;
let progress: ExternalAgentConfigImportProgressNotification =
serde_json::from_value(progress.params.expect("progress params"))?;
assert_eq!(progress.import_id, import_id);
assert_eq!(
progress.item_result.item_type,
ExternalAgentConfigMigrationItemType::Sessions
);
assert_eq!(progress.item_result.success_count, 0);
assert_eq!(progress.item_result.error_count, 1);
assert_eq!(progress.item_result.raw_errors.len(), 1);
assert_eq!(
progress.item_result.raw_errors[0].failure_stage,
"session_missing"
);
let notification = timeout(
DEFAULT_TIMEOUT,
mcp.read_stream_until_notification_message("externalAgentConfig/import/completed"),
@@ -945,18 +965,15 @@ async fn external_agent_config_import_skips_undetected_session_paths() -> Result
let completed: ExternalAgentConfigImportCompletedNotification =
serde_json::from_value(notification.params.expect("completed params"))?;
assert_eq!(completed.import_id, import_id);
let raw_errors = completed.item_results[0].raw_errors.clone();
assert_eq!(
completed.item_results,
vec![ExternalAgentConfigImportTypeResult {
item_type: ExternalAgentConfigMigrationItemType::Sessions,
success_count: 1,
error_count: 1,
raw_errors: raw_errors.clone(),
raw_errors: progress.item_result.raw_errors,
}]
);
assert_eq!(raw_errors.len(), 1);
assert_eq!(raw_errors[0].failure_stage, "session_missing");
let request_id = mcp
.send_thread_list_request(ThreadListParams {