agent/codex
1
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

Import external agent sessions in background (#20284)

Browse Source 
Summary:
- Return from external agent import before session history import
finishes
- Run session import work in the background and emit the existing
completion notification when it is done
- Serialize session imports so duplicate requests do not create
duplicate imported threads

Verification:
- cargo test -p codex-app-server external_agent_config_
- cargo test -p codex-external-agent-sessions
- just fix -p codex-app-server
- just fix -p codex-external-agent-sessions
- git diff --check
This commit is contained in:
stefanstokic-oai
2026-04-29 17:00:41 -07:00
committed by GitHub
Unverified
parent 7bcd4626c4
commit c8abcbf925
8 changed files with 465 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/codex_message_processor.rs
+1
View File
@@ -518,6 +518,7 @@ impl Drop for ActiveLogin {
}
/// Handles JSON-RPC messages for Codex threads (and legacy conversation APIs).
#[derive(Clone)]
pub(crate) struct CodexMessageProcessor {
auth_manager: Arc<AuthManager>,
thread_manager: Arc<ThreadManager>,
codex-rs/app-server/src/config/external_agent_config.rs
+18 -2
View File
@@ -144,8 +144,24 @@ impl ExternalAgentConfigService {
Ok(items)
}
pub(crate) fn detect_recent_sessions(&self) -> io::Result<Vec<ExternalAgentSessionMigration>> {
detect_recent_sessions(&self.external_agent_home, &self.codex_home)
pub(crate) fn external_agent_session_source_path(
&self,
path: &Path,
) -> io::Result<Option<PathBuf>> {
if path.extension().and_then(|value| value.to_str()) != Some("jsonl") {
return Ok(None);
}
let path = match fs::canonicalize(path) {
Ok(path) => path,
Err(err) if err.kind() == io::ErrorKind::NotFound => return Ok(None),
Err(err) => return Err(err),
};
let projects_root = match fs::canonicalize(self.external_agent_home.join("projects")) {
Ok(projects_root) => projects_root,
Err(err) if err.kind() == io::ErrorKind::NotFound => return Ok(None),
Err(err) => return Err(err),
};
Ok(path.starts_with(projects_root).then_some(path))
}
pub(crate) async fn import(
codex-rs/app-server/src/external_agent_config_api.rs
+41 -24
View File
@@ -20,16 +20,19 @@ use codex_app_server_protocol::PluginsMigration;
use codex_app_server_protocol::SubagentMigration;
use codex_external_agent_sessions::ExternalAgentSessionMigration as CoreSessionMigration;
use codex_external_agent_sessions::PendingSessionImport;
use codex_external_agent_sessions::PrepareSessionImportsError;
use codex_external_agent_sessions::prepare_pending_session_imports;
use codex_external_agent_sessions::prepare_validated_session_imports;
use codex_external_agent_sessions::record_imported_session;
use codex_protocol::ThreadId;
use std::collections::HashSet;
use std::path::PathBuf;
use std::sync::Arc;
use tokio::sync::Semaphore;
#[derive(Clone)]
pub(crate) struct ExternalAgentConfigApi {
codex_home: PathBuf,
migration_service: ExternalAgentConfigService,
session_import_permits: Arc<Semaphore>,
}
impl ExternalAgentConfigApi {
@@ -37,6 +40,7 @@ impl ExternalAgentConfigApi {
Self {
migration_service: ExternalAgentConfigService::new(codex_home.clone()),
codex_home,
session_import_permits: Arc::new(Semaphore::new(1)),
}
}
@@ -134,18 +138,10 @@ impl ExternalAgentConfigApi {
})
}
pub(crate) fn detect_recent_sessions(
&self,
) -> Result<Vec<CoreSessionMigration>, JSONRPCErrorError> {
self.migration_service
.detect_recent_sessions()
.map_err(|err| internal_error(err.to_string()))
}
pub(crate) fn prepare_pending_session_imports(
pub(crate) fn validate_pending_session_imports(
&self,
params: &ExternalAgentConfigImportParams,
) -> Result<Vec<PendingSessionImport>, JSONRPCErrorError> {
) -> Result<Vec<CoreSessionMigration>, JSONRPCErrorError> {
let sessions = params
.migration_items
.iter()
@@ -163,18 +159,32 @@ impl ExternalAgentConfigApi {
title: session.title,
})
.collect::<Vec<_>>();
let detected_sessions = if sessions.is_empty() {
Vec::new()
} else {
self.detect_recent_sessions()?
};
prepare_pending_session_imports(&self.codex_home, sessions, detected_sessions).map_err(
|err| match err {
PrepareSessionImportsError::SessionNotDetected(_) => {
invalid_params(err.to_string())
}
},
)
let mut selected_session_paths = HashSet::new();
let mut selected_sessions = Vec::new();
for session in sessions {
let Some(canonical_path) = self
.migration_service
.external_agent_session_source_path(&session.path)
.map_err(|err| internal_error(err.to_string()))?
else {
return Err(session_not_detected_error(&session.path));
};
if selected_session_paths.insert(canonical_path) {
selected_sessions.push(session);
}
}
Ok(selected_sessions)
}
pub(crate) fn prepare_validated_session_imports(
&self,
sessions: Vec<CoreSessionMigration>,
) -> Vec<PendingSessionImport> {
prepare_validated_session_imports(&self.codex_home, sessions)
}
pub(crate) fn session_import_permits(&self) -> Arc<Semaphore> {
Arc::clone(&self.session_import_permits)
}
pub(crate) fn record_imported_session(
@@ -301,3 +311,10 @@ impl ExternalAgentConfigApi {
.map_err(|err| internal_error(err.to_string()))
}
}
fn session_not_detected_error(path: &std::path::Path) -> JSONRPCErrorError {
invalid_params(format!(
"external agent session was not detected for import: {}",
path.display()
))
}
codex-rs/app-server/src/message_processor.rs
+61 -24
View File
@@ -1224,30 +1224,28 @@ impl MessageProcessor {
ExternalAgentConfigMigrationItemType::Plugins
)
});
let has_session_imports = params.migration_items.iter().any(|item| {
matches!(
item.item_type,
ExternalAgentConfigMigrationItemType::Sessions
)
});
let pending_session_imports = self
.external_agent_config_api
.prepare_pending_session_imports(&params)?;
.validate_pending_session_imports(&params)?;
let pending_plugin_imports = self.external_agent_config_api.import(params).await?;
if needs_runtime_refresh {
self.handle_config_mutation().await;
}
for pending_session_import in pending_session_imports {
let imported_thread_id = self
.codex_message_processor
.import_external_agent_session(pending_session_import.session)
.await?;
self.external_agent_config_api
.record_imported_session(&pending_session_import.source_path, imported_thread_id);
}
self.outgoing
.send_response(request_id, ExternalAgentConfigImportResponse {})
.await;
if !has_plugin_imports {
if !has_plugin_imports && !has_session_imports {
return Ok(());
}
if pending_plugin_imports.is_empty() {
if pending_plugin_imports.is_empty() && pending_session_imports.is_empty() {
self.outgoing
.send_server_notification(ServerNotification::ExternalAgentConfigImportCompleted(
ExternalAgentConfigImportCompletedNotification {},
@@ -1257,25 +1255,64 @@ impl MessageProcessor {
}
let external_agent_config_api = self.external_agent_config_api.clone();
let session_import_permits = external_agent_config_api.session_import_permits();
let codex_message_processor = self.codex_message_processor.clone();
let outgoing = Arc::clone(&self.outgoing);
let thread_manager = Arc::clone(&self.thread_manager);
tokio::spawn(async move {
for pending_plugin_import in pending_plugin_imports {
match external_agent_config_api
.complete_pending_plugin_import(pending_plugin_import)
.await
{
Ok(()) => {}
Err(error) => {
tracing::warn!(
error = %error.message,
"external agent config plugin import failed"
);
let session_external_agent_config_api = external_agent_config_api.clone();
let plugin_external_agent_config_api = external_agent_config_api;
let session_imports = async move {
if !pending_session_imports.is_empty() {
let Ok(_session_import_permit) = session_import_permits.acquire_owned().await
else {
return;
};
let pending_session_imports = session_external_agent_config_api
.prepare_validated_session_imports(pending_session_imports);
for pending_session_import in pending_session_imports {
match codex_message_processor
.import_external_agent_session(pending_session_import.session)
.await
{
Ok(imported_thread_id) => {
session_external_agent_config_api.record_imported_session(
&pending_session_import.source_path,
imported_thread_id,
);
}
Err(error) => {
tracing::warn!(
error = %error.message,
path = %pending_session_import.source_path.display(),
"external agent session import failed"
);
}
}
}
}
};
let plugin_imports = async move {
for pending_plugin_import in pending_plugin_imports {
match plugin_external_agent_config_api
.complete_pending_plugin_import(pending_plugin_import)
.await
{
Ok(()) => {}
Err(error) => {
tracing::warn!(
error = %error.message,
"external agent config plugin import failed"
);
}
}
}
};
tokio::join!(session_imports, plugin_imports);
if has_plugin_imports {
thread_manager.plugins_manager().clear_cache();
thread_manager.skills_manager().clear_cache();
}
thread_manager.plugins_manager().clear_cache();
thread_manager.skills_manager().clear_cache();
outgoing
.send_server_notification(ServerNotification::ExternalAgentConfigImportCompleted(
ExternalAgentConfigImportCompletedNotification {},