fix(composio): enrich connection picker with cached account identity (#3356) (#3400)

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();

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<SyncReason> {
     }
 }
 
+/// 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;

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::Mutex<()>> =
+        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"
+    );
+}

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<String>,
+    /// 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<String>,
+    /// 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<String>,
+    /// 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<String>,
 }
 
 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!(

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,
         }
     }
 

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()