Track connectors detected in external agent sessions (#36959)

## What changed

- Return connector candidates found in detected external agent sessions, aggregating session counts by connector name.
- Resolve connector names from both cached plugin manifests and project MCP server metadata.
- Persist detected connector clues in the session import ledger and merge them into completed imports without case-insensitive duplicates.

## Testing

- Add coverage for project MCP metadata detection and for accumulating detected and imported connector candidates.

GitOrigin-RevId: 97a7dfe4a61599686b2222cddd9595acec8b4501
This commit is contained in:
charlesgong-openai authored and copyberry committed 2026-08-04 19:45:56 +00:00
1 parent 4bd5b9fd09
commit 92689b6b7b
8 files changed
+399 -19

No files matched your search

@@ -19,6 +19,14 @@ const SESSION_HASH_BUFFER_SIZE: usize = 64 * 1024;
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct ImportedExternalAgentSessionLedger {
records: Vec<ImportedExternalAgentSessionRecord>,
#[serde(default)]
detected_connector_records: Vec<DetectedExternalAgentSessionConnectorRecord>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
struct DetectedExternalAgentSessionConnectorRecord {
source_path: PathBuf,
connector_names: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -176,40 +184,106 @@ 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;
append_connector_names(&mut record.connector_names, import.connector_names);
record.title = import.title;
ledger.records.push(record);
continue;
}
let mut connector_names = Vec::new();
append_connector_names(&mut connector_names, import.connector_names);
ledger.records.push(ImportedExternalAgentSessionRecord {
source_path: import.source_path,
content_sha256: import.source_content_sha256,
imported_thread_id: import.imported_thread_id,
imported_at,
source_modified_at,
connector_names: import.connector_names,
connector_names,
title: import.title,
});
}
save_import_ledger(codex_home, &ledger)
}
pub fn record_detected_session_connectors(
codex_home: &Path,
connector_names_by_source_path: BTreeMap<PathBuf, Vec<String>>,
) -> io::Result<()> {
if connector_names_by_source_path.is_empty() {
return Ok(());
}
let mut ledger = load_import_ledger(codex_home)?;
for (source_path, connector_names) in connector_names_by_source_path {
let source_path = canonical_source_path(&source_path)?;
if let Some(record) = ledger
.detected_connector_records
.iter_mut()
.find(|record| record.source_path == source_path)
{
append_connector_names(&mut record.connector_names, connector_names);
continue;
}
let mut detected_connector_names = Vec::new();
append_connector_names(&mut detected_connector_names, connector_names);
ledger
.detected_connector_records
.push(DetectedExternalAgentSessionConnectorRecord {
source_path,
connector_names: detected_connector_names,
});
}
save_import_ledger(codex_home, &ledger)
}
pub fn append_imported_session_connector_names(
codex_home: &Path,
connector_names_by_source_path: BTreeMap<PathBuf, Vec<String>>,
) -> io::Result<()> {
if connector_names_by_source_path.is_empty() {
return Ok(());
}
let mut ledger = load_import_ledger(codex_home)?;
for (source_path, connector_names) in connector_names_by_source_path {
let source_path = canonical_source_path(&source_path)?;
for record in ledger
.records
.iter_mut()
.filter(|record| record.source_path == source_path)
{
append_connector_names(&mut record.connector_names, connector_names.clone());
}
}
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();
let mut connector_names_by_source = BTreeMap::<PathBuf, Vec<String>>::new();
for record in ledger.detected_connector_records {
connector_names_by_source
.entry(record.source_path)
.or_default()
.extend(record.connector_names);
}
for record in ledger.records {
connector_names_by_source.insert(record.source_path, record.connector_names);
connector_names_by_source
.entry(record.source_path)
.or_default()
.extend(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
let mut connector_names_by_key = BTreeMap::new();
for name in 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 {
{
connector_names_by_key
.entry(name.to_lowercase())
.or_insert(name);
}
for (key, name) in connector_names_by_key {
let candidate = candidates_by_name
.entry(key)
.or_insert(ImportedConnectorCandidate {
@@ -224,6 +298,18 @@ pub fn read_imported_connector_candidates(
Ok(candidates)
}
fn append_connector_names(connector_names: &mut Vec<String>, additional_names: Vec<String>) {
for name in additional_names {
if connector_names
.iter()
.any(|existing_name| existing_name.eq_ignore_ascii_case(&name))
{
continue;
}
connector_names.push(name);
}
}
impl ImportedExternalAgentSessionLedger {
pub(crate) fn source_states(&self) -> HashMap<&Path, ImportedSourceState> {
let mut states = HashMap::new();
@@ -1,13 +1,16 @@
use super::CompletedExternalAgentSessionImport;
use super::ImportedConnectorCandidate;
use super::ImportedExternalAgentSessionLedger;
use super::append_imported_session_connector_names;
use super::checkpoint_existing_session_import;
use super::read_imported_connector_candidates;
use super::record_completed_session_imports;
use super::record_detected_session_connectors;
use codex_protocol::ThreadId;
use pretty_assertions::assert_eq;
use sha2::Digest;
use sha2::Sha256;
use std::collections::BTreeMap;
use std::path::Path;
use tempfile::TempDir;
@@ -94,7 +97,7 @@ 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"]);
assert_eq!(ledger.records[0].connector_names, vec!["Gmail", "Slack"]);
assert_eq!(ledger.records[0].title.as_deref(), Some("Second title"));
}
@@ -168,7 +171,7 @@ fn checkpoint_is_compare_and_swap_and_preserves_record_metadata() {
}
#[test]
fn connector_candidates_use_latest_import_for_each_source() {
fn connector_candidates_accumulate_imports_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");
@@ -207,7 +210,7 @@ fn connector_candidates_use_latest_import_for_each_source() {
vec![
ImportedConnectorCandidate {
name: "Gmail".to_string(),
session_count: 1,
session_count: 2,
},
ImportedConnectorCandidate {
name: "Slack".to_string(),
@@ -217,6 +220,76 @@ fn connector_candidates_use_latest_import_for_each_source() {
);
}
#[test]
fn connector_candidates_accumulate_detected_and_imported_session_clues() {
let root = TempDir::new().expect("tempdir");
let codex_home = root.path().join("codex-home");
let source_path = root.path().join("session.jsonl");
std::fs::write(&source_path, "session contents").expect("source");
let source_path = std::fs::canonicalize(source_path).expect("canonical source");
record_detected_session_connectors(
&codex_home,
BTreeMap::from([(source_path.clone(), vec!["Figma".to_string()])]),
)
.expect("record detected connectors");
assert_eq!(
read_imported_connector_candidates(&codex_home).expect("read detected connectors"),
vec![ImportedConnectorCandidate {
name: "Figma".to_string(),
session_count: 1,
}]
);
record_completed_session_imports(
&codex_home,
vec![CompletedExternalAgentSessionImport {
source_path: source_path.clone(),
source_content_sha256: "content-sha".to_string(),
imported_thread_id: ThreadId::new(),
connector_names: vec!["Slack".to_string()],
title: None,
}],
)
.expect("record completed import");
assert_eq!(
read_imported_connector_candidates(&codex_home).expect("read imported connectors"),
vec![
ImportedConnectorCandidate {
name: "Figma".to_string(),
session_count: 1,
},
ImportedConnectorCandidate {
name: "Slack".to_string(),
session_count: 1,
},
]
);
append_imported_session_connector_names(
&codex_home,
BTreeMap::from([(source_path, vec!["Gmail".to_string(), "FIGMA".to_string()])]),
)
.expect("append imported connector names");
assert_eq!(
read_imported_connector_candidates(&codex_home).expect("read appended connectors"),
vec![
ImportedConnectorCandidate {
name: "Figma".to_string(),
session_count: 1,
},
ImportedConnectorCandidate {
name: "Gmail".to_string(),
session_count: 1,
},
ImportedConnectorCandidate {
name: "Slack".to_string(),
session_count: 1,
},
]
);
}
fn ledger_bytes(codex_home: &Path) -> Vec<u8> {
std::fs::read(super::import_ledger_path(codex_home)).expect("ledger")
}
@@ -24,9 +24,11 @@ pub use append::append_existing_session;
use export::load_session_for_import_with_content_sha256;
pub use ledger::CompletedExternalAgentSessionImport;
pub use ledger::ImportedConnectorCandidate;
pub use ledger::append_imported_session_connector_names;
pub use ledger::has_current_session_been_imported;
pub use ledger::read_imported_connector_candidates;
pub use ledger::record_completed_session_imports;
pub use ledger::record_detected_session_connectors;
pub use records_cla::summarize_session;
const SESSION_TITLE_MAX_LEN: usize = 120;