fix(memory-sources): hide inactive Composio connections in sources list

src/openhuman/memory_sources/reconcile.rs

@@ -12,12 +12,23 @@ use crate::openhuman::config::rpc as config_rpc;
 use crate::openhuman::memory_sources::registry;
 use crate::openhuman::memory_sources::types::{MemorySourceEntry, SourceKind};
 use crate::openhuman::memory_sync::composio;
+use std::collections::HashSet;
 
 /// Current version of the caps migration. Bump when the migration logic changes
 /// so installs that ran an earlier revision re-run it exactly once.
 const CURRENT_CAPS_MIGRATION_VERSION: u32 = 1;
 
-pub async fn ensure_composio_sources() {
+/// Reconcile active Composio connections into the memory sources registry and
+/// return the live active-connection set scanned this call.
+///
+/// Returns `Some(connection_ids)` — the `connection_id`s of every active sync
+/// target — when the live Composio scan **succeeded**, so callers (notably
+/// `rpc::list_rpc`) can filter the listing down to connections that are still
+/// active and dedupe identical rows. Returns `None` when the scan could not run
+/// (config load / network / auth failure); callers must treat `None` as "active
+/// set unavailable" and **not** hide any sources — an empty scan from a transient
+/// blip must never be read as "everything is inactive".
+pub async fn ensure_composio_sources() -> Option<HashSet<String>> {
     tracing::debug!("[memory_sources:reconcile] starting composio reconciliation");
 
     let config = match config_rpc::load_config_with_timeout().await {
@@ -27,7 +38,7 @@ pub async fn ensure_composio_sources() {
                 error = %e,
                 "[memory_sources:reconcile] failed to load config; skipping"
             );
-            return;
+            return None;
         }
     };
 
@@ -40,7 +51,7 @@ pub async fn ensure_composio_sources() {
                 error = %e,
                 "[memory_sources:reconcile] no composio sync targets available; skipping"
             );
-            return;
+            return None;
         }
     };
 
@@ -85,6 +96,12 @@ pub async fn ensure_composio_sources() {
             "[memory_sources:reconcile] caps migration failed (non-fatal, will retry next time)"
         );
     }
+
+    // The scan succeeded — surface the live active-connection set so the list
+    // path can hide rows for connections that are no longer active (re-auth /
+    // token expiry mints a fresh connection_id, stranding the old row) and
+    // collapse identical same-id duplicates.
+    Some(targets.iter().map(|t| t.connection_id.clone()).collect())
 }
 
 /// Apply conservative default caps in-place to every cap-less source.

src/openhuman/memory_sources/rpc.rs

@@ -20,9 +20,63 @@ pub async fn list_rpc() -> Result<RpcOutcome<ListResponse>, String> {
     // without waiting for a restart or for the connection_created hook
     // to fire (which only triggers on OAuth handoff, not on first launch
     // after the user previously connected something).
-    crate::openhuman::memory_sources::reconcile::ensure_composio_sources().await;
+    //
+    // The reconcile also hands back the live active-connection set it just
+    // scanned, which we reuse to hide Composio rows whose connection is no
+    // longer active (re-auth / token expiry leaves a stale row behind) and to
+    // collapse identical same-id duplicates from any reconcile race. This is a
+    // display-layer filter only — no row, setting, or ingested memory is
+    // removed; an inactive connection's row simply reappears once it re-activates.
+    let active = crate::openhuman::memory_sources::reconcile::ensure_composio_sources().await;
     let sources = registry::list_sources().await?;
-    Ok(RpcOutcome::new(ListResponse { sources }, vec![]))
+    let filtered = filter_to_active_composio_sources(sources, active.as_ref());
+    tracing::debug!(
+        active_known = active.is_some(),
+        active = active.as_ref().map(|a| a.len()).unwrap_or(0),
+        returned = filtered.len(),
+        "[memory_sources] list_rpc: filtered listing to active connections"
+    );
+    Ok(RpcOutcome::new(ListResponse { sources: filtered }, vec![]))
+}
+
+/// Filter the registry listing down to the live, deduplicated set of sources.
+///
+/// Composio sources are kept only when their `connection_id` is in `active`
+/// (the live active-connection set scanned by `ensure_composio_sources` this
+/// poll), collapsed to one row per `connection_id` so a non-atomic
+/// `upsert_composio_source` race can't surface identical duplicate rows.
+/// Non-Composio sources (folder / git / …) have no connection and are always
+/// shown.
+///
+/// `active == None` means the live scan was unavailable (config / network /
+/// auth failure). We must NOT read that as "everything is inactive" and hide
+/// every Composio source — so on `None` the list passes through untouched. This
+/// is hide-not-delete: the worst case is a stale row showing briefly until the
+/// next good scan, fully reversible. Pure (no I/O) so it is unit-tested directly.
+fn filter_to_active_composio_sources(
+    mut sources: Vec<MemorySourceEntry>,
+    active: Option<&std::collections::HashSet<String>>,
+) -> Vec<MemorySourceEntry> {
+    let Some(active) = active else {
+        // Scan unavailable — show everything rather than hiding all Composio rows.
+        return sources;
+    };
+    let mut seen = std::collections::HashSet::new();
+    sources.retain(|s| {
+        if s.kind != SourceKind::Composio {
+            return true; // no connection to reconcile against — always show
+        }
+        match s.connection_id.as_deref() {
+            // Active connection, first occurrence of this id → keep.
+            // Inactive (`!contains`) → hidden (RC-A); later duplicate of the
+            // same id (`!seen.insert`) → collapsed (RC-B).
+            Some(id) => active.contains(id) && seen.insert(id.to_string()),
+            // Malformed Composio row with no connection_id — keep it visible
+            // rather than silently dropping a user's source.
+            None => true,
+        }
+    });
+    sources
 }
 
 // ── Get ──
@@ -540,6 +594,152 @@ pub async fn apply_all_in_rpc() -> Result<RpcOutcome<AllInResponse>, String> {
     ))
 }
 
+#[cfg(test)]
+mod filter_tests {
+    use super::*;
+    use std::collections::HashSet;
+
+    fn composio_entry(id: &str, connection_id: &str) -> MemorySourceEntry {
+        MemorySourceEntry {
+            id: id.to_string(),
+            kind: SourceKind::Composio,
+            label: format!("Gmail · {connection_id}"),
+            enabled: true,
+            toolkit: Some("gmail".to_string()),
+            connection_id: Some(connection_id.to_string()),
+            path: None,
+            glob: None,
+            url: None,
+            branch: None,
+            paths: Vec::new(),
+            max_commits: None,
+            max_issues: None,
+            max_prs: None,
+            query: None,
+            since_days: None,
+            max_items: Some(100),
+            selector: None,
+            max_tokens_per_sync: None,
+            max_cost_per_sync_usd: None,
+            sync_depth_days: Some(30),
+        }
+    }
+
+    fn local_entry(id: &str) -> MemorySourceEntry {
+        MemorySourceEntry {
+            id: id.to_string(),
+            kind: SourceKind::Folder,
+            label: "Notes".to_string(),
+            enabled: true,
+            toolkit: None,
+            connection_id: None,
+            path: Some("/tmp/notes".to_string()),
+            glob: None,
+            url: None,
+            branch: None,
+            paths: Vec::new(),
+            max_commits: None,
+            max_issues: None,
+            max_prs: None,
+            query: None,
+            since_days: None,
+            max_items: None,
+            selector: None,
+            max_tokens_per_sync: None,
+            max_cost_per_sync_usd: None,
+            sync_depth_days: None,
+        }
+    }
+
+    fn active_set(ids: &[&str]) -> HashSet<String> {
+        ids.iter().map(|s| s.to_string()).collect()
+    }
+
+    /// RC-A: an inactive connection's row is hidden, only the active one shows —
+    /// but the input list is preserved (hide-not-delete; the filter never removes
+    /// entries from `config.memory_sources`, it only subtracts from the view).
+    #[test]
+    fn hides_inactive_connection_keeps_active() {
+        let sources = vec![
+            composio_entry("src_a", "conn_A"), // inactive
+            composio_entry("src_b", "conn_B"), // active
+        ];
+        let active = active_set(&["conn_B"]);
+        let out = filter_to_active_composio_sources(sources.clone(), Some(&active));
+
+        assert_eq!(out.len(), 1);
+        assert_eq!(out[0].connection_id.as_deref(), Some("conn_B"));
+        // Both rows still exist in the original list — nothing was deleted.
+        assert_eq!(sources.len(), 2);
+    }
+
+    /// RC-B: two rows with the same active connection_id collapse to one.
+    #[test]
+    fn dedupes_identical_connection_ids() {
+        let sources = vec![
+            composio_entry("src_1", "conn_B"),
+            composio_entry("src_2", "conn_B"),
+        ];
+        let active = active_set(&["conn_B"]);
+        let out = filter_to_active_composio_sources(sources, Some(&active));
+
+        assert_eq!(out.len(), 1, "identical-id rows must collapse to one");
+        assert_eq!(out[0].connection_id.as_deref(), Some("conn_B"));
+    }
+
+    /// A previously-inactive connection reappears (with its settings intact) once
+    /// it is back in the active set — confirming hiding is transient, not removal.
+    #[test]
+    fn reactivated_connection_reappears_with_settings() {
+        let sources = vec![composio_entry("src_a", "conn_A")];
+        let active = active_set(&["conn_A"]);
+        let out = filter_to_active_composio_sources(sources, Some(&active));
+
+        assert_eq!(out.len(), 1);
+        assert_eq!(out[0].connection_id.as_deref(), Some("conn_A"));
+        // Settings (caps) survive the round-trip — the row was never mutated.
+        assert_eq!(out[0].max_items, Some(100));
+        assert_eq!(out[0].sync_depth_days, Some(30));
+    }
+
+    /// Non-Composio sources have no connection and are always shown, regardless
+    /// of the active set.
+    #[test]
+    fn non_composio_sources_always_shown() {
+        let sources = vec![local_entry("src_local"), composio_entry("src_x", "conn_X")];
+        // Active set excludes the composio connection entirely.
+        let active = active_set(&[]);
+        let out = filter_to_active_composio_sources(sources, Some(&active));
+
+        assert_eq!(out.len(), 1);
+        assert_eq!(out[0].kind, SourceKind::Folder);
+    }
+
+    /// Safety: when the scan was unavailable (`None`), the list passes through
+    /// untouched — we must never hide every Composio source on a transient blip.
+    #[test]
+    fn none_active_set_shows_all() {
+        let sources = vec![
+            composio_entry("src_a", "conn_A"),
+            composio_entry("src_b", "conn_B"),
+            local_entry("src_local"),
+        ];
+        let out = filter_to_active_composio_sources(sources, None);
+        assert_eq!(out.len(), 3, "failed scan must show all sources, hide none");
+    }
+
+    /// A Composio row missing its connection_id is kept rather than silently
+    /// dropped, even when an active set is present.
+    #[test]
+    fn composio_without_connection_id_is_kept() {
+        let mut orphan = composio_entry("src_orphan", "conn_unused");
+        orphan.connection_id = None;
+        let active = active_set(&["conn_B"]);
+        let out = filter_to_active_composio_sources(vec![orphan], Some(&active));
+        assert_eq!(out.len(), 1);
+    }
+}
+
 #[cfg(test)]
 mod supported_toolkits_tests {
     use super::*;