agent/codex
2
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

[codex] Record external agent import results (#28396)

Browse Source 
## Summary
- restore `externalAgentConfig/import/progress` notifications while
keeping `externalAgentConfig/import/completed` as the must-deliver event
- persist completed external-agent config imports in state DB by
`importId`, including concrete success/failure details for config,
AGENTS.md, skills, plugins, MCP servers, subagents, hooks, commands, and
sessions
- add `externalAgentConfig/import/readHistories` so clients can recover
persisted import results after missing the live completion notification
- include `errorType` on import failures in protocol
responses/notifications and persisted DB JSON so future code can
classify failures without another wire/storage shape change

## Validation
- `git diff --check`
- `just test -p codex-state external_agent_config_imports`
- `just test -p codex-app-server-protocol`
- `CODEX_SQLITE_HOME=/private/tmp/codex-app-server-sqlite-read-details
just test -p codex-app-server
external_agent_config_import_sends_completion_notification_for_sync_only_import`

Also ran earlier broader checks before publishing:
- `just test -p codex-state`
-
`CODEX_SQLITE_HOME=/private/tmp/codex-app-server-external-agent-test-sqlite
just test -p codex-app-server external_agent_config`
- `just test -p codex-external-agent-migration`
This commit is contained in:
charlesgong-openai
2026-06-15 23:17:24 -07:00
committed by GitHub
Unverified
parent 1e015884c5
commit 314fa3d25b
30 changed files with 1370 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/config/external_agent_config.rs
+85 -58
View File
@@ -129,11 +129,6 @@ impl ExternalAgentConfigImportItemResult {
}
}
pub(crate) fn record_successes(&mut self, count: usize) {
let count = u32::try_from(count).unwrap_or(u32::MAX);
self.success_count = self.success_count.saturating_add(count);
}
pub(crate) fn record_error(&mut self, raw_error: ExternalAgentConfigImportRawError) {
self.error_count = self.error_count.saturating_add(1);
self.raw_errors.push(raw_error);
@@ -161,6 +156,7 @@ pub(crate) struct ExternalAgentConfigImportSuccess {
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct ExternalAgentConfigImportRawError {
pub item_type: ExternalAgentConfigMigrationItemType,
pub error_type: Option<String>,
pub failure_stage: String,
pub message: String,
pub cwd: Option<PathBuf>,
@@ -254,33 +250,41 @@ impl ExternalAgentConfigService {
);
let import_result = match migration_item.item_type {
ExternalAgentConfigMigrationItemType::Config => (|| {
let migrated_count = self.import_config(migration_item.cwd.as_deref())?;
if let Some((source, target)) =
self.import_config(migration_item.cwd.as_deref())?
{
item_result.record_success(Some(source), Some(target));
}
emit_migration_metric(
EXTERNAL_AGENT_CONFIG_IMPORT_METRIC,
ExternalAgentConfigMigrationItemType::Config,
/*skills_count*/ None,
);
item_result.record_successes(migrated_count);
Ok(())
})(),
ExternalAgentConfigMigrationItemType::Skills => (|| {
let skills_count = self.import_skills(migration_item.cwd.as_deref())?;
let imported_skills = self.import_skills(migration_item.cwd.as_deref())?;
emit_migration_metric(
EXTERNAL_AGENT_CONFIG_IMPORT_METRIC,
ExternalAgentConfigMigrationItemType::Skills,
Some(skills_count),
Some(imported_skills.len()),
);
item_result.record_successes(skills_count);
for skill_name in imported_skills {
item_result.record_success(Some(skill_name.clone()), Some(skill_name));
}
Ok(())
})(),
ExternalAgentConfigMigrationItemType::AgentsMd => (|| {
let migrated_count = self.import_agents_md(migration_item.cwd.as_deref())?;
if let Some((source, target)) =
self.import_agents_md(migration_item.cwd.as_deref())?
{
item_result.record_success(Some(source), Some(target));
}
emit_migration_metric(
EXTERNAL_AGENT_CONFIG_IMPORT_METRIC,
ExternalAgentConfigMigrationItemType::AgentsMd,
/*skills_count*/ None,
);
item_result.record_successes(migrated_count);
Ok(())
})(),
ExternalAgentConfigMigrationItemType::Plugins => {
@@ -357,44 +361,54 @@ impl ExternalAgentConfigService {
.await
}
ExternalAgentConfigMigrationItemType::McpServerConfig => (|| {
let migrated_count =
let migrated_server_names =
self.import_mcp_server_config(migration_item.cwd.as_deref())?;
emit_migration_metric(
EXTERNAL_AGENT_CONFIG_IMPORT_METRIC,
ExternalAgentConfigMigrationItemType::McpServerConfig,
/*skills_count*/ None,
);
item_result.record_successes(migrated_count);
for server_name in migrated_server_names {
item_result.record_success(Some(server_name.clone()), Some(server_name));
}
Ok(())
})(),
ExternalAgentConfigMigrationItemType::Subagents => (|| {
let subagents_count = self.import_subagents(migration_item.cwd.as_deref())?;
let imported_subagents =
self.import_subagents(migration_item.cwd.as_deref())?;
emit_migration_metric(
EXTERNAL_AGENT_CONFIG_IMPORT_METRIC,
ExternalAgentConfigMigrationItemType::Subagents,
Some(subagents_count),
Some(imported_subagents.len()),
);
item_result.record_successes(subagents_count);
for subagent_name in imported_subagents {
item_result
.record_success(Some(subagent_name.clone()), Some(subagent_name));
}
Ok(())
})(),
ExternalAgentConfigMigrationItemType::Hooks => (|| {
let migrated_count = self.import_hooks(migration_item.cwd.as_deref())?;
let migrated_hook_names = self.import_hooks(migration_item.cwd.as_deref())?;
emit_migration_metric(
EXTERNAL_AGENT_CONFIG_IMPORT_METRIC,
ExternalAgentConfigMigrationItemType::Hooks,
/*skills_count*/ None,
);
item_result.record_successes(migrated_count);
for hook_name in migrated_hook_names {
item_result.record_success(Some(hook_name.clone()), Some(hook_name));
}
Ok(())
})(),
ExternalAgentConfigMigrationItemType::Commands => (|| {
let commands_count = self.import_commands(migration_item.cwd.as_deref())?;
let imported_commands = self.import_commands(migration_item.cwd.as_deref())?;
emit_migration_metric(
EXTERNAL_AGENT_CONFIG_IMPORT_METRIC,
ExternalAgentConfigMigrationItemType::Commands,
Some(commands_count),
Some(imported_commands.len()),
);
item_result.record_successes(commands_count);
for command_name in imported_commands {
item_result.record_success(Some(command_name.clone()), Some(command_name));
}
Ok(())
})(),
ExternalAgentConfigMigrationItemType::Sessions => Ok(()),
@@ -957,7 +971,7 @@ impl ExternalAgentConfigService {
Ok(outcome)
}
fn import_config(&self, cwd: Option<&Path>) -> io::Result<usize> {
fn import_config(&self, cwd: Option<&Path>) -> io::Result<Option<(String, String)>> {
let repo_root = find_repo_root(cwd)?;
let (source_settings, target_config) = if let Some(repo_root) = repo_root.as_ref() {
(
@@ -965,7 +979,7 @@ impl ExternalAgentConfigService {
repo_root.join(".codex").join("config.toml"),
)
} else if cwd.is_some_and(|cwd| !cwd.as_os_str().is_empty()) {
return Ok(0);
return Ok(None);
} else {
(
self.external_agent_home.join("settings.json"),
@@ -973,11 +987,11 @@ impl ExternalAgentConfigService {
)
};
let Some(settings) = effective_external_settings(&source_settings)? else {
return Ok(0);
return Ok(None);
};
let migrated = build_config_from_external(&settings)?;
if is_empty_toml_table(&migrated) {
return Ok(0);
return Ok(None);
}
let Some(target_parent) = target_config.parent() else {
@@ -986,7 +1000,10 @@ impl ExternalAgentConfigService {
fs::create_dir_all(target_parent)?;
if !target_config.exists() {
write_toml_file(&target_config, &migrated)?;
return Ok(1);
return Ok(Some((
source_settings.display().to_string(),
target_config.display().to_string(),
)));
}
let existing_raw = fs::read_to_string(&target_config)?;
@@ -999,14 +1016,17 @@ impl ExternalAgentConfigService {
let changed = merge_missing_toml_values(&mut existing, &migrated)?;
if !changed {
return Ok(0);
return Ok(None);
}
write_toml_file(&target_config, &existing)?;
Ok(1)
Ok(Some((
source_settings.display().to_string(),
target_config.display().to_string(),
)))
}
fn import_mcp_server_config(&self, cwd: Option<&Path>) -> io::Result<usize> {
fn import_mcp_server_config(&self, cwd: Option<&Path>) -> io::Result<Vec<String>> {
let repo_root = find_repo_root(cwd)?;
let (source_settings, target_config) = if let Some(repo_root) = repo_root.as_ref() {
(
@@ -1014,7 +1034,7 @@ impl ExternalAgentConfigService {
repo_root.join(".codex").join("config.toml"),
)
} else if cwd.is_some_and(|cwd| !cwd.as_os_str().is_empty()) {
return Ok(0);
return Ok(Vec::new());
} else {
(
self.external_agent_home.join("settings.json"),
@@ -1031,7 +1051,7 @@ impl ExternalAgentConfigService {
settings.as_ref(),
)?;
if is_empty_toml_table(&migrated) {
return Ok(0);
return Ok(Vec::new());
}
let Some(target_parent) = target_config.parent() else {
@@ -1039,9 +1059,9 @@ impl ExternalAgentConfigService {
};
fs::create_dir_all(target_parent)?;
if !target_config.exists() {
let migrated_count = migrated_mcp_server_names(&migrated).len();
let migrated_server_names = migrated_mcp_server_names(&migrated);
write_toml_file(&target_config, &migrated)?;
return Ok(migrated_count);
return Ok(migrated_server_names);
}
let existing_raw = fs::read_to_string(&target_config)?;
@@ -1051,21 +1071,21 @@ impl ExternalAgentConfigService {
toml::from_str::<TomlValue>(&existing_raw)
.map_err(|err| invalid_data_error(format!("invalid existing config.toml: {err}")))?
};
let merged_server_count = merge_missing_mcp_servers(&mut existing, &migrated)?.len();
if merged_server_count > 0 {
let merged_server_names = merge_missing_mcp_servers(&mut existing, &migrated)?;
if !merged_server_names.is_empty() {
write_toml_file(&target_config, &existing)?;
}
Ok(merged_server_count)
Ok(merged_server_names)
}
fn import_subagents(&self, cwd: Option<&Path>) -> io::Result<usize> {
fn import_subagents(&self, cwd: Option<&Path>) -> io::Result<Vec<String>> {
let (source_agents, target_agents) = if let Some(repo_root) = find_repo_root(cwd)? {
(
repo_root.join(EXTERNAL_AGENT_DIR).join("agents"),
repo_root.join(".codex").join("agents"),
)
} else if cwd.is_some_and(|cwd| !cwd.as_os_str().is_empty()) {
return Ok(0);
return Ok(Vec::new());
} else {
(
self.external_agent_home.join("agents"),
@@ -1076,7 +1096,7 @@ impl ExternalAgentConfigService {
import_subagents(&source_agents, &target_agents)
}
fn import_hooks(&self, cwd: Option<&Path>) -> io::Result<usize> {
fn import_hooks(&self, cwd: Option<&Path>) -> io::Result<Vec<String>> {
let (source_external_agent_dir, target_hooks) =
if let Some(repo_root) = find_repo_root(cwd)? {
(
@@ -1084,7 +1104,7 @@ impl ExternalAgentConfigService {
repo_root.join(".codex").join("hooks.json"),
)
} else if cwd.is_some_and(|cwd| !cwd.as_os_str().is_empty()) {
return Ok(0);
return Ok(Vec::new());
} else {
(
self.external_agent_home.clone(),
@@ -1092,20 +1112,22 @@ impl ExternalAgentConfigService {
)
};
Ok(usize::from(import_hooks(
&source_external_agent_dir,
&target_hooks,
)?))
let hook_names = hook_migration_event_names(&source_external_agent_dir, &target_hooks)?;
if import_hooks(&source_external_agent_dir, &target_hooks)? {
Ok(hook_names)
} else {
Ok(Vec::new())
}
}
fn import_commands(&self, cwd: Option<&Path>) -> io::Result<usize> {
fn import_commands(&self, cwd: Option<&Path>) -> io::Result<Vec<String>> {
let (source_commands, target_skills) = if let Some(repo_root) = find_repo_root(cwd)? {
(
repo_root.join(EXTERNAL_AGENT_DIR).join("commands"),
repo_root.join(".agents").join("skills"),
)
} else if cwd.is_some_and(|cwd| !cwd.as_os_str().is_empty()) {
return Ok(0);
return Ok(Vec::new());
} else {
(
self.external_agent_home.join("commands"),
@@ -1116,14 +1138,14 @@ impl ExternalAgentConfigService {
import_commands(&source_commands, &target_skills)
}
fn import_skills(&self, cwd: Option<&Path>) -> io::Result<usize> {
fn import_skills(&self, cwd: Option<&Path>) -> io::Result<Vec<String>> {
let (source_skills, target_skills) = if let Some(repo_root) = find_repo_root(cwd)? {
(
repo_root.join(EXTERNAL_AGENT_DIR).join("skills"),
repo_root.join(".agents").join("skills"),
)
} else if cwd.is_some_and(|cwd| !cwd.as_os_str().is_empty()) {
return Ok(0);
return Ok(Vec::new());
} else {
(
self.external_agent_home.join("skills"),
@@ -1131,11 +1153,11 @@ impl ExternalAgentConfigService {
)
};
if !source_skills.is_dir() {
return Ok(0);
return Ok(Vec::new());
}
fs::create_dir_all(&target_skills)?;
let mut copied_count = 0usize;
let mut copied_names = Vec::new();
for entry in fs::read_dir(&source_skills)? {
let entry = entry?;
@@ -1150,20 +1172,20 @@ impl ExternalAgentConfigService {
}
copy_dir_recursive(&entry.path(), &target)?;
copied_count += 1;
copied_names.push(entry.file_name().to_string_lossy().to_string());
}
Ok(copied_count)
Ok(copied_names)
}
fn import_agents_md(&self, cwd: Option<&Path>) -> io::Result<usize> {
fn import_agents_md(&self, cwd: Option<&Path>) -> io::Result<Option<(String, String)>> {
let (source_agents_md, target_agents_md) = if let Some(repo_root) = find_repo_root(cwd)? {
let Some(source_agents_md) = find_repo_agents_md_source(&repo_root)? else {
return Ok(0);
return Ok(None);
};
(source_agents_md, repo_root.join("AGENTS.md"))
} else if cwd.is_some_and(|cwd| !cwd.as_os_str().is_empty()) {
return Ok(0);
return Ok(None);
} else {
(
self.external_agent_home.join(EXTERNAL_AGENT_CONFIG_MD),
@@ -1173,7 +1195,7 @@ impl ExternalAgentConfigService {
if !is_non_empty_text_file(&source_agents_md)?
|| !is_missing_or_empty_text_file(&target_agents_md)?
{
return Ok(0);
return Ok(None);
}
let Some(target_parent) = target_agents_md.parent() else {
@@ -1182,7 +1204,10 @@ impl ExternalAgentConfigService {
fs::create_dir_all(target_parent)?;
rewrite_and_copy_text_file(&source_agents_md, &target_agents_md)?;
Ok(1)
Ok(Some((
source_agents_md.display().to_string(),
target_agents_md.display().to_string(),
)))
}
}
@@ -1826,6 +1851,7 @@ pub(crate) fn record_import_error(
) {
result.record_error(ExternalAgentConfigImportRawError {
item_type: result.item_type,
error_type: None,
failure_stage: failure_stage.to_string(),
message: message.into(),
cwd: result.cwd.clone(),
@@ -1856,6 +1882,7 @@ fn plugin_import_raw_error(
) -> ExternalAgentConfigImportRawError {
ExternalAgentConfigImportRawError {
item_type: ExternalAgentConfigMigrationItemType::Plugins,
error_type: None,
failure_stage: failure_stage.to_string(),
message,
cwd: cwd.map(Path::to_path_buf),
codex-rs/app-server/src/config/external_agent_config_tests.rs
+48 -7
View File
@@ -47,11 +47,26 @@ fn assert_single_plugin_raw_error(
ExternalAgentConfigMigrationItemType::Plugins
);
assert_eq!(raw_error.failure_stage, failure_stage);
assert_eq!(raw_error.error_type, None);
assert_eq!(raw_error.cwd, None);
assert_eq!(raw_error.source.as_deref(), Some(source));
assert!(!raw_error.message.is_empty());
}
fn import_success(
item_type: ExternalAgentConfigMigrationItemType,
cwd: Option<PathBuf>,
source: impl Into<String>,
target: impl Into<String>,
) -> ExternalAgentConfigImportSuccess {
ExternalAgentConfigImportSuccess {
item_type,
cwd,
source: Some(source.into()),
target: Some(target.into()),
}
}
#[tokio::test]
async fn detect_home_lists_config_skills_and_agents_md() {
let (_root, external_agent_home, codex_home) = fixture_paths();
@@ -1232,7 +1247,15 @@ async fn import_repo_agents_md_rewrites_terms_and_skips_non_empty_targets() {
cwd: Some(repo_root.clone()),
success_count: 1,
error_count: 0,
successes: Vec::new(),
successes: vec![import_success(
ExternalAgentConfigMigrationItemType::AgentsMd,
Some(repo_root.clone()),
repo_root
.join(EXTERNAL_AGENT_CONFIG_MD)
.display()
.to_string(),
repo_root.join("AGENTS.md").display().to_string(),
)],
raw_errors: Vec::new(),
},
ExternalAgentConfigImportItemResult {
@@ -1290,7 +1313,15 @@ async fn import_repo_agents_md_overwrites_empty_targets() {
cwd: Some(repo_root.clone()),
success_count: 1,
error_count: 0,
successes: Vec::new(),
successes: vec![import_success(
ExternalAgentConfigMigrationItemType::AgentsMd,
Some(repo_root.clone()),
repo_root
.join(EXTERNAL_AGENT_CONFIG_MD)
.display()
.to_string(),
repo_root.join("AGENTS.md").display().to_string(),
)],
raw_errors: Vec::new(),
}]
);
@@ -1383,7 +1414,12 @@ async fn import_repo_hooks_preserves_disabled_codex_hooks_feature() {
cwd: Some(repo_root.clone()),
success_count: 1,
error_count: 0,
successes: Vec::new(),
successes: vec![import_success(
ExternalAgentConfigMigrationItemType::Hooks,
Some(repo_root.clone()),
"Stop",
"Stop",
)],
raw_errors: Vec::new(),
}]
);
@@ -1456,7 +1492,12 @@ async fn import_repo_mcp_uses_home_settings_toggles_when_repo_settings_missing()
cwd: Some(repo_root.clone()),
success_count: 1,
error_count: 0,
successes: Vec::new(),
successes: vec![import_success(
ExternalAgentConfigMigrationItemType::McpServerConfig,
Some(repo_root.clone()),
"allowed",
"allowed",
)],
raw_errors: Vec::new(),
}]
);
@@ -2745,7 +2786,7 @@ async fn import_plugins_supports_project_relative_external_agent_plugin_marketpl
}
#[test]
fn import_skills_returns_only_new_skill_directory_count() {
fn import_skills_returns_only_new_skill_directory_names() {
let (_root, external_agent_home, codex_home) = fixture_paths();
let agents_skills = codex_home
.parent()
@@ -2757,9 +2798,9 @@ fn import_skills_returns_only_new_skill_directory_count() {
.expect("create source b");
fs::create_dir_all(agents_skills.join("skill-a")).expect("create existing target");
let copied_count = service_for_paths(external_agent_home, codex_home)
let copied_names = service_for_paths(external_agent_home, codex_home)
.import_skills(/*cwd*/ None)
.expect("import skills");
assert_eq!(copied_count, 1);
assert_eq!(copied_names, vec!["skill-b".to_string()]);
}
codex-rs/app-server/src/message_processor.rs
+18 -10
View File
@@ -24,6 +24,7 @@ use crate::request_processors::CommandExecRequestProcessor;
use crate::request_processors::ConfigRequestProcessor;
use crate::request_processors::EnvironmentRequestProcessor;
use crate::request_processors::ExternalAgentConfigRequestProcessor;
use crate::request_processors::ExternalAgentConfigRequestProcessorArgs;
use crate::request_processors::FeedbackRequestProcessor;
use crate::request_processors::FsRequestProcessor;
use crate::request_processors::GitRequestProcessor;
@@ -475,7 +476,7 @@ impl MessageProcessor {
thread_watch_manager.clone(),
Arc::clone(&thread_list_state_permit),
thread_goal_processor.clone(),
state_db,
state_db.clone(),
log_db,
Arc::clone(&skills_watcher),
);
@@ -511,15 +512,17 @@ impl MessageProcessor {
thread_manager.clone(),
analytics_events_client,
);
let external_agent_config_processor = ExternalAgentConfigRequestProcessor::new(
outgoing.clone(),
Arc::clone(&thread_manager),
Arc::clone(&thread_store),
config_manager.clone(),
config_processor.clone(),
arg0_paths,
config.codex_home.to_path_buf(),
);
let external_agent_config_processor =
ExternalAgentConfigRequestProcessor::new(ExternalAgentConfigRequestProcessorArgs {
outgoing: outgoing.clone(),
thread_manager: Arc::clone(&thread_manager),
thread_store: Arc::clone(&thread_store),
config_manager: config_manager.clone(),
config_processor: config_processor.clone(),
state_db,
arg0_paths,
codex_home: config.codex_home.to_path_buf(),
});
let environment_processor =
EnvironmentRequestProcessor::new(thread_manager.environment_manager());
let fs_processor = FsRequestProcessor::new(
@@ -947,6 +950,11 @@ impl MessageProcessor {
.import(request_id.clone(), params)
.await
.map(|()| None),
ClientRequest::ExternalAgentConfigImportHistoriesRead { .. } => self
.external_agent_config_processor
.read_import_histories()
.await
.map(|response| Some(response.into())),
ClientRequest::ConfigValueWrite { params, .. } => {
self.config_processor.value_write(params).await.map(Some)
}
codex-rs/app-server/src/request_processors.rs
+1
View File
@@ -506,6 +506,7 @@ pub(crate) use command_exec_processor::CommandExecRequestProcessor;
pub(crate) use config_processor::ConfigRequestProcessor;
pub(crate) use environment_processor::EnvironmentRequestProcessor;
pub(crate) use external_agent_config_processor::ExternalAgentConfigRequestProcessor;
pub(crate) use external_agent_config_processor::ExternalAgentConfigRequestProcessorArgs;
pub(crate) use feedback_processor::FeedbackRequestProcessor;
pub(crate) use fs_processor::FsRequestProcessor;
pub(crate) use git_processor::GitRequestProcessor;
codex-rs/app-server/src/request_processors/external_agent_config_processor.rs
+217 -12
View File
@@ -19,9 +19,12 @@ 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::ExternalAgentConfigImportHistoriesReadResponse;
use codex_app_server_protocol::ExternalAgentConfigImportHistory;
use codex_app_server_protocol::ExternalAgentConfigImportItemTypeFailure as ProtocolImportFailure;
use codex_app_server_protocol::ExternalAgentConfigImportItemTypeSuccess as ProtocolImportSuccess;
use codex_app_server_protocol::ExternalAgentConfigImportParams;
use codex_app_server_protocol::ExternalAgentConfigImportProgressNotification;
use codex_app_server_protocol::ExternalAgentConfigImportResponse;
use codex_app_server_protocol::ExternalAgentConfigImportTypeResult as ProtocolImportTypeResult;
use codex_app_server_protocol::ExternalAgentConfigMigrationItem;
@@ -35,6 +38,9 @@ use codex_app_server_protocol::ServerNotification;
use codex_arg0::Arg0DispatchPaths;
use codex_core::ThreadManager;
use codex_external_agent_sessions::ExternalAgentSessionMigration as CoreSessionMigration;
use codex_rollout::StateDbHandle;
use codex_state::ExternalAgentConfigImportFailureRecord;
use codex_state::ExternalAgentConfigImportSuccessRecord;
use codex_thread_store::ThreadStore;
use std::collections::HashSet;
use std::path::PathBuf;
@@ -50,18 +56,32 @@ pub(crate) struct ExternalAgentConfigRequestProcessor {
session_importer: ExternalAgentSessionImporter,
thread_manager: Arc<ThreadManager>,
config_processor: ConfigRequestProcessor,
state_db: Option<StateDbHandle>,
}
pub(crate) struct ExternalAgentConfigRequestProcessorArgs {
pub(crate) outgoing: Arc<OutgoingMessageSender>,
pub(crate) thread_manager: Arc<ThreadManager>,
pub(crate) thread_store: Arc<dyn ThreadStore>,
pub(crate) config_manager: ConfigManager,
pub(crate) config_processor: ConfigRequestProcessor,
pub(crate) state_db: Option<StateDbHandle>,
pub(crate) arg0_paths: Arg0DispatchPaths,
pub(crate) codex_home: PathBuf,
}
impl ExternalAgentConfigRequestProcessor {
pub(crate) fn new(
outgoing: Arc<OutgoingMessageSender>,
thread_manager: Arc<ThreadManager>,
thread_store: Arc<dyn ThreadStore>,
config_manager: ConfigManager,
config_processor: ConfigRequestProcessor,
arg0_paths: Arg0DispatchPaths,
codex_home: PathBuf,
) -> Self {
pub(crate) fn new(args: ExternalAgentConfigRequestProcessorArgs) -> Self {
let ExternalAgentConfigRequestProcessorArgs {
outgoing,
thread_manager,
thread_store,
config_manager,
config_processor,
state_db,
arg0_paths,
codex_home,
} = args;
let session_importer = ExternalAgentSessionImporter::new(
codex_home.clone(),
Arc::clone(&thread_manager),
@@ -75,6 +95,7 @@ impl ExternalAgentConfigRequestProcessor {
session_importer,
thread_manager,
config_processor,
state_db,
}
}
@@ -207,23 +228,31 @@ 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);
}
let has_background_imports = !import_outcome.pending_plugin_imports.is_empty()
|| !pending_session_imports.is_empty();
if !has_background_imports {
send_completed_import_notification(&self.outgoing, import_id, &completed_item_results)
.await;
send_completed_import_notification(
&self.outgoing,
self.state_db.as_ref(),
import_id,
&completed_item_results,
)
.await;
return Ok(());
}
let session_importer = self.session_importer.clone();
let plugin_processor = self.clone();
let outgoing = Arc::clone(&self.outgoing);
let state_db = self.state_db.clone();
let thread_manager = Arc::clone(&self.thread_manager);
let session_import_result = (!pending_session_imports.is_empty()).then(|| {
CoreImportItemResult::new(
@@ -234,13 +263,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,6 +300,12 @@ impl ExternalAgentConfigRequestProcessor {
);
}
}
send_import_progress(
&plugin_progress_outgoing,
&plugin_import_id,
&item_result,
)
.await;
item_results.push(item_result);
}
item_results
@@ -280,12 +321,37 @@ impl ExternalAgentConfigRequestProcessor {
thread_manager.plugins_manager().clear_cache();
thread_manager.skills_manager().clear_cache();
}
send_completed_import_notification(&outgoing, import_id, &completed_item_results).await;
send_completed_import_notification(
&outgoing,
state_db.as_ref(),
import_id,
&completed_item_results,
)
.await;
});
Ok(())
}
pub(crate) async fn read_import_histories(
&self,
) -> Result<ExternalAgentConfigImportHistoriesReadResponse, JSONRPCErrorError> {
let state_db = self
.state_db
.as_ref()
.ok_or_else(|| internal_error("state database is unavailable"))?;
let histories = state_db
.external_agent_config_import_history_records()
.await
.map_err(|err| internal_error(format!("failed to read import histories: {err}")))?;
let data = histories
.into_iter()
.map(protocol_import_history)
.collect::<Result<Vec<_>, _>>()?;
Ok(ExternalAgentConfigImportHistoriesReadResponse { data })
}
fn validate_pending_session_imports(
&self,
params: &ExternalAgentConfigImportParams,
@@ -467,12 +533,37 @@ 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_type_results: vec![protocol_import_type_result(item_result)],
},
))
.await;
}
async fn send_completed_import_notification(
outgoing: &OutgoingMessageSender,
state_db: Option<&StateDbHandle>,
import_id: String,
item_results: &[CoreImportItemResult],
) {
let notification = completed_notification(import_id, item_results);
if let Some(state_db) = state_db
&& let Err(err) = record_completed_import_notification(state_db, &notification).await
{
tracing::warn!(
import_id = %notification.import_id,
error = %err,
"failed to record external agent config import completion"
);
}
outgoing
.send_server_notification(ServerNotification::ExternalAgentConfigImportCompleted(
notification,
@@ -480,6 +571,103 @@ async fn send_completed_import_notification(
.await;
}
async fn record_completed_import_notification(
state_db: &StateDbHandle,
notification: &ExternalAgentConfigImportCompletedNotification,
) -> anyhow::Result<()> {
let successes = notification
.item_type_results
.iter()
.flat_map(|type_result| type_result.successes.iter())
.map(|success| {
Ok(ExternalAgentConfigImportSuccessRecord {
item_type: serde_json::from_value(serde_json::to_value(success.item_type)?)?,
cwd: success.cwd.clone(),
source: success.source.clone(),
target: success.target.clone(),
})
})
.collect::<anyhow::Result<Vec<_>>>()?;
let failures = notification
.item_type_results
.iter()
.flat_map(|type_result| type_result.failures.iter())
.map(|failure| {
Ok(ExternalAgentConfigImportFailureRecord {
item_type: serde_json::from_value(serde_json::to_value(failure.item_type)?)?,
error_type: failure.error_type.clone(),
failure_stage: failure.failure_stage.clone(),
message: failure.message.clone(),
cwd: failure.cwd.clone(),
source: failure.source.clone(),
})
})
.collect::<anyhow::Result<Vec<_>>>()?;
state_db
.record_external_agent_config_import_completed(
notification.import_id.as_str(),
&successes,
&failures,
)
.await
}
fn protocol_import_history(
record: codex_state::ExternalAgentConfigImportHistoryRecord,
) -> Result<ExternalAgentConfigImportHistory, JSONRPCErrorError> {
let successes = record
.successes
.into_iter()
.map(protocol_import_success_record)
.collect::<Result<Vec<_>, _>>()?;
let failures = record
.failures
.into_iter()
.map(protocol_import_failure_record)
.collect::<Result<Vec<_>, _>>()?;
Ok(ExternalAgentConfigImportHistory {
import_id: record.import_id,
completed_at_ms: record.completed_at_ms,
successes,
failures,
})
}
fn protocol_import_success_record(
record: ExternalAgentConfigImportSuccessRecord,
) -> Result<ProtocolImportSuccess, JSONRPCErrorError> {
Ok(ProtocolImportSuccess {
item_type: protocol_import_record_item_type(record.item_type)?,
cwd: record.cwd,
source: record.source,
target: record.target,
})
}
fn protocol_import_failure_record(
record: ExternalAgentConfigImportFailureRecord,
) -> Result<ProtocolImportFailure, JSONRPCErrorError> {
Ok(ProtocolImportFailure {
item_type: protocol_import_record_item_type(record.item_type)?,
error_type: record.error_type,
failure_stage: record.failure_stage,
message: record.message,
cwd: record.cwd,
source: record.source,
})
}
fn protocol_import_record_item_type(
item_type: String,
) -> Result<ExternalAgentConfigMigrationItemType, JSONRPCErrorError> {
serde_json::from_value(serde_json::Value::String(item_type.clone())).map_err(|err| {
internal_error(format!(
"failed to decode import item type {item_type}: {err}"
))
})
}
fn completed_notification(
import_id: String,
item_results: &[CoreImportItemResult],
@@ -529,6 +717,22 @@ fn completed_notification(
}
}
fn protocol_import_type_result(item_result: &CoreImportItemResult) -> ProtocolImportTypeResult {
ProtocolImportTypeResult {
item_type: protocol_migration_item_type(item_result.item_type),
successes: item_result
.successes
.iter()
.map(protocol_import_success)
.collect(),
failures: item_result
.raw_errors
.iter()
.map(protocol_import_raw_error)
.collect(),
}
}
fn protocol_import_success(
success: &crate::config::external_agent_config::ExternalAgentConfigImportSuccess,
) -> ProtocolImportSuccess {
@@ -543,6 +747,7 @@ fn protocol_import_success(
fn protocol_import_raw_error(raw_error: &CoreImportRawError) -> ProtocolImportFailure {
ProtocolImportFailure {
item_type: protocol_migration_item_type(raw_error.item_type),
error_type: raw_error.error_type.clone(),
failure_stage: raw_error.failure_stage.clone(),
message: raw_error.message.clone(),
cwd: raw_error.cwd.clone(),