feat(memory): self-identity tagging via Composio identity registry (#1365) (#1381)

Cargo.lock

@@ -4520,7 +4520,7 @@ checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381"
 
 [[package]]
 name = "openhuman"
-version = "0.53.19"
+version = "0.53.20"
 dependencies = [
  "aes-gcm",
  "anyhow",

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

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<RpcOutcome<RefreshIdentitiesReport>> {
+    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<String> = 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"`

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 {

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

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<ProviderUserProfile, String> {
         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<String> = None;
+        let mut email: Option<String> = None;
+        let mut avatar_url: Option<String> = 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)
     }

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<ControllerSchema> {
         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<RegisteredController> {
             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<String, Value>) -> ControllerFuture {
     })
 }
 
+fn handle_refresh_all_identities(_params: Map<String, Value>) -> 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<String, Value>) -> ControllerFuture {
     Box::pin(async move {
         let config = config_rpc::load_config_with_timeout().await?;

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<IdentityKind> {
+    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 (`"<kind>:<value>"`) 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 "<kind>:<value>" 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<String>,
+    /// #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::<rusqlite::Result<Vec<_>>>()?;

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<T>(
     // 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)
 }
 

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`.

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<String, f64> = 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<HotnessDelta> = 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, &current, 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<Vec<(String, f64)>> {
+/// 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<Vec<CurrentHotness>> {
     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::<Result<Vec<_>, _>>()?;
         Ok(rows)

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| {