diff --git a/src/openhuman/composio/client.rs b/src/openhuman/composio/client.rs index 1e33a361b..86a6f8459 100644 --- a/src/openhuman/composio/client.rs +++ b/src/openhuman/composio/client.rs @@ -975,6 +975,12 @@ pub async fn direct_list_connections( toolkit, status, created_at: item.created_at.clone(), + // Identity fields are populated by + // `enrich_connections_with_identity` in ops.rs after + // the full list is fetched, using cached profile data. + account_email: None, + workspace: None, + username: None, }) }) .collect(); diff --git a/src/openhuman/composio/ops.rs b/src/openhuman/composio/ops.rs index 05a04594b..4e4793a7c 100644 --- a/src/openhuman/composio/ops.rs +++ b/src/openhuman/composio/ops.rs @@ -374,6 +374,7 @@ pub async fn composio_list_connections( // connected-toolkits view stays in sync within one poll // interval. sync_cache_with_connections(&resp.connections); + let resp = enrich_connections_with_identity(resp); return Ok(RpcOutcome::new( resp, vec![format!( @@ -395,6 +396,7 @@ pub async fn composio_list_connections( // timeout fires before the user finishes the hosted flow) is still // reflected in chat within one poll interval. sync_cache_with_connections(&resp.connections); + let resp = enrich_connections_with_identity(resp); Ok(RpcOutcome::new( resp, vec![format!( @@ -1506,6 +1508,74 @@ fn parse_sync_reason(raw: Option<&str>) -> OpResult { } } +/// Enrich each [`super::types::ComposioConnection`] with human-readable +/// identity fields (`account_email`, `workspace`, `username`) from the +/// persisted provider profile cache so the UI picker can show +/// "Gmail · user@example.com" instead of a generic "Account N" label. +/// +/// This is best-effort — no live API calls are made (one SQLite read per poll). +/// If the memory client is not ready yet (first launch before any sync) +/// or no profile rows exist for a connection, that connection is returned +/// unchanged and the UI falls back to its numbered label logic. +/// +/// Connections that already carry identity fields (e.g. from the +/// backend-proxied path) are left untouched. +fn enrich_connections_with_identity( + mut resp: super::types::ComposioConnectionsResponse, +) -> super::types::ComposioConnectionsResponse { + use std::collections::HashMap; + + use super::providers::profile::{load_connected_identities, normalize_connection_identifier}; + + let identities = load_connected_identities(); + if identities.is_empty() { + tracing::debug!( + "[composio] enrich_connections_with_identity: no cached identities yet \ + — picker will fall back to numbered labels until first sync completes" + ); + return resp; + } + + // (normalized_toolkit, normalized_conn_id) → identity + let lookup: HashMap<(String, String), _> = identities + .iter() + .map(|id| ((id.source.clone(), id.identifier.clone()), id)) + .collect(); + + tracing::debug!( + total = resp.connections.len(), + cached_identities = identities.len(), + "[composio] enrich_connections_with_identity: enriching connection labels" + ); + + for conn in &mut resp.connections { + // Skip connections already carrying identity info (backend-proxied + // path may supply them directly). + if conn.account_email.is_some() || conn.workspace.is_some() || conn.username.is_some() { + continue; + } + let toolkit_key = normalize_connection_identifier(&conn.toolkit); + let conn_id_key = normalize_connection_identifier(&conn.id); + if let Some(identity) = lookup.get(&(toolkit_key, conn_id_key)) { + conn.account_email = identity.email.clone(); + // display_name carries the user's name; for Slack it falls back + // to the team/workspace name when no per-user display name exists, + // making it the best available workspace signal. + conn.workspace = identity.display_name.clone(); + conn.username = identity.handle.clone(); + tracing::debug!( + toolkit = %conn.toolkit, + connection_id = %conn.id, + has_email = conn.account_email.is_some(), + has_workspace = conn.workspace.is_some(), + has_username = conn.username.is_some(), + "[composio] enrich_connections_with_identity: enriched connection" + ); + } + } + resp +} + #[cfg(test)] pub(crate) use super::connected_integrations::cache_key; use super::connected_integrations::sync_cache_with_connections; diff --git a/src/openhuman/composio/ops_tests.rs b/src/openhuman/composio/ops_tests.rs index 50a9eaf01..4e5e470c1 100644 --- a/src/openhuman/composio/ops_tests.rs +++ b/src/openhuman/composio/ops_tests.rs @@ -2257,3 +2257,196 @@ fn composio_direct_500_does_not_demote() { "composio-direct 500 with no auth body must remain an unclassified bug shape" ); } + +// ── enrich_connections_with_identity ────────────────────────────────── + +/// Helper: bind the process-global memory client to a fresh temp workspace. +/// +/// Initialise the global memory client to an isolated temp workspace and return +/// a serialisation lock that must be held for the duration of the test. +/// +/// The global rebinds whenever the workspace path changes (see +/// `memory::global::init`), so each test that passes a unique `TempDir` gets +/// isolated profile storage. The returned guard prevents concurrent tests from +/// rebinding the singleton while the calling test is running — hold it with +/// `let _guard = init_memory_client(tmp.path());`. +fn init_memory_client(workspace: &std::path::Path) -> std::sync::MutexGuard<'static, ()> { + static ENRICH_IDENTITY_TEST_LOCK: std::sync::OnceLock> = + std::sync::OnceLock::new(); + let guard = ENRICH_IDENTITY_TEST_LOCK + .get_or_init(|| std::sync::Mutex::new(())) + .lock() + .unwrap_or_else(|e| e.into_inner()); + let _ = crate::openhuman::memory::global::init(workspace.to_path_buf()); + guard +} + +fn make_connections_response( + conns: &[(&str, &str, &str)], +) -> super::super::types::ComposioConnectionsResponse { + let connections = conns + .iter() + .map(|(id, toolkit, status)| conn(id, toolkit, status)) + .collect(); + super::super::types::ComposioConnectionsResponse { connections } +} + +#[tokio::test] +async fn enrich_does_nothing_when_no_cached_identities() { + // Hold the lock so no sibling test can rebind the global to a workspace + // that has a profile row matching "c1". The fresh temp workspace has no + // profiles, so load_connected_identities returns Vec::new() and the + // connection is returned unchanged. + let tmp = tempfile::tempdir().unwrap(); + let _guard = init_memory_client(tmp.path()); + let resp = make_connections_response(&[("c1", "gmail", "ACTIVE")]); + let enriched = enrich_connections_with_identity(resp); + assert_eq!(enriched.connections.len(), 1); + assert!(enriched.connections[0].account_email.is_none()); + assert!(enriched.connections[0].workspace.is_none()); + assert!(enriched.connections[0].username.is_none()); +} + +#[tokio::test] +async fn enrich_populates_email_from_cached_profile() { + use crate::openhuman::memory_sync::composio::providers::{ + profile::persist_provider_profile, ProviderUserProfile, + }; + let tmp = tempfile::tempdir().unwrap(); + let _guard = init_memory_client(tmp.path()); + + persist_provider_profile(&ProviderUserProfile { + toolkit: "gmail".to_string(), + connection_id: Some("conn-gmail-1".to_string()), + email: Some("alice@example.com".to_string()), + display_name: Some("Alice Smith".to_string()), + ..Default::default() + }); + + let resp = make_connections_response(&[("conn-gmail-1", "gmail", "ACTIVE")]); + let enriched = enrich_connections_with_identity(resp); + + assert_eq!( + enriched.connections[0].account_email.as_deref(), + Some("alice@example.com"), + "email should be populated from cached gmail profile" + ); + assert_eq!( + enriched.connections[0].workspace.as_deref(), + Some("Alice Smith"), + "workspace (display_name) should be populated" + ); + assert!( + enriched.connections[0].username.is_none(), + "username (handle) should be absent for gmail" + ); +} + +#[tokio::test] +async fn enrich_populates_handle_for_github() { + use crate::openhuman::memory_sync::composio::providers::{ + profile::persist_provider_profile, ProviderUserProfile, + }; + let tmp = tempfile::tempdir().unwrap(); + let _guard = init_memory_client(tmp.path()); + + persist_provider_profile(&ProviderUserProfile { + toolkit: "github".to_string(), + connection_id: Some("conn-gh-1".to_string()), + username: Some("octocat".to_string()), + ..Default::default() + }); + + let resp = make_connections_response(&[("conn-gh-1", "github", "ACTIVE")]); + let enriched = enrich_connections_with_identity(resp); + + // GitHub uses `handle` kind (the catch-all branch in expand_identity_rows). + assert_eq!( + enriched.connections[0].username.as_deref(), + Some("octocat"), + "username (handle) should be populated for github" + ); + assert!(enriched.connections[0].account_email.is_none()); +} + +#[tokio::test] +async fn enrich_skips_connection_already_having_identity() { + // If the backend-proxied path already populated account_email, the + // enricher must NOT overwrite it with a potentially stale cached value. + let mut resp = make_connections_response(&[("c-preloaded", "gmail", "ACTIVE")]); + resp.connections[0].account_email = Some("preloaded@example.com".to_string()); + + let enriched = enrich_connections_with_identity(resp); + assert_eq!( + enriched.connections[0].account_email.as_deref(), + Some("preloaded@example.com"), + "pre-populated identity must not be overwritten" + ); +} + +#[tokio::test] +async fn enrich_handles_multiple_connections_same_toolkit() { + // Two Gmail accounts — each gets its own identity label, not "Account N". + use crate::openhuman::memory_sync::composio::providers::{ + profile::persist_provider_profile, ProviderUserProfile, + }; + let tmp = tempfile::tempdir().unwrap(); + let _guard = init_memory_client(tmp.path()); + + persist_provider_profile(&ProviderUserProfile { + toolkit: "gmail".to_string(), + connection_id: Some("g1".to_string()), + email: Some("alice@example.com".to_string()), + ..Default::default() + }); + persist_provider_profile(&ProviderUserProfile { + toolkit: "gmail".to_string(), + connection_id: Some("g2".to_string()), + email: Some("bob@example.com".to_string()), + ..Default::default() + }); + + let resp = make_connections_response(&[("g1", "gmail", "ACTIVE"), ("g2", "gmail", "ACTIVE")]); + let enriched = enrich_connections_with_identity(resp); + + let emails: Vec<_> = enriched + .connections + .iter() + .map(|c| c.account_email.as_deref()) + .collect(); + assert!( + emails.contains(&Some("alice@example.com")), + "first gmail account should carry alice's email" + ); + assert!( + emails.contains(&Some("bob@example.com")), + "second gmail account should carry bob's email" + ); +} + +#[tokio::test] +async fn enrich_leaves_unmatched_connection_unchanged() { + // Connection whose id has no cached profile row is returned with all + // identity fields as None — the UI falls back to "Account N". + use crate::openhuman::memory_sync::composio::providers::{ + profile::persist_provider_profile, ProviderUserProfile, + }; + let tmp = tempfile::tempdir().unwrap(); + let _guard = init_memory_client(tmp.path()); + + // Persist a profile for a DIFFERENT connection id. + persist_provider_profile(&ProviderUserProfile { + toolkit: "gmail".to_string(), + connection_id: Some("other-conn".to_string()), + email: Some("other@example.com".to_string()), + ..Default::default() + }); + + let resp = make_connections_response(&[("no-profile-conn", "gmail", "ACTIVE")]); + let enriched = enrich_connections_with_identity(resp); + + assert!( + enriched.connections[0].account_email.is_none(), + "connection with no cached profile must remain unenriched" + ); +} diff --git a/src/openhuman/composio/types.rs b/src/openhuman/composio/types.rs index 9dfbd6679..3da35db38 100644 --- a/src/openhuman/composio/types.rs +++ b/src/openhuman/composio/types.rs @@ -126,6 +126,26 @@ pub struct ComposioConnection { /// ISO timestamp (backend passes this through from Composio). #[serde(rename = "createdAt", default, skip_serializing_if = "Option::is_none")] pub created_at: Option, + /// Account email — populated from the cached provider profile when + /// the toolkit reports an email address (e.g. Gmail, Google Calendar, + /// Google Sheets). Lets the UI picker show "Gmail · user@example.com" + /// instead of a generic "Account N" label. + #[serde( + rename = "accountEmail", + default, + skip_serializing_if = "Option::is_none" + )] + pub account_email: Option, + /// Workspace or team display name — populated for workspace-based + /// services (e.g. Slack: user display name / team name, Notion: workspace + /// name). Used by the picker when no email is available. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub workspace: Option, + /// Screen name or handle — populated for username-based services + /// (e.g. GitHub login, Twitter handle). Used by the picker as a + /// last-resort identity hint after email and workspace. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub username: Option, } impl ComposioConnection { @@ -425,6 +445,9 @@ mod tests { toolkit: "slack".into(), status: status.into(), created_at: None, + account_email: None, + workspace: None, + username: None, }; assert!(conn.is_active(), "status {status:?} should be active"); } @@ -435,6 +458,9 @@ mod tests { toolkit: "slack".into(), status: status.into(), created_at: None, + account_email: None, + workspace: None, + username: None, }; assert!(!conn.is_active(), "status {status:?} should not be active"); } @@ -447,6 +473,9 @@ mod tests { toolkit: " Slack ".into(), status: "ACTIVE".into(), created_at: None, + account_email: None, + workspace: None, + username: None, }; assert_eq!(conn.normalized_toolkit(), "slack"); } @@ -494,6 +523,9 @@ mod tests { toolkit: "notion".into(), status: "PENDING".into(), created_at: None, + account_email: None, + workspace: None, + username: None, }; let s = serde_json::to_value(&conn).unwrap(); assert!( diff --git a/src/openhuman/heartbeat/planner/collectors.rs b/src/openhuman/heartbeat/planner/collectors.rs index 2ddae239c..831f65bc8 100644 --- a/src/openhuman/heartbeat/planner/collectors.rs +++ b/src/openhuman/heartbeat/planner/collectors.rs @@ -506,6 +506,9 @@ mod tests { toolkit: toolkit.to_string(), status: status.to_string(), created_at: None, + account_email: None, + workspace: None, + username: None, } } diff --git a/tests/composio_raw_coverage_e2e.rs b/tests/composio_raw_coverage_e2e.rs index 16bbcfd43..5714a0e1a 100644 --- a/tests/composio_raw_coverage_e2e.rs +++ b/tests/composio_raw_coverage_e2e.rs @@ -1303,6 +1303,9 @@ fn composio_types_roundtrip_connection_tool_trigger_and_history_shapes() { toolkit: "notion".into(), status: "FAILED".into(), created_at: None, + account_email: None, + workspace: None, + username: None, }; assert!(serde_json::to_value(default_connection) .unwrap()