Expose connector candidates from imported sessions (#33414)

## What changed

- Correlate MCP server attribution in imported session records with connector
  metadata, and persist normalized connector names in the session import ledger.
- Extend `externalAgentConfig/import/readHistories` with connector candidates,
  including each connector's display name, imported-session count, and detection
  source.

## Testing

- Cover connector detection through the session import and history-read flow.
- Verify candidate aggregation uses the latest import for each source session.

GitOrigin-RevId: 7f1f7f596925b2e6efcd17afd4490148e1ac9ff1
This commit is contained in:
charlesgong-openai authored and copyberry committed 2026-07-15 20:06:16 +00:00
1 parent 8530440b91
commit 58e4ed17a0
21 files changed
+587 -38

No files matched your search

@@ -0,0 +1,115 @@
use serde::Deserialize;
use std::collections::BTreeMap;
use std::collections::BTreeSet;
use std::fs;
use std::path::Path;
use std::path::PathBuf;
const SESSION_MANIFESTS_DIR: &str = "claude-code-sessions";
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ImportedSessionConnectorAttribution {
pub session_id: String,
pub server_ids: BTreeSet<String>,
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct SessionManifest {
cli_session_id: Option<String>,
#[serde(default)]
remote_mcp_servers_config: Vec<RemoteMcpServerConfig>,
}
#[derive(Deserialize)]
struct RemoteMcpServerConfig {
name: Option<String>,
uuid: Option<String>,
}
pub fn detect_imported_cla_session_connectors(
session_attributions: &[ImportedSessionConnectorAttribution],
connector_metadata_roots: &[PathBuf],
) -> BTreeMap<String, Vec<String>> {
if session_attributions.is_empty() {
return BTreeMap::new();
}
let attributed_server_ids_by_session = session_attributions
.iter()
.map(|attribution| {
(
attribution.session_id.clone(),
attribution.server_ids.clone(),
)
})
.collect::<BTreeMap<_, _>>();
let mut connector_names_by_session = BTreeMap::<String, BTreeMap<String, String>>::new();
for metadata_root in connector_metadata_roots {
let manifests_root = metadata_root.join(SESSION_MANIFESTS_DIR);
for manifest_path in json_files_recursively(&manifests_root) {
let Some(manifest) = read_session_manifest(&manifest_path) else {
continue;
};
let Some(session_id) = manifest.cli_session_id else {
continue;
};
let Some(attributed_server_ids) = attributed_server_ids_by_session.get(&session_id)
else {
continue;
};
if attributed_server_ids.is_empty() {
continue;
}
let connector_names = connector_names_by_session.entry(session_id).or_default();
for server in manifest.remote_mcp_servers_config {
let Some(uuid) = server.uuid else {
continue;
};
if !attributed_server_ids.contains(&uuid) {
continue;
}
let Some(name) = super::normalized_connector_display_name(server.name.as_deref())
else {
continue;
};
connector_names.entry(name.to_lowercase()).or_insert(name);
}
}
}
connector_names_by_session
.into_iter()
.map(|(session_id, names)| (session_id, names.into_values().collect()))
.collect()
}
fn read_session_manifest(path: &Path) -> Option<SessionManifest> {
let contents = fs::read_to_string(path).ok()?;
serde_json::from_str(&contents).ok()
}
fn json_files_recursively(root: &Path) -> Vec<PathBuf> {
let mut files = Vec::new();
let mut pending = vec![root.to_path_buf()];
while let Some(directory) = pending.pop() {
let Ok(entries) = fs::read_dir(directory) else {
continue;
};
for entry in entries.flatten() {
let Ok(file_type) = entry.file_type() else {
continue;
};
if file_type.is_dir() {
pending.push(entry.path());
} else if file_type.is_file()
&& entry.path().extension().and_then(|value| value.to_str()) == Some("json")
{
files.push(entry.path());
}
}
}
files
}
@@ -18,6 +18,7 @@ use codex_protocol::protocol::TurnCompleteEvent;
use codex_protocol::protocol::TurnStartedEvent;
use codex_protocol::protocol::UserMessageEvent;
use codex_utils_output_truncation::approx_tokens_from_byte_count_i64;
use std::collections::BTreeSet;
use std::io;
use std::path::Path;
@@ -25,19 +26,18 @@ const EXTERNAL_SESSION_IMPORTED_MARKER: &str = "<EXTERNAL SESSION IMPORTED>";
#[cfg(test)]
fn load_session_for_import(path: &Path) -> io::Result<Option<ImportedExternalAgentSession>> {
Ok(
load_session_for_import_with_content_sha256(path)?
.map(|(session, _content_sha256)| session),
)
Ok(load_session_for_import_with_content_sha256(path)?
.map(|(session, _content_sha256, _attributed_mcp_server_ids)| session))
}
pub(crate) fn load_session_for_import_with_content_sha256(
path: &Path,
) -> io::Result<Option<(ImportedExternalAgentSession, String)>> {
) -> io::Result<Option<(ImportedExternalAgentSession, String, BTreeSet<String>)>> {
let parsed = read_session_import(path)?;
let Some(cwd) = parsed.cwd else {
return Ok(None);
};
let attributed_mcp_server_ids = parsed.attributed_mcp_server_ids;
let messages = parsed.messages;
let first_user_message_text = messages
.iter()
@@ -67,6 +67,7 @@ pub(crate) fn load_session_for_import_with_content_sha256(
rollout_items,
},
parsed.content_sha256,
attributed_mcp_server_ids,
)))
}
@@ -4,6 +4,7 @@ use serde::Deserialize;
use serde::Serialize;
use sha2::Digest;
use sha2::Sha256;
use std::collections::BTreeMap;
use std::collections::HashMap;
use std::fs;
use std::fs::File;
@@ -28,6 +29,8 @@ struct ImportedExternalAgentSessionRecord {
imported_at: i64,
#[serde(default)]
source_modified_at: Option<i64>,
#[serde(default)]
connector_names: Vec<String>,
}
#[derive(Debug, PartialEq, Eq)]
@@ -35,6 +38,13 @@ pub struct CompletedExternalAgentSessionImport {
pub source_path: PathBuf,
pub source_content_sha256: String,
pub imported_thread_id: ThreadId,
pub connector_names: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ImportedConnectorCandidate {
pub name: String,
pub session_count: u32,
}
#[derive(Debug, Clone, Copy)]
@@ -63,6 +73,7 @@ pub(crate) fn record_imported_session(
source_content_sha256: session_content_sha256(&source_path)?,
source_path,
imported_thread_id,
connector_names: Vec::new(),
}],
)
}
@@ -86,6 +97,7 @@ pub fn record_completed_session_imports(
record.imported_thread_id = import.imported_thread_id;
record.imported_at = imported_at;
record.source_modified_at = source_modified_at.or(record.source_modified_at);
record.connector_names = import.connector_names;
ledger.records.push(record);
continue;
}
@@ -95,11 +107,42 @@ pub fn record_completed_session_imports(
imported_thread_id: import.imported_thread_id,
imported_at,
source_modified_at,
connector_names: import.connector_names,
});
}
save_import_ledger(codex_home, &ledger)
}
pub fn read_imported_connector_candidates(
codex_home: &Path,
) -> io::Result<Vec<ImportedConnectorCandidate>> {
let ledger = load_import_ledger(codex_home)?;
let mut connector_names_by_source = BTreeMap::new();
for record in ledger.records {
connector_names_by_source.insert(record.source_path, record.connector_names);
}
let mut candidates_by_name = BTreeMap::<String, ImportedConnectorCandidate>::new();
for connector_names in connector_names_by_source.into_values() {
let connector_names = connector_names
.into_iter()
.filter_map(|name| super::normalized_connector_display_name(Some(&name)))
.map(|name| (name.to_lowercase(), name))
.collect::<BTreeMap<_, _>>();
for (key, name) in connector_names {
let candidate = candidates_by_name
.entry(key)
.or_insert(ImportedConnectorCandidate {
name,
session_count: 0,
});
candidate.session_count = candidate.session_count.saturating_add(1);
}
}
let mut candidates = candidates_by_name.into_values().collect::<Vec<_>>();
candidates.sort_by(|left, right| left.name.cmp(&right.name));
Ok(candidates)
}
impl ImportedExternalAgentSessionLedger {
pub(super) fn source_states(&self) -> HashMap<&Path, ImportedSourceState> {
let mut states = HashMap::new();
@@ -1,5 +1,7 @@
use super::CompletedExternalAgentSessionImport;
use super::ImportedConnectorCandidate;
use super::ImportedExternalAgentSessionLedger;
use super::read_imported_connector_candidates;
use super::record_completed_session_imports;
use codex_protocol::ThreadId;
use sha2::Digest;
@@ -35,6 +37,7 @@ fn completed_imports_do_not_read_source_files() {
source_path: source_path.clone(),
source_content_sha256: format!("{:x}", Sha256::digest(contents)),
imported_thread_id,
connector_names: Vec::new(),
}],
)
.expect("record completed imports");
@@ -64,6 +67,7 @@ fn completed_import_refreshes_existing_record_metadata() {
source_path: source_path.clone(),
source_content_sha256: content_sha256.clone(),
imported_thread_id: first_thread_id,
connector_names: vec!["Gmail".to_string()],
}],
)
.expect("record first import");
@@ -73,6 +77,7 @@ fn completed_import_refreshes_existing_record_metadata() {
source_path: source_path.clone(),
source_content_sha256: content_sha256,
imported_thread_id: second_thread_id,
connector_names: vec!["Slack".to_string()],
}],
)
.expect("record replacement import");
@@ -82,4 +87,52 @@ fn completed_import_refreshes_existing_record_metadata() {
assert_eq!(ledger.records[0].source_path, source_path);
assert_eq!(ledger.records[0].imported_thread_id, second_thread_id);
assert!(ledger.records[0].source_modified_at.is_some());
assert_eq!(ledger.records[0].connector_names, vec!["Slack"]);
}
#[test]
fn connector_candidates_use_latest_import_for_each_source() {
let root = TempDir::new().expect("tempdir");
let codex_home = root.path().join("codex-home");
let first_source = root.path().join("first.jsonl");
let second_source = root.path().join("second.jsonl");
record_completed_session_imports(
&codex_home,
vec![
CompletedExternalAgentSessionImport {
source_path: first_source.clone(),
source_content_sha256: "first-version".to_string(),
imported_thread_id: ThreadId::new(),
connector_names: vec!["Gmail".to_string()],
},
CompletedExternalAgentSessionImport {
source_path: first_source,
source_content_sha256: "second-version".to_string(),
imported_thread_id: ThreadId::new(),
connector_names: vec!["Slack".to_string()],
},
CompletedExternalAgentSessionImport {
source_path: second_source,
source_content_sha256: "only-version".to_string(),
imported_thread_id: ThreadId::new(),
connector_names: vec!["Gmail".to_string(), "Slack".to_string()],
},
],
)
.expect("record imports");
assert_eq!(
read_imported_connector_candidates(&codex_home).expect("read connector candidates"),
vec![
ImportedConnectorCandidate {
name: "Gmail".to_string(),
session_count: 1,
},
ImportedConnectorCandidate {
name: "Slack".to_string(),
session_count: 2,
},
]
);
}
@@ -1,5 +1,6 @@
//! Parsing and export helpers for external-agent session histories.
mod connectors_cla;
mod detect_cla;
mod export;
mod ledger;
@@ -7,20 +8,31 @@ mod records;
mod title;
use codex_protocol::protocol::RolloutItem;
use std::collections::BTreeSet;
use std::io;
use std::path::Path;
use std::path::PathBuf;
pub use connectors_cla::ImportedSessionConnectorAttribution;
pub use connectors_cla::detect_imported_cla_session_connectors;
pub use detect_cla::detect_recent_cla_sessions;
use export::load_session_for_import_with_content_sha256;
pub use ledger::CompletedExternalAgentSessionImport;
pub use ledger::ImportedConnectorCandidate;
pub use ledger::has_current_session_been_imported;
pub use ledger::read_imported_connector_candidates;
pub use ledger::record_completed_session_imports;
pub use records::SessionSummary;
pub use records::summarize_session;
const SESSION_TITLE_MAX_LEN: usize = 120;
fn normalized_connector_display_name(name: Option<&str>) -> Option<String> {
name.map(str::trim)
.filter(|name| !name.is_empty())
.map(ToOwned::to_owned)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ExternalAgentSessionMigration {
pub path: PathBuf,
@@ -40,6 +52,7 @@ pub struct ImportedExternalAgentSession {
pub struct PendingSessionImport {
pub source_path: PathBuf,
pub source_content_sha256: String,
pub attributed_mcp_server_ids: BTreeSet<String>,
pub session: ImportedExternalAgentSession,
}
@@ -51,32 +64,25 @@ pub fn prepare_validated_session_import(
if has_been_imported {
return Ok(None);
}
let Some((source_path, imported_session, source_content_sha256)) =
load_importable_session(&session.path)?
else {
return Ok(None);
};
Ok(Some(PendingSessionImport {
source_path,
source_content_sha256,
session: imported_session,
}))
load_importable_session(&session.path)
}
fn load_importable_session(
path: &Path,
) -> io::Result<Option<(PathBuf, ImportedExternalAgentSession, String)>> {
fn load_importable_session(path: &Path) -> io::Result<Option<PendingSessionImport>> {
let source_path = std::fs::canonicalize(path)?;
let Some((imported_session, source_content_sha256)) =
let Some((imported_session, source_content_sha256, attributed_mcp_server_ids)) =
load_session_for_import_with_content_sha256(&source_path)?
else {
return Ok(None);
};
Ok(imported_session.cwd.is_dir().then_some((
source_path,
imported_session,
source_content_sha256,
)))
Ok(imported_session
.cwd
.is_dir()
.then_some(PendingSessionImport {
source_path,
source_content_sha256,
attributed_mcp_server_ids,
session: imported_session,
}))
}
#[derive(Debug, Clone)]
@@ -8,6 +8,7 @@ use super::truncate;
use serde_json::Value as JsonValue;
use sha2::Digest;
use sha2::Sha256;
use std::collections::BTreeSet;
use std::fs::File;
use std::io;
use std::io::BufRead;
@@ -31,6 +32,7 @@ pub(super) struct ParsedSessionImport {
pub ai_title: Option<String>,
pub messages: Vec<ConversationMessage>,
pub content_sha256: String,
pub attributed_mcp_server_ids: BTreeSet<String>,
}
pub fn summarize_session(path: &Path) -> io::Result<Option<SessionSummary>> {
@@ -114,6 +116,7 @@ pub(super) fn read_session_import(path: &Path) -> io::Result<ParsedSessionImport
let mut custom_title = None;
let mut ai_title = None;
let mut messages = Vec::new();
let mut attributed_mcp_server_ids = BTreeSet::new();
let mut line = String::new();
let mut hasher = Sha256::new();
loop {
@@ -129,6 +132,14 @@ pub(super) fn read_session_import(path: &Path) -> io::Result<ParsedSessionImport
let Ok(mut record) = serde_json::from_str::<JsonValue>(trimmed) else {
continue;
};
if let Some(server_id) = record
.get("attributionMcpServer")
.and_then(JsonValue::as_str)
.map(str::trim)
.filter(|server_id| !server_id.is_empty())
{
attributed_mcp_server_ids.insert(server_id.to_string());
}
if cwd.is_none() {
cwd = record
.get("cwd")
@@ -151,6 +162,7 @@ pub(super) fn read_session_import(path: &Path) -> io::Result<ParsedSessionImport
ai_title,
messages,
content_sha256: format!("{:x}", hasher.finalize()),
attributed_mcp_server_ids,
})
}