From b8922c2e2898ffd475c2e598a0d9efb7a4109522 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Sat, 9 May 2026 07:29:18 +0530 Subject: [PATCH] feat(memory): self-identity tagging via Composio identity registry (#1365) (#1381) --- Cargo.lock | 2 +- app/src-tauri/Cargo.lock | 4 +- src/openhuman/composio/ops.rs | 104 +++ .../composio/providers/gmail/provider.rs | 39 +- .../composio/providers/notion/provider.rs | 37 +- src/openhuman/composio/providers/profile.rs | 861 ++++++++++++------ .../composio/providers/slack/provider.rs | 156 +++- src/openhuman/composio/schemas.rs | 30 + src/openhuman/memory/tree/score/store.rs | 70 +- src/openhuman/memory/tree/store.rs | 14 + src/openhuman/subconscious/prompt.rs | 21 + .../subconscious/situation_report/hotness.rs | 104 ++- .../subconscious/situation_report/mod.rs | 58 +- 13 files changed, 1112 insertions(+), 388 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 72ee1df3b..9381c0b38 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4520,7 +4520,7 @@ checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" [[package]] name = "openhuman" -version = "0.53.19" +version = "0.53.20" dependencies = [ "aes-gcm", "anyhow", diff --git a/app/src-tauri/Cargo.lock b/app/src-tauri/Cargo.lock index e09b8f885..fdbf85fba 100644 --- a/app/src-tauri/Cargo.lock +++ b/app/src-tauri/Cargo.lock @@ -4,7 +4,7 @@ version = 4 [[package]] name = "OpenHuman" -version = "0.53.19" +version = "0.53.20" dependencies = [ "anyhow", "async-trait", @@ -4551,7 +4551,7 @@ dependencies = [ [[package]] name = "openhuman" -version = "0.53.19" +version = "0.53.20" dependencies = [ "aes-gcm", "anyhow", diff --git a/src/openhuman/composio/ops.rs b/src/openhuman/composio/ops.rs index 8693c7e5a..caf5fb92f 100644 --- a/src/openhuman/composio/ops.rs +++ b/src/openhuman/composio/ops.rs @@ -470,6 +470,110 @@ pub async fn composio_get_user_profile( )) } +/// `openhuman.composio_refresh_all_identities` — re-fetch the user +/// profile for every active connection and persist via `identity_set`. +/// Used to populate kind-tagged `user_profile` rows on existing +/// connections after the #1365 schema rewrite without waiting for the +/// next periodic sync tick. +/// +/// Best-effort per connection: a failure on one toolkit does not abort +/// the others. Returns aggregate counts plus a per-connection trail in +/// the envelope messages. +pub async fn composio_refresh_all_identities( + config: &Config, +) -> OpResult> { + tracing::info!("[composio] rpc refresh_all_identities"); + let client = resolve_client(config)?; + let conns = client + .list_connections() + .await + .map_err(|e| format!("[composio] list_connections failed: {e:#}"))?; + + let mut report = RefreshIdentitiesReport::default(); + let mut messages: Vec = Vec::with_capacity(conns.connections.len() + 1); + + for conn in conns.connections { + if !conn.is_active() { + report.skipped_inactive += 1; + continue; + } + let toolkit = conn.toolkit.clone(); + let connection_id = conn.id.clone(); + + let Some(provider) = get_provider(&toolkit) else { + tracing::debug!( + toolkit = %toolkit, + connection_id = %connection_id, + "[composio] refresh_all_identities: no native provider — skipping" + ); + report.skipped_no_provider += 1; + messages.push(format!( + "{toolkit}/{connection_id}: skipped (no native provider)" + )); + continue; + }; + + let ctx = ProviderContext { + config: Arc::new(config.clone()), + client: client.clone(), + toolkit: toolkit.clone(), + connection_id: Some(connection_id.clone()), + }; + + match provider.fetch_user_profile(&ctx).await { + Ok(profile) => { + let rows = provider.identity_set(&profile); + report.refreshed += 1; + report.rows_written += rows; + tracing::debug!( + toolkit = %toolkit, + connection_id = %connection_id, + rows_written = rows, + "[composio] refresh_all_identities: identity_set ok" + ); + messages.push(format!("{toolkit}/{connection_id}: {rows} row(s)")); + } + Err(e) => { + report.failed += 1; + tracing::warn!( + toolkit = %toolkit, + connection_id = %connection_id, + error = %e, + "[composio] refresh_all_identities: fetch_user_profile failed" + ); + messages.push(format!("{toolkit}/{connection_id}: ERROR — {e}")); + } + } + } + + let summary = format!( + "composio: refreshed {ok}/{tried} active conn(s) — {rows} rows; \ + {fail} failed, {nopv} skipped (no provider), {inact} inactive", + ok = report.refreshed, + // `tried` is the count of active connections we actually scanned — + // include `skipped_no_provider` so the denominator covers the full + // active set, not just provider-backed ones (#1381 review). + tried = report.refreshed + report.failed + report.skipped_no_provider, + rows = report.rows_written, + fail = report.failed, + nopv = report.skipped_no_provider, + inact = report.skipped_inactive, + ); + let mut envelope = vec![summary]; + envelope.extend(messages); + Ok(RpcOutcome::new(report, envelope)) +} + +/// Aggregate result of [`composio_refresh_all_identities`]. +#[derive(Debug, Default, Clone, serde::Serialize, serde::Deserialize)] +pub struct RefreshIdentitiesReport { + pub refreshed: usize, + pub failed: usize, + pub skipped_no_provider: usize, + pub skipped_inactive: usize, + pub rows_written: usize, +} + /// `openhuman.composio_sync` — run a sync pass for a connected account /// by dispatching to the toolkit's registered provider. `reason` is /// `"manual"` by default; the periodic scheduler passes `"periodic"` diff --git a/src/openhuman/composio/providers/gmail/provider.rs b/src/openhuman/composio/providers/gmail/provider.rs index 0bf1e4db5..01b032da2 100644 --- a/src/openhuman/composio/providers/gmail/provider.rs +++ b/src/openhuman/composio/providers/gmail/provider.rs @@ -124,37 +124,20 @@ impl ComposioProvider for GmailProvider { return Err(format!("[composio:gmail] {ACTION_GET_PROFILE}: {err}")); } + // `data` is the inner Composio payload — paths here are relative + // to it. (The previous `data.*` paths were dead — `pick_str` + // does dotted-path traversal, so `data.emailAddress` looked for + // a nested `data.data.emailAddress` that never exists.) let data = &resp.data; - let email = pick_str( - data, - &[ - "data.emailAddress", - "data.email", - "emailAddress", - "email", - "data.profile.emailAddress", - ], - ); - let display_name = pick_str( - data, - &[ - "data.name", - "data.profile.name", - "name", - "displayName", - "data.displayName", - ], - ) - .or_else(|| email.clone()); + let email = pick_str(data, &["emailAddress", "email", "profile.emailAddress"]); + // Don't fall back to the email when no name is returned — that + // produces duplicated `display_name == email` rows in the + // identity registry (#1365). Gmail's `GMAIL_GET_PROFILE` action + // doesn't return a name today, so this stays None. + let display_name = pick_str(data, &["name", "profile.name", "displayName"]); let profile_url = pick_str( data, - &[ - "data.profileUrl", - "data.profile_url", - "data.profile.url", - "profileUrl", - "profile_url", - ], + &["display_url", "profileUrl", "profile_url", "profile.url"], ); let profile = ProviderUserProfile { diff --git a/src/openhuman/composio/providers/notion/provider.rs b/src/openhuman/composio/providers/notion/provider.rs index e59e1eb5f..6c6b42578 100644 --- a/src/openhuman/composio/providers/notion/provider.rs +++ b/src/openhuman/composio/providers/notion/provider.rs @@ -101,43 +101,28 @@ impl ComposioProvider for NotionProvider { return Err(format!("[composio:notion] {ACTION_GET_ABOUT_ME}: {err}")); } + // `data` is already the inner Composio response payload — paths + // here are relative to it. For bot-token connections the + // top-level `name` is the *integration's* name (e.g. "Composio"), + // and the actual owning user lives at `bot.owner.user.*`. Probe + // the bot-owner paths first so identity reflects the user (#1365). let data = &resp.data; - let display_name = pick_str( - data, - &[ - "data.name", - "data.user.name", - "name", - "data.bot.owner.user.name", - ], - ); + let display_name = pick_str(data, &["bot.owner.user.name", "user.name", "name"]); let email = pick_str( data, &[ - "data.person.email", - "data.user.person.email", + "bot.owner.user.person.email", + "user.person.email", "person.email", "email", ], ); - let username = pick_str( - data, - &["data.bot.owner.user.id", "data.id", "id", "data.user.id"], - ); + let username = pick_str(data, &["bot.owner.user.id", "user.id", "id"]); let avatar_url = pick_str( data, - &["data.avatar_url", "data.user.avatar_url", "avatar_url"], - ); - let profile_url = pick_str( - data, - &[ - "data.url", - "data.profile_url", - "data.profile.url", - "url", - "profile_url", - ], + &["bot.owner.user.avatar_url", "user.avatar_url", "avatar_url"], ); + let profile_url = pick_str(data, &["url", "profile_url", "profile.url"]); Ok(ProviderUserProfile { toolkit: "notion".to_string(), diff --git a/src/openhuman/composio/providers/profile.rs b/src/openhuman/composio/providers/profile.rs index deab58e30..396f8149d 100644 --- a/src/openhuman/composio/providers/profile.rs +++ b/src/openhuman/composio/providers/profile.rs @@ -1,40 +1,135 @@ -//! Profile persistence bridge — maps [`ProviderUserProfile`] fields -//! into the local `user_profile` facet table so provider-sourced -//! identity data (display name, email, username, avatar) accumulates -//! alongside conversation-extracted preferences. +//! Profile persistence — maps [`ProviderUserProfile`] (and provider-specific +//! `extras`) into [`IdentityKind`]-tagged facet rows so the self-identity +//! matcher can join directly against the memory tree's `EntityKind` and the +//! structural sender field on chunks. //! -//! Each non-`None` field becomes a [`FacetType::Context`] facet keyed -//! as `skill:{toolkit}:{identifier}:{field}`. Confidence is set to 0.95 because -//! this data comes directly from the upstream provider API — it's -//! authoritative, not inferred from conversation. +//! Schema: `user_profile.facet_type='skill'`, +//! `key = "skill:{toolkit}:{conn_id}:{identity_kind}"`, `value` = +//! canonicalized identifier. Confidence is set per-kind so the matcher can +//! refuse to auto-promote weak signals (display_name) to `is_self`. //! -//! Callers are expected to invoke [`persist_provider_profile`] after -//! every successful `fetch_user_profile` call — from -//! `on_connection_created`, periodic syncs, and the -//! `composio_get_user_profile` RPC op. +//! One [`ProviderUserProfile`] expands to multiple rows — including +//! identifiers carried in `extras` that the previous fixed-fields shape +//! dropped on the floor (e.g. Slack screen-name handle). +//! +//! Callers invoke [`persist_provider_profile`] after every successful +//! `fetch_user_profile` call — from `on_connection_created`, periodic syncs, +//! and the `composio_get_user_profile` / `composio_refresh_all_identities` +//! RPC ops. use super::ProviderUserProfile; use crate::openhuman::memory::store::profile::{self, FacetType}; use rusqlite::params; +use serde_json::Value; use std::collections::BTreeMap; -/// Confidence level assigned to provider-sourced profile data. -/// -/// This is higher than conversation-inferred facets (typically 0.5–0.7) -/// because the data comes directly from the upstream provider API. -const PROVIDER_CONFIDENCE: f64 = 0.95; +// ──────────────────────────────────────────────────────────────────────── +// IdentityKind — the matching axis +// ──────────────────────────────────────────────────────────────────────── -/// Persist the non-`None` fields of a [`ProviderUserProfile`] into the -/// local `user_profile` facet table. -/// -/// Returns the number of facets written (0–4). Silently returns 0 if -/// the global memory client is not yet initialised (user not signed in, -/// startup race, etc.) — callers should treat that as non-fatal. +/// Shape of an identifier persisted against a connection. Mirrors the +/// matching dimensions of the memory tree's +/// `crate::openhuman::memory::tree::score::extract::EntityKind` so the +/// self-check is a direct `(toolkit, kind, value)` lookup. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum IdentityKind { + /// Platform-canonical immutable id — Slack `U123ABC`, Notion UUID. + UserId, + Email, + /// `@`-style screen name, canonicalised without the leading `@`. + Handle, + /// E.164 phone number. + Phone, + /// Human display label. Weak signal — never auto-promotes to is_self. + DisplayName, + /// Not for matching; kept for UI / prompt rendering. + AvatarUrl, + /// Not for matching; kept for UI / prompt rendering. + ProfileUrl, +} + +impl IdentityKind { + pub fn as_str(self) -> &'static str { + match self { + Self::UserId => "user_id", + Self::Email => "email", + Self::Handle => "handle", + Self::Phone => "phone", + Self::DisplayName => "display_name", + Self::AvatarUrl => "avatar_url", + Self::ProfileUrl => "profile_url", + } + } + + pub fn parse(s: &str) -> Option { + Some(match s { + "user_id" => Self::UserId, + "email" => Self::Email, + "handle" => Self::Handle, + "phone" => Self::Phone, + "display_name" => Self::DisplayName, + "avatar_url" => Self::AvatarUrl, + "profile_url" => Self::ProfileUrl, + _ => return None, + }) + } + + /// Confidence the matcher records on the row. Hard kinds auto-promote + /// a chunk to `is_self`; weak kinds require corroboration. + pub fn confidence(self) -> f64 { + match self { + Self::UserId | Self::Phone => 1.00, + Self::Email => 0.95, + Self::Handle => 0.70, + Self::DisplayName => 0.40, + Self::AvatarUrl | Self::ProfileUrl => 0.50, + } + } + + /// True if this kind is a real identity signal worth running through + /// the matcher (vs. UI-only fields). + pub fn is_matchable(self) -> bool { + matches!( + self, + Self::UserId | Self::Email | Self::Handle | Self::Phone | Self::DisplayName + ) + } +} + +/// Canonicalize a raw value for storage and lookup. The same routine runs +/// on the entity side at match time, so equality of canonical forms is the +/// matcher's only test — no `COLLATE NOCASE`, no per-call lowercasing. +pub fn canonicalize(kind: IdentityKind, raw: &str) -> Option { + let trimmed = raw.trim(); + if trimmed.is_empty() { + return None; + } + Some(match kind { + IdentityKind::Email => trimmed.to_lowercase(), + IdentityKind::Handle => trimmed.trim_start_matches('@').to_lowercase(), + IdentityKind::Phone => trimmed + .chars() + .filter(|c| c.is_ascii_digit() || *c == '+') + .collect(), + IdentityKind::DisplayName => trimmed.split_whitespace().collect::>().join(" "), + IdentityKind::UserId | IdentityKind::AvatarUrl | IdentityKind::ProfileUrl => { + trimmed.to_string() + } + }) +} + +// ──────────────────────────────────────────────────────────────────────── +// Persist +// ──────────────────────────────────────────────────────────────────────── + +/// Persist a provider profile as one facet row per (kind, value). Returns +/// the number of rows written. Silently no-ops if the memory client isn't +/// ready (startup race / unauthenticated CLI). pub fn persist_provider_profile(profile: &ProviderUserProfile) -> usize { let Some(client) = crate::openhuman::memory::global::client_if_ready() else { tracing::debug!( toolkit = %profile.toolkit, - "[composio:profile] memory client not ready, skipping profile persist" + "[composio:profile] memory client not ready, skipping persist" ); return 0; }; @@ -49,38 +144,27 @@ pub fn persist_provider_profile(profile: &ProviderUserProfile) -> usize { .filter(|v| !v.is_empty()) .unwrap_or_else(|| "default".to_string()); - let fields: &[(&str, &Option)] = &[ - ("display_name", &profile.display_name), - ("email", &profile.email), - ("username", &profile.username), - ("avatar_url", &profile.avatar_url), - ("profile_url", &profile.profile_url), - ]; + let rows = expand_identity_rows(&toolkit, profile); let mut written = 0usize; - for (field_name, value) in fields { - let Some(val) = value else { continue }; - if val.trim().is_empty() { - continue; - } - - let key = format!("skill:{toolkit}:{identifier}:{field_name}"); - let facet_id = format!("skill-{toolkit}-{identifier}-{field_name}"); + for (kind, value) in rows { + let key = format!("skill:{toolkit}:{identifier}:{}", kind.as_str()); + let facet_id = format!("skill-{toolkit}-{identifier}-{}", kind.as_str()); if let Err(e) = profile::profile_upsert( &conn, &facet_id, &FacetType::Skill, &key, - val, - PROVIDER_CONFIDENCE, - None, // no source segment — this comes from a provider, not a conversation + &value, + kind.confidence(), + None, now, ) { tracing::warn!( toolkit = %toolkit, identifier = %identifier, - field = %field_name, + kind = kind.as_str(), error = %e, "[composio:profile] profile_upsert failed (non-fatal)" ); @@ -93,25 +177,83 @@ pub fn persist_provider_profile(profile: &ProviderUserProfile) -> usize { tracing::debug!( toolkit = %toolkit, identifier = %identifier, - facets_written = written, - "[composio:profile] persisted provider profile facets" + rows_written = written, + "[composio:profile] persisted identity rows" ); } written } -#[derive(Debug, Clone, PartialEq, Eq)] +/// Expand a [`ProviderUserProfile`] (and provider-specific `extras`) into +/// the canonical (kind, value) rows. **All per-toolkit quirks live here**; +/// the matcher only sees normalized tuples. +fn expand_identity_rows( + toolkit: &str, + profile: &ProviderUserProfile, +) -> Vec<(IdentityKind, String)> { + let mut rows: Vec<(IdentityKind, String)> = Vec::new(); + let mut push = |kind: IdentityKind, raw: Option<&str>| { + if let Some(v) = raw.and_then(|s| canonicalize(kind, s)) { + rows.push((kind, v)); + } + }; + + push(IdentityKind::DisplayName, profile.display_name.as_deref()); + push(IdentityKind::Email, profile.email.as_deref()); + push(IdentityKind::AvatarUrl, profile.avatar_url.as_deref()); + push(IdentityKind::ProfileUrl, profile.profile_url.as_deref()); + + match toolkit { + "slack" => { + // After the auth.test + users.info fix in slack/provider.rs: + // profile.username == Slack user_id (e.g. U123ABC) + // extras.handle == Slack screen_name (e.g. "cyrus") + // extras.team_* → workspace context, not identity + push(IdentityKind::UserId, profile.username.as_deref()); + push(IdentityKind::Handle, json_str(&profile.extras, "handle")); + } + "notion" => { + // Notion's `username` is the user UUID + // (`data.bot.owner.user.id` per notion/provider.rs). + push(IdentityKind::UserId, profile.username.as_deref()); + } + "gmail" => { + // Email + display_name only — no platform user_id worth matching. + } + _ => { + // Unknown toolkit: best-effort. If `username` is set treat it + // as a handle so weak-match logic (medium confidence) applies. + push(IdentityKind::Handle, profile.username.as_deref()); + } + } + + rows +} + +fn json_str<'a>(v: &'a Value, key: &str) -> Option<&'a str> { + v.get(key).and_then(|x| x.as_str()) +} + +// ──────────────────────────────────────────────────────────────────────── +// Read paths +// ──────────────────────────────────────────────────────────────────────── + +#[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct ConnectedIdentity { pub source: String, pub identifier: String, pub display_name: Option, pub email: Option, - pub username: Option, + pub handle: Option, + pub phone: Option, + pub user_id: Option, + pub avatar_url: Option, pub profile_url: Option, } -/// Load provider-sourced identity fragments from profile facets and -/// group them by `source + identifier`. +/// Load all provider-sourced identities, grouped by `(source, conn_id)`. +/// Rows whose last segment is not a known [`IdentityKind`] are silently +/// skipped — that includes legacy `username` rows from before the rewrite. pub fn load_connected_identities() -> Vec { let Some(client) = crate::openhuman::memory::global::client_if_ready() else { tracing::debug!("[composio:profile] load_connected_identities: memory client not ready"); @@ -119,11 +261,10 @@ pub fn load_connected_identities() -> Vec { }; let conn = client.profile_conn(); let facets = match profile::profile_facets_by_type(&conn, &FacetType::Skill) { - Ok(facets) => facets, + Ok(f) => f, Err(error) => { tracing::warn!( error = %error, - facet_type = ?FacetType::Skill, "[composio:profile] load_connected_identities: profile_facets_by_type failed" ); return Vec::new(); @@ -132,7 +273,10 @@ pub fn load_connected_identities() -> Vec { let mut grouped: BTreeMap<(String, String), ConnectedIdentity> = BTreeMap::new(); for facet in facets { - let Some((source, identifier, field)) = parse_skill_identity_key(&facet.key) else { + let Some((source, identifier, kind_str)) = parse_skill_identity_key(&facet.key) else { + continue; + }; + let Some(kind) = IdentityKind::parse(&kind_str) else { continue; }; let entry = grouped @@ -140,61 +284,121 @@ pub fn load_connected_identities() -> Vec { .or_insert_with(|| ConnectedIdentity { source, identifier, - display_name: None, - email: None, - username: None, - profile_url: None, + ..Default::default() }); - match field.as_str() { - "display_name" => entry.display_name = Some(facet.value.clone()), - "email" => entry.email = Some(facet.value.clone()), - "username" => entry.username = Some(facet.value.clone()), - "profile_url" => entry.profile_url = Some(facet.value.clone()), - _ => {} + match kind { + IdentityKind::DisplayName => entry.display_name = Some(facet.value), + IdentityKind::Email => entry.email = Some(facet.value), + IdentityKind::Handle => entry.handle = Some(facet.value), + IdentityKind::Phone => entry.phone = Some(facet.value), + IdentityKind::UserId => entry.user_id = Some(facet.value), + IdentityKind::AvatarUrl => entry.avatar_url = Some(facet.value), + IdentityKind::ProfileUrl => entry.profile_url = Some(facet.value), } } grouped.into_values().collect() } -/// Render a compact markdown section for prompt injection. +/// Direct self-check for the entity matcher and the chunk-build hook. +/// Returns true if any connection of `toolkit` has a row with this +/// `(kind, value)` after canonicalization. Non-matchable kinds +/// (avatar_url, profile_url) always return false. +pub fn is_self_identity(toolkit: &str, kind: IdentityKind, raw_value: &str) -> bool { + if !kind.is_matchable() { + return false; + } + let Some(canonical) = canonicalize(kind, raw_value) else { + return false; + }; + let Some(client) = crate::openhuman::memory::global::client_if_ready() else { + return false; + }; + let conn = client.profile_conn(); + let conn = conn.lock(); + + let key_pattern = format!("skill:{}:%:{}", normalize_token(toolkit), kind.as_str()); + conn.query_row( + "SELECT 1 FROM user_profile + WHERE facet_type = 'skill' + AND key LIKE ?1 + AND value = ?2 + LIMIT 1", + params![key_pattern, canonical], + |_| Ok(()), + ) + .is_ok() +} + +/// Cross-toolkit variant — matches against every connected provider's +/// rows of this kind. Used for marking memory-tree entity rows: an email +/// in a Slack message that matches the user's Gmail address is still +/// "me," regardless of which source produced the chunk. +pub fn is_self_identity_any_toolkit(kind: IdentityKind, raw_value: &str) -> bool { + if !kind.is_matchable() { + return false; + } + let Some(canonical) = canonicalize(kind, raw_value) else { + return false; + }; + let Some(client) = crate::openhuman::memory::global::client_if_ready() else { + return false; + }; + let conn = client.profile_conn(); + let conn = conn.lock(); + + let key_pattern = format!("skill:%:%:{}", kind.as_str()); + conn.query_row( + "SELECT 1 FROM user_profile + WHERE facet_type = 'skill' + AND key LIKE ?1 + AND value = ?2 + LIMIT 1", + params![key_pattern, canonical], + |_| Ok(()), + ) + .is_ok() +} + +/// Render a compact section for prompt injection. Skips `user_id` (not +/// human-readable), prefixes `handle` with `@`. pub fn render_connected_identities_section(identities: &[ConnectedIdentity]) -> String { if identities.is_empty() { return String::new(); } let mut out = String::from("## Connected Identities\n\n"); - for identity in identities { - let mut fields = Vec::new(); - if let Some(display_name) = identity.display_name.as_deref() { - let cleaned = sanitize_prompt_value(display_name); - if !cleaned.is_empty() { - fields.push(cleaned); + for id in identities { + let mut fields = Vec::::new(); + if let Some(v) = id.display_name.as_deref() { + let v = sanitize_prompt_value(v); + if !v.is_empty() { + fields.push(v); } } - if let Some(email) = identity.email.as_deref() { - let cleaned = sanitize_prompt_value(email); - if !cleaned.is_empty() { - fields.push(cleaned); + if let Some(v) = id.email.as_deref() { + let v = sanitize_prompt_value(v); + if !v.is_empty() { + fields.push(v); } } - if let Some(username) = identity.username.as_deref() { - let cleaned = sanitize_prompt_value(username); - if !cleaned.is_empty() { - fields.push(format!("@{cleaned}")); + if let Some(v) = id.handle.as_deref() { + let v = sanitize_prompt_value(v); + if !v.is_empty() { + fields.push(format!("@{v}")); } } - if let Some(profile_url) = identity.profile_url.as_deref() { - let cleaned = sanitize_prompt_value(profile_url); - if !cleaned.is_empty() { - fields.push(cleaned); + if let Some(v) = id.profile_url.as_deref() { + let v = sanitize_prompt_value(v); + if !v.is_empty() { + fields.push(v); } } if fields.is_empty() { continue; } - let identifier = sanitize_prompt_value(&identity.identifier); + let identifier = sanitize_prompt_value(&id.identifier); out.push_str(&format!( "- {} ({}): {}\n", - title_case(&identity.source), + title_case(&id.source), identifier, fields.join(" | ") )); @@ -205,20 +409,63 @@ pub fn render_connected_identities_section(identities: &[ConnectedIdentity]) -> out } +/// Delete every row for a `(source, conn_id)` pair — used on disconnect. +pub fn delete_connected_identity_facets(source: &str, identifier: &str) -> usize { + // `persist_provider_profile` writes keys with `normalize_token`-applied + // segments; compare against the same normalized form here so a caller + // passing the raw toolkit/connection_id still matches stored rows + // (otherwise rows would survive disconnect and the user-tagger would + // keep treating the removed account as the user — #1381 review). + let source = normalize_token(source); + let identifier = normalize_token(identifier); + let Some(client) = crate::openhuman::memory::global::client_if_ready() else { + tracing::debug!( + source = %source, + identifier = %identifier, + "[composio:profile] delete_connected_identity_facets: memory client not ready" + ); + return 0; + }; + let conn = client.profile_conn(); + let Ok(facets) = profile::profile_facets_by_type(&conn, &FacetType::Skill) else { + return 0; + }; + let mut deleted = 0usize; + for facet in facets { + let Some((s, i, _kind)) = parse_skill_identity_key(&facet.key) else { + continue; + }; + if s == source && i == identifier { + let conn_guard = conn.lock(); + if conn_guard + .execute( + "DELETE FROM user_profile WHERE facet_id = ?1", + params![facet.facet_id], + ) + .unwrap_or(0) + > 0 + { + deleted += 1; + } + } + } + deleted +} + +// ──────────────────────────────────────────────────────────────────────── +// Helpers +// ──────────────────────────────────────────────────────────────────────── + fn parse_skill_identity_key(key: &str) -> Option<(String, String, String)> { let mut parts = key.split(':'); let prefix = parts.next()?; let source = parts.next()?; let identifier = parts.next()?; - let field = parts.next()?; + let kind = parts.next()?; if prefix != "skill" || parts.next().is_some() { return None; } - Some(( - source.to_string(), - identifier.to_string(), - field.to_string(), - )) + Some((source.to_string(), identifier.to_string(), kind.to_string())) } fn normalize_token(raw: &str) -> String { @@ -247,43 +494,6 @@ fn sanitize_prompt_value(raw: &str) -> String { replaced.split_whitespace().collect::>().join(" ") } -/// Delete provider-sourced skill facets for one connection identifier. -pub fn delete_connected_identity_facets(source: &str, identifier: &str) -> usize { - let Some(client) = crate::openhuman::memory::global::client_if_ready() else { - tracing::debug!( - source = %source, - identifier = %identifier, - "[composio:profile] delete_connected_identity_facets: memory client not ready" - ); - return 0; - }; - let conn = client.profile_conn(); - let Ok(facets) = profile::profile_facets_by_type(&conn, &FacetType::Skill) else { - return 0; - }; - let mut deleted = 0usize; - for facet in facets { - let Some((facet_source, facet_identifier, _field)) = parse_skill_identity_key(&facet.key) - else { - continue; - }; - if facet_source == source && facet_identifier == identifier { - let conn_guard = conn.lock(); - if conn_guard - .execute( - "DELETE FROM user_profile WHERE facet_id = ?1", - params![facet.facet_id], - ) - .unwrap_or(0) - > 0 - { - deleted += 1; - } - } - } - deleted -} - fn now_secs() -> f64 { use std::time::{SystemTime, UNIX_EPOCH}; SystemTime::now() @@ -292,12 +502,17 @@ fn now_secs() -> f64 { .unwrap_or(0.0) } +// ──────────────────────────────────────────────────────────────────────── +// Tests +// ──────────────────────────────────────────────────────────────────────── + #[cfg(test)] mod tests { use super::*; use crate::openhuman::memory::store::profile::{profile_load_all, PROFILE_INIT_SQL}; use parking_lot::Mutex; use rusqlite::Connection; + use serde_json::json; use std::sync::Arc; fn setup_db() -> Arc> { @@ -306,169 +521,301 @@ mod tests { Arc::new(Mutex::new(conn)) } - /// Directly exercise `profile_upsert` with provider-style keys to - /// verify the facet schema and conflict resolution work end-to-end. + // ── IdentityKind ─────────────────────────────────────────────── + #[test] - fn provider_fields_map_to_facets() { - let conn = setup_db(); - let now = 1000.0; - - // Simulate what persist_provider_profile does internally. - profile::profile_upsert( - &conn, - "skill-gmail-conn-1-email", - &FacetType::Skill, - "skill:gmail:conn-1:email", - "user@example.com", - PROVIDER_CONFIDENCE, - None, - now, - ) - .unwrap(); - - profile::profile_upsert( - &conn, - "skill-gmail-conn-1-display_name", - &FacetType::Skill, - "skill:gmail:conn-1:display_name", - "Jane Doe", - PROVIDER_CONFIDENCE, - None, - now, - ) - .unwrap(); - - let facets = profile_load_all(&conn).unwrap(); - assert_eq!(facets.len(), 2); - - let email_facet = facets.iter().find(|f| f.key == "skill:gmail:conn-1:email"); - assert!(email_facet.is_some()); - let email_facet = email_facet.unwrap(); - assert_eq!(email_facet.value, "user@example.com"); - assert!((email_facet.confidence - PROVIDER_CONFIDENCE).abs() < f64::EPSILON); - assert_eq!(email_facet.evidence_count, 1); + fn identity_kind_round_trips_through_str() { + for kind in [ + IdentityKind::UserId, + IdentityKind::Email, + IdentityKind::Handle, + IdentityKind::Phone, + IdentityKind::DisplayName, + IdentityKind::AvatarUrl, + IdentityKind::ProfileUrl, + ] { + assert_eq!(IdentityKind::parse(kind.as_str()), Some(kind)); + } } #[test] - fn repeated_persist_increments_evidence() { + fn identity_kind_parse_rejects_unknown() { + assert_eq!(IdentityKind::parse("username"), None); + assert_eq!(IdentityKind::parse(""), None); + assert_eq!(IdentityKind::parse("UserId"), None); + } + + #[test] + fn matchable_kinds_exclude_url_fields() { + assert!(IdentityKind::UserId.is_matchable()); + assert!(IdentityKind::Email.is_matchable()); + assert!(IdentityKind::Handle.is_matchable()); + assert!(IdentityKind::Phone.is_matchable()); + assert!(IdentityKind::DisplayName.is_matchable()); + assert!(!IdentityKind::AvatarUrl.is_matchable()); + assert!(!IdentityKind::ProfileUrl.is_matchable()); + } + + #[test] + fn confidence_orders_hard_above_weak() { + assert!(IdentityKind::UserId.confidence() > IdentityKind::Email.confidence()); + assert!(IdentityKind::Email.confidence() > IdentityKind::Handle.confidence()); + assert!(IdentityKind::Handle.confidence() > IdentityKind::DisplayName.confidence()); + } + + // ── canonicalize ────────────────────────────────────────────── + + #[test] + fn canonicalize_email_lowercases_and_trims() { + assert_eq!( + canonicalize(IdentityKind::Email, " Cyrus@Example.COM "), + Some("cyrus@example.com".to_string()) + ); + } + + #[test] + fn canonicalize_handle_strips_at_and_lowercases() { + assert_eq!( + canonicalize(IdentityKind::Handle, "@Cyrus"), + Some("cyrus".to_string()) + ); + assert_eq!( + canonicalize(IdentityKind::Handle, "cyrus"), + Some("cyrus".to_string()) + ); + } + + #[test] + fn canonicalize_phone_keeps_only_digits_and_plus() { + assert_eq!( + canonicalize(IdentityKind::Phone, "+1 (555) 123-4567"), + Some("+15551234567".to_string()) + ); + } + + #[test] + fn canonicalize_display_name_collapses_whitespace() { + assert_eq!( + canonicalize(IdentityKind::DisplayName, " Cyrus Smith "), + Some("Cyrus Smith".to_string()) + ); + } + + #[test] + fn canonicalize_user_id_preserved_as_is() { + // Slack user_ids are case-sensitive; do not lowercase. + assert_eq!( + canonicalize(IdentityKind::UserId, "U123ABC"), + Some("U123ABC".to_string()) + ); + } + + #[test] + fn canonicalize_empty_returns_none() { + assert_eq!(canonicalize(IdentityKind::Email, ""), None); + assert_eq!(canonicalize(IdentityKind::Email, " "), None); + } + + // ── expand_identity_rows ────────────────────────────────────── + + fn fixture_profile( + toolkit: &str, + username: Option<&str>, + extras: Value, + ) -> ProviderUserProfile { + ProviderUserProfile { + toolkit: toolkit.into(), + connection_id: Some("conn-1".into()), + display_name: Some("Cyrus Smith".into()), + email: Some("cyrus@example.com".into()), + username: username.map(str::to_string), + avatar_url: None, + profile_url: Some("https://example.com/cyrus".into()), + extras, + } + } + + #[test] + fn expand_slack_promotes_username_to_user_id_and_extras_handle() { + let p = fixture_profile("slack", Some("U123ABC"), json!({ "handle": "cyrus" })); + let rows = expand_identity_rows("slack", &p); + + assert!(rows.contains(&(IdentityKind::UserId, "U123ABC".to_string()))); + assert!(rows.contains(&(IdentityKind::Handle, "cyrus".to_string()))); + assert!(rows.contains(&(IdentityKind::Email, "cyrus@example.com".to_string()))); + assert!(rows.contains(&(IdentityKind::DisplayName, "Cyrus Smith".to_string()))); + assert!(rows.contains(&( + IdentityKind::ProfileUrl, + "https://example.com/cyrus".to_string() + ))); + } + + #[test] + fn expand_gmail_skips_username_with_no_user_id_concept() { + let p = fixture_profile("gmail", None, Value::Null); + let rows = expand_identity_rows("gmail", &p); + + assert!(rows + .iter() + .all(|(k, _)| !matches!(k, IdentityKind::UserId | IdentityKind::Handle))); + assert!(rows.contains(&(IdentityKind::Email, "cyrus@example.com".to_string()))); + } + + #[test] + fn expand_notion_treats_username_as_user_id() { + let p = fixture_profile( + "notion", + Some("f3c1a8e2-b9b7-4a8d-9d5b-31a2e9f44e2f"), + Value::Null, + ); + let rows = expand_identity_rows("notion", &p); + + assert!(rows.contains(&( + IdentityKind::UserId, + "f3c1a8e2-b9b7-4a8d-9d5b-31a2e9f44e2f".to_string() + ))); + } + + #[test] + fn expand_unknown_toolkit_falls_back_to_handle() { + let p = fixture_profile("hypothetical", Some("alice"), Value::Null); + let rows = expand_identity_rows("hypothetical", &p); + + assert!(rows.contains(&(IdentityKind::Handle, "alice".to_string()))); + } + + #[test] + fn expand_empty_profile_emits_nothing_matchable() { + let p = ProviderUserProfile { + toolkit: "gmail".into(), + connection_id: Some("c-1".into()), + display_name: None, + email: None, + username: None, + avatar_url: None, + profile_url: None, + extras: Value::Null, + }; + let rows = expand_identity_rows("gmail", &p); + assert!(rows.is_empty()); + } + + // ── upsert wiring (uses the underlying profile_upsert directly) ─ + + #[test] + fn upsert_writes_kind_tagged_key() { let conn = setup_db(); - // First write. profile::profile_upsert( &conn, - "skill-notion-default-email", + "skill-slack-conn-1-user_id", &FacetType::Skill, - "skill:notion:default:email", - "user@workspace.com", - PROVIDER_CONFIDENCE, + "skill:slack:conn-1:user_id", + "U123ABC", + IdentityKind::UserId.confidence(), None, 1000.0, ) .unwrap(); - // Second write — same key, same value (periodic re-sync). - profile::profile_upsert( - &conn, - "skill-notion-default-email-2", - &FacetType::Skill, - "skill:notion:default:email", - "user@workspace.com", - PROVIDER_CONFIDENCE, - None, - 2000.0, - ) - .unwrap(); + let facets = profile_load_all(&conn).unwrap(); + let row = facets + .iter() + .find(|f| f.key == "skill:slack:conn-1:user_id") + .expect("row exists"); + assert_eq!(row.value, "U123ABC"); + assert!((row.confidence - 1.00).abs() < f64::EPSILON); + } + + #[test] + fn upsert_repeated_increments_evidence() { + let conn = setup_db(); + + for now in [1000.0, 2000.0] { + profile::profile_upsert( + &conn, + "skill-notion-default-email", + &FacetType::Skill, + "skill:notion:default:email", + "user@workspace.com", + IdentityKind::Email.confidence(), + None, + now, + ) + .unwrap(); + } let facets = profile_load_all(&conn).unwrap(); - assert_eq!(facets.len(), 1, "duplicate key should merge into one row"); + assert_eq!(facets.len(), 1); assert_eq!(facets[0].evidence_count, 2); } - #[test] - fn now_secs_returns_recent_unix_seconds() { - // Sanity check: the helper just wraps SystemTime::now() into f64. - let t = now_secs(); - assert!(t > 1_000_000_000.0, "expected unix epoch seconds, got {t}"); - } + // ── parse_skill_identity_key ────────────────────────────────── #[test] - fn persist_provider_profile_returns_zero_when_memory_client_not_ready() { - // The global memory client is gated behind login; in the test - // binary it may or may not be initialised depending on test - // ordering. We just exercise the entrypoint to cover the - // early-return branch — if the global IS ready we accept the - // returned count without further assertions. - let profile = ProviderUserProfile { - toolkit: "gmail".into(), - connection_id: Some("c-1".into()), - display_name: Some("Jane".into()), - email: Some("jane@example.com".into()), - username: None, - avatar_url: None, - profile_url: None, - extras: serde_json::Value::Null, - }; - let _written = persist_provider_profile(&profile); - } - - #[test] - fn empty_fields_are_skipped() { - let profile = ProviderUserProfile { - toolkit: "gmail".into(), - connection_id: Some("conn-1".into()), - display_name: Some("Jane".into()), - email: None, - username: Some("".into()), // empty string — should be skipped - avatar_url: Some(" ".into()), // whitespace — should be skipped - profile_url: Some("https://mail.google.com".into()), - extras: serde_json::Value::Null, - }; - - // We can't call persist_provider_profile directly without the - // global memory singleton, but we can verify the filtering logic - // by checking the field iteration manually. - let fields: &[(&str, &Option)] = &[ - ("display_name", &profile.display_name), - ("email", &profile.email), - ("username", &profile.username), - ("avatar_url", &profile.avatar_url), - ("profile_url", &profile.profile_url), - ]; - let non_empty_count = fields - .iter() - .filter(|(_, v)| v.as_deref().is_some_and(|s| !s.trim().is_empty())) - .count(); - assert_eq!( - non_empty_count, 2, - "display_name and profile_url should pass" - ); - } - - #[test] - fn parse_skill_identity_key_accepts_valid_key() { - let parsed = parse_skill_identity_key("skill:gmail:conn_1:email"); + fn parse_key_round_trip() { + let parsed = parse_skill_identity_key("skill:slack:conn_1:user_id"); assert_eq!( parsed, Some(( - "gmail".to_string(), + "slack".to_string(), "conn_1".to_string(), - "email".to_string() + "user_id".to_string() )) ); } #[test] - fn render_connected_identities_section_formats_lines() { + fn parse_key_rejects_wrong_prefix() { + assert!(parse_skill_identity_key("preference:slack:c:email").is_none()); + } + + #[test] + fn parse_key_rejects_extra_segments() { + assert!(parse_skill_identity_key("skill:slack:c:email:extra").is_none()); + } + + // ── render ──────────────────────────────────────────────────── + + #[test] + fn render_includes_handle_with_at_and_omits_user_id() { let rendered = render_connected_identities_section(&[ConnectedIdentity { - source: "gmail".into(), - identifier: "default".into(), - display_name: Some("Jane Doe".into()), - email: Some("jane@example.com".into()), - username: None, - profile_url: Some("https://mail.google.com".into()), + source: "slack".into(), + identifier: "T01ABC".into(), + display_name: Some("Cyrus Smith".into()), + email: Some("cyrus@example.com".into()), + handle: Some("cyrus".into()), + phone: None, + user_id: Some("U123ABC".into()), + avatar_url: None, + profile_url: None, }]); assert!(rendered.contains("## Connected Identities")); - assert!(rendered - .contains("- Gmail (default): Jane Doe | jane@example.com | https://mail.google.com")); + assert!(rendered.contains("- Slack (T01ABC): Cyrus Smith | cyrus@example.com | @cyrus")); + assert!( + !rendered.contains("U123ABC"), + "user_id should not appear in prompt" + ); + } + + #[test] + fn render_empty_list_returns_empty_string() { + assert_eq!(render_connected_identities_section(&[]), ""); + } + + // ── now_secs sanity ─────────────────────────────────────────── + + #[test] + fn now_secs_returns_recent_unix_seconds() { + let t = now_secs(); + assert!(t > 1_000_000_000.0); + } + + #[test] + fn persist_returns_zero_when_memory_client_not_ready() { + // Exercise the early-return branch. Global client may or may + // not be initialised in the test binary depending on ordering. + let p = fixture_profile("gmail", None, Value::Null); + let _ = persist_provider_profile(&p); } } diff --git a/src/openhuman/composio/providers/slack/provider.rs b/src/openhuman/composio/providers/slack/provider.rs index 074c5edca..83d85a0cd 100644 --- a/src/openhuman/composio/providers/slack/provider.rs +++ b/src/openhuman/composio/providers/slack/provider.rs @@ -55,6 +55,13 @@ const ACTION_LIST_CONVERSATIONS: &str = "SLACK_LIST_CONVERSATIONS"; const ACTION_FETCH_HISTORY: &str = "SLACK_FETCH_CONVERSATION_HISTORY"; /// Composio action slug for team/workspace profile fetch. const ACTION_FETCH_TEAM_INFO: &str = "SLACK_FETCH_TEAM_INFO"; +/// Composio action slug for Slack `auth.test` — returns the authed +/// user's id, handle, and team. Required for self-identity capture. +const ACTION_AUTH_TEST: &str = "SLACK_TEST_AUTH"; +/// Composio action slug for Slack `users.info` — returns the user's +/// profile (email, real_name, avatar). Optional; needs `users:read.email` +/// scope for the email field. +const ACTION_USERS_INFO: &str = "SLACK_RETRIEVE_DETAILED_USER_INFORMATION"; /// Default backfill window (days) applied when a channel has no /// cursor yet. @@ -233,59 +240,142 @@ impl ComposioProvider for SlackProvider { ) -> Result { tracing::debug!( connection_id = ?ctx.connection_id, - "[composio:slack] fetch_user_profile via {ACTION_FETCH_TEAM_INFO}" + "[composio:slack] fetch_user_profile via {ACTION_AUTH_TEST}" ); - let resp = ctx + // Step 1 — auth.test: required. Returns user_id (canonical sender + // id on Slack messages), the user's handle, and the team. + let auth_resp = ctx .client - .execute_tool(ACTION_FETCH_TEAM_INFO, Some(json!({}))) + .execute_tool(ACTION_AUTH_TEST, Some(json!({}))) .await - .map_err(|e| format!("[composio:slack] {ACTION_FETCH_TEAM_INFO} failed: {e:#}"))?; + .map_err(|e| format!("[composio:slack] {ACTION_AUTH_TEST} failed: {e:#}"))?; - if !resp.successful { - let err = resp + if !auth_resp.successful { + let err = auth_resp .error .clone() .unwrap_or_else(|| "provider reported failure".to_string()); - return Err(format!("[composio:slack] {ACTION_FETCH_TEAM_INFO}: {err}")); + return Err(format!("[composio:slack] {ACTION_AUTH_TEST}: {err}")); } - let data = &resp.data; - let display_name = pick_str(data, &["data.team.name", "data.name", "team.name", "name"]); - let profile_url = pick_str(data, &["data.team.url", "data.url", "team.url", "url"]); - let email_domain = pick_str( - data, - &[ - "data.team.email_domain", - "data.email_domain", - "team.email_domain", - "email_domain", - ], - ); - let avatar_url = pick_str( - data, - &[ - "data.team.icon.image_132", - "data.team.icon.image_68", - "team.icon.image_132", - ], - ); + // `auth_data` is the inner Composio payload — paths are relative + // to it. Slack's auth.test returns user_id/user/team/team_id at + // the top of `data`. + let auth_data = &auth_resp.data; + let user_id = pick_str(auth_data, &["user_id"]); + let handle = pick_str(auth_data, &["user"]); + let team_id = pick_str(auth_data, &["team_id"]); + let team_name = pick_str(auth_data, &["team"]); + + // Step 2 — users.info: optional. Needs `users:read.email` scope + // for `email`; falls back to `auth.test` data on missing-scope or + // any other failure so the profile still carries user_id+handle. + let mut display_name: Option = None; + let mut email: Option = None; + let mut avatar_url: Option = None; + + if let Some(uid) = user_id.as_deref() { + match ctx + .client + .execute_tool(ACTION_USERS_INFO, Some(json!({ "user": uid }))) + .await + { + Ok(info) if info.successful => { + let d = &info.data; + email = pick_str(d, &["user.profile.email", "profile.email"]); + display_name = pick_str( + d, + &[ + "user.profile.real_name", + "user.real_name", + "user.profile.display_name", + ], + ); + avatar_url = pick_str(d, &["user.profile.image_192", "user.profile.image_72"]); + } + Ok(info) => { + tracing::info!( + connection_id = ?ctx.connection_id, + error = ?info.error, + "[composio:slack] {ACTION_USERS_INFO} returned non-success — \ + falling back to auth.test data only (likely missing users:read scope)" + ); + } + Err(e) => { + tracing::info!( + connection_id = ?ctx.connection_id, + error = %e, + "[composio:slack] {ACTION_USERS_INFO} call failed — \ + falling back to auth.test data only" + ); + } + } + } + + // Step 3 — team_info: optional. Adds workspace context to `extras` + // (email_domain, icon) so the prompt section / UI can show it. + let (team_email_domain, team_icon) = match ctx + .client + .execute_tool(ACTION_FETCH_TEAM_INFO, Some(json!({}))) + .await + { + Ok(resp) if resp.successful => { + let d = &resp.data; + let domain = pick_str(d, &["team.email_domain", "email_domain"]); + let icon = pick_str(d, &["team.icon.image_132", "team.icon.image_68"]); + (domain, icon) + } + _ => (None, None), + }; + + // Display name preference: users.info real_name > auth.test handle + // > team_name (last-resort so the prompt isn't empty). + let final_display_name = display_name + .clone() + .or_else(|| handle.clone()) + .or_else(|| team_name.clone()); + + // Profile URL: users.info doesn't return one for the user + // directly; the workspace URL is acceptable as a navigational + // fallback. (Slack user profile pages are workspace-scoped and + // not stably linkable from auth.test alone.) + let profile_url = pick_str(auth_data, &["url"]); + + let avatar_url = avatar_url.or(team_icon); let profile = ProviderUserProfile { toolkit: "slack".to_string(), connection_id: ctx.connection_id.clone(), - display_name, - email: None, - username: None, + display_name: final_display_name, + email, + // username carries the platform-canonical sender id so the + // self-identity matcher can compare against Slack message + // sender_user_id directly. Handle moves into `extras` — + // `expand_identity_rows` lifts it back out as IdentityKind::Handle. + username: user_id, avatar_url, profile_url, - extras: json!({ "email_domain": email_domain, "raw": data }), + extras: json!({ + "handle": handle, + "team_id": team_id, + "team_name": team_name, + "team_email_domain": team_email_domain, + }), }; + let has_email = profile.email.is_some(); + let email_domain = profile + .email + .as_deref() + .and_then(|e| e.split('@').nth(1)) + .map(|d| d.to_string()); tracing::info!( connection_id = ?profile.connection_id, - display_name = ?profile.display_name, - "[composio:slack] fetched team info" + has_email, + email_domain = ?email_domain, + has_user_id = profile.username.is_some(), + "[composio:slack] fetched user profile" ); Ok(profile) } diff --git a/src/openhuman/composio/schemas.rs b/src/openhuman/composio/schemas.rs index a72fcb62d..03e9680a2 100644 --- a/src/openhuman/composio/schemas.rs +++ b/src/openhuman/composio/schemas.rs @@ -11,6 +11,7 @@ //! - `composio.list_github_repos` → `openhuman.composio_list_github_repos` //! - `composio.create_trigger` → `openhuman.composio_create_trigger` //! - `composio.get_user_profile` → `openhuman.composio_get_user_profile` +//! - `composio.refresh_all_identities` → `openhuman.composio_refresh_all_identities` //! - `composio.sync` → `openhuman.composio_sync` use serde::de::DeserializeOwned; @@ -67,6 +68,7 @@ pub fn all_controller_schemas() -> Vec { schemas("list_github_repos"), schemas("create_trigger"), schemas("get_user_profile"), + schemas("refresh_all_identities"), schemas("sync"), schemas("list_trigger_history"), schemas("get_user_scopes"), @@ -116,6 +118,10 @@ pub fn all_registered_controllers() -> Vec { schema: schemas("get_user_profile"), handler: handle_get_user_profile, }, + RegisteredController { + schema: schemas("refresh_all_identities"), + handler: handle_refresh_all_identities, + }, RegisteredController { schema: schemas("sync"), handler: handle_sync, @@ -335,6 +341,23 @@ pub fn schemas(function: &str) -> ControllerSchema { required: true, }], }, + "refresh_all_identities" => ControllerSchema { + namespace: "composio", + function: "refresh_all_identities", + description: + "Re-fetch user profile for every active Composio connection and persist as \ + IdentityKind-tagged rows in user_profile (#1365). Best-effort per connection \ + — failures don't abort the others.", + inputs: vec![], + outputs: vec![FieldSchema { + name: "report", + ty: TypeSchema::Json, + comment: "{ refreshed, failed, skippedNoProvider, skippedInactive, \ + rowsWritten } — aggregate counts; per-connection trail in envelope \ + messages.", + required: true, + }], + }, "sync" => ControllerSchema { namespace: "composio", function: "sync", @@ -654,6 +677,13 @@ fn handle_get_user_profile(params: Map) -> ControllerFuture { }) } +fn handle_refresh_all_identities(_params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + to_json(super::ops::composio_refresh_all_identities(&config).await?) + }) +} + fn handle_sync(params: Map) -> ControllerFuture { Box::pin(async move { let config = config_rpc::load_config_with_timeout().await?; diff --git a/src/openhuman/memory/tree/score/store.rs b/src/openhuman/memory/tree/score/store.rs index 3c8604185..4d982054a 100644 --- a/src/openhuman/memory/tree/score/store.rs +++ b/src/openhuman/memory/tree/score/store.rs @@ -12,12 +12,51 @@ use anyhow::Result; use rusqlite::{params, Connection, OptionalExtension, Transaction}; use serde::{Deserialize, Serialize}; +use crate::openhuman::composio::providers::profile::{is_self_identity_any_toolkit, IdentityKind}; use crate::openhuman::config::Config; use crate::openhuman::memory::tree::score::extract::EntityKind; use crate::openhuman::memory::tree::score::resolver::CanonicalEntity; use crate::openhuman::memory::tree::score::signals::ScoreSignals; use crate::openhuman::memory::tree::store::with_connection; +/// Map a memory-tree `EntityKind` to the Composio identity-registry +/// [`IdentityKind`] used for self-matching, or `None` for kinds that +/// don't represent identity (Url, Hashtag, Topic, Org, Loc, ...). `Person` +/// is intentionally omitted — display-name matches are weak and would +/// false-positive any contact with a similar name. +fn entity_kind_to_identity_kind(k: EntityKind) -> Option { + Some(match k { + EntityKind::Email => IdentityKind::Email, + EntityKind::Handle => IdentityKind::Handle, + _ => return None, + }) +} + +/// Resolve `is_user` for one canonical entity — true iff the surface +/// matches a self-handle of a matchable kind in the identity registry. +fn entity_is_user(entity: &CanonicalEntity) -> bool { + let Some(kind) = entity_kind_to_identity_kind(entity.kind) else { + return false; + }; + is_self_identity_any_toolkit(kind, &entity.surface) +} + +/// Same as [`entity_is_user`] but for the summary-index path where only +/// the canonical id (`":"`) is in scope. Returns `false` if +/// the id is malformed or the kind isn't matchable. +fn canonical_id_is_user(canonical_id: &str) -> bool { + let Some((kind_str, value)) = canonical_id.split_once(':') else { + return false; + }; + let Ok(kind) = EntityKind::parse(kind_str) else { + return false; + }; + let Some(idk) = entity_kind_to_identity_kind(kind) else { + return false; + }; + is_self_identity_any_toolkit(idk, value) +} + /// Serialized per-chunk score rationale. Mirrors the `mem_tree_score` row. #[derive(Clone, Debug, Serialize, Deserialize)] pub struct ScoreRow { @@ -142,12 +181,13 @@ pub fn index_entity( timestamp_ms: i64, tree_id: Option<&str>, ) -> Result<()> { + let is_user = entity_is_user(entity); with_connection(config, |conn| { conn.execute( "INSERT OR REPLACE INTO mem_tree_entity_index ( entity_id, node_id, node_kind, entity_kind, surface, - score, timestamp_ms, tree_id - ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", + score, timestamp_ms, tree_id, is_user + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)", params![ entity.canonical_id, node_id, @@ -157,6 +197,7 @@ pub fn index_entity( entity.score, timestamp_ms, tree_id, + is_user as i32, ], )?; Ok(()) @@ -181,8 +222,8 @@ pub fn index_entities( let mut stmt = tx.prepare( "INSERT OR REPLACE INTO mem_tree_entity_index ( entity_id, node_id, node_kind, entity_kind, surface, - score, timestamp_ms, tree_id - ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", + score, timestamp_ms, tree_id, is_user + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)", )?; for e in entities { stmt.execute(params![ @@ -194,6 +235,7 @@ pub fn index_entities( e.score, timestamp_ms, tree_id, + entity_is_user(e) as i32, ])?; } } @@ -250,8 +292,8 @@ pub(crate) fn index_summary_entity_ids_tx( let mut stmt = tx.prepare( "INSERT OR REPLACE INTO mem_tree_entity_index ( entity_id, node_id, node_kind, entity_kind, surface, - score, timestamp_ms, tree_id - ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", + score, timestamp_ms, tree_id, is_user + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)", )?; for canonical_id in entity_ids { // Canonical ids follow Phase 2's ":" convention. @@ -277,6 +319,7 @@ pub(crate) fn index_summary_entity_ids_tx( score, timestamp_ms, tree_id, + canonical_id_is_user(canonical_id) as i32, ])?; } Ok(entity_ids.len()) @@ -296,8 +339,8 @@ pub(crate) fn index_entities_tx( let mut stmt = tx.prepare( "INSERT OR REPLACE INTO mem_tree_entity_index ( entity_id, node_id, node_kind, entity_kind, surface, - score, timestamp_ms, tree_id - ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", + score, timestamp_ms, tree_id, is_user + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)", )?; for e in entities { stmt.execute(params![ @@ -309,6 +352,7 @@ pub(crate) fn index_entities_tx( e.score, timestamp_ms, tree_id, + entity_is_user(e) as i32, ])?; } Ok(entities.len()) @@ -325,6 +369,12 @@ pub struct EntityHit { pub score: f32, pub timestamp_ms: i64, pub tree_id: Option, + /// #1365: true when the canonical id matched the Composio identity + /// registry at index time (e.g. `email:cyrus@example.com` matches the + /// user's Gmail). Subconscious filters/weights by this flag so + /// first-person reflections only quote first-person sources. + #[serde(default)] + pub is_user: bool, } /// Find all nodes indexed against `entity_id`, newest first. @@ -339,7 +389,7 @@ pub fn lookup_entity( with_connection(config, |conn| { let mut stmt = conn.prepare( "SELECT entity_id, node_id, node_kind, entity_kind, surface, - score, timestamp_ms, tree_id + score, timestamp_ms, tree_id, is_user FROM mem_tree_entity_index WHERE entity_id = ?1 ORDER BY timestamp_ms DESC @@ -355,6 +405,7 @@ pub fn lookup_entity( e.into(), ) })?; + let is_user_int: i32 = row.get(8)?; Ok(EntityHit { entity_id: row.get(0)?, node_id: row.get(1)?, @@ -364,6 +415,7 @@ pub fn lookup_entity( score: row.get(5)?, timestamp_ms: row.get(6)?, tree_id: row.get(7)?, + is_user: is_user_int != 0, }) })? .collect::>>()?; diff --git a/src/openhuman/memory/tree/store.rs b/src/openhuman/memory/tree/store.rs index 94c9ff34d..2451facf4 100644 --- a/src/openhuman/memory/tree/store.rs +++ b/src/openhuman/memory/tree/store.rs @@ -82,6 +82,9 @@ CREATE INDEX IF NOT EXISTS idx_mem_tree_score_dropped ON mem_tree_score(dropped); -- Phase 2 (#708): inverted index entity_id -> node_id for retrieval. +-- is_user (#1365) is set at index time via the Composio identity registry +-- (is_self_identity_any_toolkit). Default 0 so legacy rows read back as +-- non-user until the backfill job re-tags them. CREATE TABLE IF NOT EXISTS mem_tree_entity_index ( entity_id TEXT NOT NULL, node_id TEXT NOT NULL, @@ -91,6 +94,7 @@ CREATE TABLE IF NOT EXISTS mem_tree_entity_index ( score REAL NOT NULL, timestamp_ms INTEGER NOT NULL, tree_id TEXT, + is_user INTEGER NOT NULL DEFAULT 0, PRIMARY KEY (entity_id, node_id) ); @@ -719,6 +723,16 @@ pub(crate) fn with_connection( // disk via `content_path` or falling back to the SQL `content` // preview. Nullable so legacy chunks keep working unchanged. add_column_if_missing(&conn, "mem_tree_chunks", "raw_refs_json", "TEXT")?; + // #1365: is_user flag on indexed entity rows. Set at write time by + // running the canonical id through the Composio identity registry + // (`is_self_identity_any_toolkit`). Default 0 so legacy rows read + // back as "not user" until the backfill job re-tags them. + add_column_if_missing( + &conn, + "mem_tree_entity_index", + "is_user", + "INTEGER NOT NULL DEFAULT 0", + )?; f(&conn) } diff --git a/src/openhuman/subconscious/prompt.rs b/src/openhuman/subconscious/prompt.rs index f51671379..a88cd05b6 100644 --- a/src/openhuman/subconscious/prompt.rs +++ b/src/openhuman/subconscious/prompt.rs @@ -54,6 +54,27 @@ daily digest, Recap window). Reflections are *not* task evaluations — they let you point out something the user should know, even if no task covers it. +**Self vs. others (#1365)**: the situation report includes a *Your +Identifiers* section listing the user's connected-account handles, +emails, and user_ids. Use it as the source of truth for who the user is. + +- *Hotness deltas* tagged `(you)` are the user's own identifiers. Frame + reflections grounded in those in second person: *"Your phoenix mentions + surged 4× this hour."* Untagged hotness items are someone or something + else — reflect on them as *about that other entity*, not as the user. +- For *Recent summaries*, *Latest global digest*, and *Recap window* + body text, the prose mentions people by name, email, or handle inline. + Match each mention against the *Your Identifiers* list: if it appears + there, that sentence is about the user; otherwise it's about a + collaborator/contact. **Never attribute another person's activity to + the user.** A digest line "Cyrus deployed phoenix" is the user's + activity only when *Cyrus* (or the matching email/handle) appears in + *Your Identifiers* — otherwise Cyrus is someone the user is reading + about, and the reflection should reflect that. +- If a reflection mixes self and other signals, separate them + explicitly: *"You spent the morning on phoenix; meanwhile, Sam pushed + a refactor."* + For each reflection: - `kind`: one of `hotness_spike` | `cross_source_pattern` | `daily_digest` | `due_item` | `risk` | `opportunity`. diff --git a/src/openhuman/subconscious/situation_report/hotness.rs b/src/openhuman/subconscious/situation_report/hotness.rs index bb1fdcf66..90046a0c7 100644 --- a/src/openhuman/subconscious/situation_report/hotness.rs +++ b/src/openhuman/subconscious/situation_report/hotness.rs @@ -22,7 +22,9 @@ const MAX_DELTAS: usize = 10; pub async fn build_section(config: &Config, workspace_dir: &Path, _last_tick_at: f64) -> String { log::debug!("[subconscious::situation_report::hotness] building section"); - // 1. Read current hotness from the memory_tree DB. + // 1. Read current hotness from the memory_tree DB. `is_user` joins + // against the entity index (#1365) so reflection generation can + // tell which movers are the user vs other people. let current = match read_current_hotness(config) { Ok(rows) => rows, Err(e) => { @@ -32,7 +34,6 @@ pub async fn build_section(config: &Config, workspace_dir: &Path, _last_tick_at: }; if current.is_empty() { - // Refresh snapshots to empty so future deltas are honest. let _ = update_snapshots(workspace_dir, &[]); return "## Hotness deltas\n\nNo entity hotness data yet.\n".to_string(); } @@ -47,26 +48,36 @@ pub async fn build_section(config: &Config, workspace_dir: &Path, _last_tick_at: }); let prev_map: std::collections::HashMap = previous.into_iter().collect(); - // 3. Compute deltas. - let mut deltas: Vec<(String, f64, f64)> = current + // 3. Compute deltas; carry is_user through. + let mut deltas: Vec = current .iter() - .map(|(eid, score)| { - let prev = prev_map.get(eid).copied().unwrap_or(0.0); - (eid.clone(), *score, score - prev) + .map(|row| { + let prev = prev_map.get(&row.entity_id).copied().unwrap_or(0.0); + HotnessDelta { + entity_id: row.entity_id.clone(), + score: row.score, + delta: row.score - prev, + is_user: row.is_user, + } }) .collect(); // Highest |delta| first; ties broken by current score. deltas.sort_by(|a, b| { - b.2.abs() - .partial_cmp(&a.2.abs()) + b.delta + .abs() + .partial_cmp(&a.delta.abs()) .unwrap_or(std::cmp::Ordering::Equal) - .then_with(|| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal)) + .then_with(|| { + b.score + .partial_cmp(&a.score) + .unwrap_or(std::cmp::Ordering::Equal) + }) }); // 4. Format top-K. - let top: Vec<&(String, f64, f64)> = deltas + let top: Vec<&HotnessDelta> = deltas .iter() - .filter(|(_, _, delta)| delta.abs() > f64::EPSILON) + .filter(|d| d.delta.abs() > f64::EPSILON) .take(MAX_DELTAS) .collect(); @@ -76,13 +87,23 @@ pub async fn build_section(config: &Config, workspace_dir: &Path, _last_tick_at: } else { let _ = writeln!( section, - "Top {} entity movers (score = post-delta, Δ = change):", + "Top {} entity movers (score = post-delta, Δ = change). \ + Items tagged `(you)` are the user's own identifiers — \ + reflect on these in second person; for everything else, \ + reflect on what *others* are doing or talking about.", top.len() ); section.push('\n'); - for (eid, score, delta) in &top { - let arrow = if *delta > 0.0 { "▲" } else { "▼" }; - let _ = writeln!(section, "- {arrow} {eid} (score={score:.2}, Δ={delta:+.2})"); + for d in &top { + let arrow = if d.delta > 0.0 { "▲" } else { "▼" }; + let self_marker = if d.is_user { " (you)" } else { "" }; + let _ = writeln!( + section, + "- {arrow} {eid}{self_marker} (score={score:.2}, Δ={delta:+.2})", + eid = d.entity_id, + score = d.score, + delta = d.delta + ); } } @@ -91,27 +112,62 @@ pub async fn build_section(config: &Config, workspace_dir: &Path, _last_tick_at: .duration_since(std::time::UNIX_EPOCH) .map(|d| d.as_secs_f64()) .unwrap_or(0.0); - if let Err(e) = update_snapshots_with_now(workspace_dir, ¤t, now) { + let snapshot_pairs: Vec<(String, f64)> = current + .iter() + .map(|r| (r.entity_id.clone(), r.score)) + .collect(); + if let Err(e) = update_snapshots_with_now(workspace_dir, &snapshot_pairs, now) { log::warn!("[subconscious::situation_report::hotness] snapshot refresh failed: {e}"); } section } -/// Read `(entity_id, last_hotness)` rows from the memory_tree DB, -/// filtering nulls. Returns rows ordered by score desc. -fn read_current_hotness(config: &Config) -> anyhow::Result> { +/// One row from `read_current_hotness`. `is_user` is OR'd across all +/// indexed nodes for the entity — true if any mention of this entity in +/// the tree resolved against the Composio identity registry. +struct CurrentHotness { + entity_id: String, + score: f64, + is_user: bool, +} + +/// Internal: a delta row with the carry-through identity flag. +struct HotnessDelta { + entity_id: String, + score: f64, + delta: f64, + is_user: bool, +} + +/// Read `(entity_id, last_hotness, is_user)` rows from the memory_tree +/// DB, filtering nulls. The `is_user` flag is computed via a correlated +/// subquery over `mem_tree_entity_index` (#1365): true iff any indexed +/// row for this entity has `is_user = 1`. +fn read_current_hotness(config: &Config) -> anyhow::Result> { crate::openhuman::memory::tree::store::with_connection(config, |conn| { let mut stmt = conn.prepare( - "SELECT entity_id, last_hotness FROM mem_tree_entity_hotness - WHERE last_hotness IS NOT NULL - ORDER BY last_hotness DESC", + "SELECT h.entity_id, + h.last_hotness, + EXISTS ( + SELECT 1 FROM mem_tree_entity_index i + WHERE i.entity_id = h.entity_id + AND i.is_user = 1 + ) AS is_user + FROM mem_tree_entity_hotness h + WHERE h.last_hotness IS NOT NULL + ORDER BY h.last_hotness DESC", )?; let rows = stmt .query_map([], |row| { let id: String = row.get(0)?; let score: f64 = row.get(1)?; - Ok((id, score)) + let is_user_int: i64 = row.get(2)?; + Ok(CurrentHotness { + entity_id: id, + score, + is_user: is_user_int != 0, + }) })? .collect::, _>>()?; Ok(rows) diff --git a/src/openhuman/subconscious/situation_report/mod.rs b/src/openhuman/subconscious/situation_report/mod.rs index 55732cf5b..a5276a82d 100644 --- a/src/openhuman/subconscious/situation_report/mod.rs +++ b/src/openhuman/subconscious/situation_report/mod.rs @@ -4,14 +4,19 @@ //! from the memory tree: //! //! 1. **Environment** (kept): host/OS/workspace/time anchor. -//! 2. **Pending Tasks** (kept): subconscious task list from SQLite. -//! 3. **Hotness deltas** (new): top movers in `mem_tree_entity_hotness` -//! since the last tick. Highest signal density. -//! 4. **Recently-sealed summaries** (new): rows from `mem_tree_summaries` +//! 2. **Your Identifiers** (#1365): the user's connected-account +//! identifiers (Slack/Gmail/Notion handles, emails, user_ids) so the +//! reflection LLM can disambiguate body-text mentions — "Cyrus said X" +//! is the user iff `Cyrus` (or the email/handle) appears in this list. +//! 3. **Pending Tasks** (kept): subconscious task list from SQLite. +//! 4. **Hotness deltas** (new): top movers in `mem_tree_entity_hotness` +//! since the last tick. Highest signal density. Items tagged `(you)` +//! are the user's own identifiers (#1365). +//! 5. **Recently-sealed summaries** (new): rows from `mem_tree_summaries` //! grouped by tree. -//! 5. **Latest global L0 digest** (new): most recent daily digest body. -//! 6. **`query_global` recap window** (new): since `last_tick_at`. -//! 7. **Recent reflections** (new): the last N reflections from the +//! 6. **Latest global L0 digest** (new): most recent daily digest body. +//! 7. **`query_global` recap window** (new): since `last_tick_at`. +//! 8. **Recent reflections** (new): the last N reflections from the //! subconscious store, used by the LLM as anti-double-emit context. //! //! Sections are appended in priority order; truncation drops the tail @@ -59,7 +64,13 @@ pub async fn build_situation_report( let env_section = build_environment_section(workspace_dir); append_section(&mut report, &mut remaining, &env_section); - // Section 2: pending subconscious tasks. + // Section 2 (#1365): the user's connected-account identifiers, so + // the reflection LLM can disambiguate "Cyrus said X" from body text + // — that's the user iff the identifier list claims it. + let identifiers_section = build_identifiers_section(); + append_section(&mut report, &mut remaining, &identifiers_section); + + // Section 3: pending subconscious tasks. let tasks_section = build_tasks_section(workspace_dir); append_section(&mut report, &mut remaining, &tasks_section); @@ -108,6 +119,37 @@ fn build_environment_section(workspace_dir: &Path) -> String { ) } +/// Render the user's connected-account identifiers (#1365) so the +/// reflection LLM can correlate body-text mentions back to the user. +/// Empty string when no providers are connected — the section just +/// disappears rather than rendering an empty header. +fn build_identifiers_section() -> String { + let identities = crate::openhuman::composio::providers::profile::load_connected_identities(); + if identities.is_empty() { + return String::new(); + } + let body = crate::openhuman::composio::providers::profile::render_connected_identities_section( + &identities, + ); + if body.trim().is_empty() { + return String::new(); + } + // The shared renderer emits "## Connected Identities". Rename the + // heading for the situation-report context so the LLM knows this is + // *the user's* identity surface, not a list of contacts. + let renamed = body.replacen("## Connected Identities", "## Your Identifiers", 1); + let mut out = renamed; + if !out.ends_with('\n') { + out.push('\n'); + } + out.push_str( + "\nWhen body text in later sections mentions any of the above (name, email, \ + handle, or user_id), treat it as the user's own activity. Anything else is \ + someone else.\n", + ); + out +} + fn build_tasks_section(workspace_dir: &Path) -> String { use std::fmt::Write; let tasks = match super::store::with_connection(workspace_dir, |conn| {