Improve mentions for agents + people (#942)

Author: klopez4212

crates/sprout-acp/src/lib.rs

@@ -1415,7 +1415,7 @@ async fn tokio_main() -> Result<()> {
 
                                     if let Some(filter) = config::resolve_dynamic_channel_filter(&config, ch, &rules) {
                                         tracing::info!(channel_id = %ch, "membership notification: subscribing to new channel");
-                                        if let Err(e) = relay.subscribe_channel(ch, filter).await {
+                                        if let Err(e) = relay.subscribe_channel_from(ch, filter, Some(ts)).await {
                                             tracing::warn!("failed to subscribe to new channel {ch}: {e}");
                                         }
                                     } else {

crates/sprout-acp/src/relay.rs

@@ -114,6 +114,13 @@ const REST_RETRY_BASE_DELAYS: [Duration; 3] = [
     Duration::from_millis(2000),
 ];
 
+fn unix_now_secs() -> u64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_secs()
+}
+
 impl RestClient {
     // ── NIP-98 signing ────────────────────────────────────────────────────
 
@@ -370,6 +377,7 @@ enum RelayCommand {
     Subscribe {
         channel_id: Uuid,
         filter: ChannelFilter,
+        replay_since: Option<u64>,
     },
     /// Unsubscribe from a channel (sends a NIP-01 CLOSE).
     Unsubscribe { channel_id: Uuid },
@@ -634,9 +642,28 @@ impl HarnessRelay {
         &mut self,
         channel_id: Uuid,
         filter: ChannelFilter,
+    ) -> Result<(), RelayError> {
+        self.subscribe_channel_from(channel_id, filter, None).await
+    }
+
+    /// Subscribe to events in a channel, replaying from a known timestamp.
+    ///
+    /// Used for channels discovered from membership notifications: the mention
+    /// that invited an agent can be published immediately after the membership
+    /// event, before this subscription is active. Replaying from the membership
+    /// event timestamp closes that race.
+    pub async fn subscribe_channel_from(
+        &mut self,
+        channel_id: Uuid,
+        filter: ChannelFilter,
+        replay_since: Option<u64>,
     ) -> Result<(), RelayError> {
         self.cmd_tx
-            .send(RelayCommand::Subscribe { channel_id, filter })
+            .send(RelayCommand::Subscribe {
+                channel_id,
+                filter,
+                replay_since,
+            })
             .await
             .map_err(|_| RelayError::ConnectionClosed)?;
         debug!("queued subscribe for channel {channel_id}");
@@ -888,12 +915,12 @@ struct BgState {
     /// (Finding #22). Used as the floor `since` for membership notification
     /// replay so events predating this session are never re-delivered.
     startup_watermark: Option<u64>,
-    /// Wall-clock timestamp when each channel was first subscribed.
+    /// Replay floor captured when each channel was first subscribed.
     /// Used as the `since` fallback on reconnect for channels that have no
     /// `last_seen` or `channel_dropped_since`. This prevents channels joined
-    /// after startup from replaying from `startup_watermark` (which could be
-    /// hours old), while still allowing startup-era channels to use the
-    /// startup watermark via their `subscribe_since ≈ startup_watermark`.
+    /// after startup from replaying from an hours-old `startup_watermark`.
+    /// Startup-era channels use the startup watermark; dynamic channels use
+    /// the membership notification timestamp that caused the subscription.
     subscribe_since: HashMap<Uuid, u64>,
 }
 
@@ -977,20 +1004,22 @@ impl BgState {
 /// arm here is a logic error.
 fn apply_command_to_state(state: &mut BgState, cmd: RelayCommand) {
     match cmd {
-        RelayCommand::Subscribe { channel_id, filter } => {
+        RelayCommand::Subscribe {
+            channel_id,
+            filter,
+            replay_since,
+        } => {
             state
                 .active_subscriptions
                 .insert(channel_id, channel_sub_id(channel_id));
             state.active_filters.insert(channel_id, filter);
             state.subscribe_since.entry(channel_id).or_insert_with(|| {
-                // Use startup_watermark as floor when available — closes the
+                // Use an explicit replay floor when available (dynamic
+                // membership), otherwise startup_watermark closes the startup
                 // blind spot between watermark capture and first REQ.
-                state.startup_watermark.unwrap_or_else(|| {
-                    std::time::SystemTime::now()
-                        .duration_since(std::time::UNIX_EPOCH)
-                        .unwrap_or_default()
-                        .as_secs()
-                })
+                replay_since
+                    .or(state.startup_watermark)
+                    .unwrap_or_else(unix_now_secs)
             });
         }
         RelayCommand::Unsubscribe { channel_id } => {
@@ -1040,19 +1069,18 @@ async fn execute_connected_command(
     cmd: RelayCommand,
 ) -> bool {
     match cmd {
-        RelayCommand::Subscribe { channel_id, filter } => {
+        RelayCommand::Subscribe {
+            channel_id,
+            filter,
+            replay_since,
+        } => {
             // Seed subscribe_since BEFORE computing since — on first
             // subscribe, this provides the fallback timestamp that
-            // closes the startup blind spot. Use startup_watermark as
-            // floor when available so events between watermark capture
-            // and this REQ are not missed.
+            // closes the startup/dynamic-membership blind spot.
             state.subscribe_since.entry(channel_id).or_insert_with(|| {
-                state.startup_watermark.unwrap_or_else(|| {
-                    std::time::SystemTime::now()
-                        .duration_since(std::time::UNIX_EPOCH)
-                        .unwrap_or_default()
-                        .as_secs()
-                })
+                replay_since
+                    .or(state.startup_watermark)
+                    .unwrap_or_else(unix_now_secs)
             });
             let since = state
                 .last_seen
@@ -1070,7 +1098,14 @@ async fn execute_connected_command(
             } else {
                 // Send failed — record intent so reconnect restores it.
                 warn!("subscribe REQ failed for channel {channel_id} — recording intent for reconnect");
-                apply_command_to_state(state, RelayCommand::Subscribe { channel_id, filter });
+                apply_command_to_state(
+                    state,
+                    RelayCommand::Subscribe {
+                        channel_id,
+                        filter,
+                        replay_since,
+                    },
+                );
                 false
             }
         }
@@ -3352,6 +3387,38 @@ mod tests {
         );
     }
 
+    #[test]
+    fn dynamic_subscribe_records_membership_replay_floor() {
+        let mut state = BgState::new();
+        state.startup_watermark = Some(2_000);
+        let channel_id = Uuid::new_v4();
+        let membership_ts = 10_000;
+        let filter = ChannelFilter {
+            kinds: Some(vec![9]),
+            require_mention: true,
+        };
+
+        apply_command_to_state(
+            &mut state,
+            RelayCommand::Subscribe {
+                channel_id,
+                filter,
+                replay_since: Some(membership_ts),
+            },
+        );
+
+        assert_eq!(
+            state.subscribe_since.get(&channel_id).copied(),
+            Some(membership_ts),
+            "dynamic channel subscriptions should replay from the membership notification, not startup"
+        );
+        assert_eq!(
+            state.channel_since(&channel_id),
+            Some(membership_ts),
+            "channel_since should use the dynamic replay floor until an event is seen"
+        );
+    }
+
     // ── Membership dedup regression tests (M4) ───────────────────────────
 
     /// Membership dedup must NOT contaminate per-channel `last_seen`.

desktop/src-tauri/src/commands/channels.rs

@@ -287,21 +287,31 @@ pub async fn get_channel_members(
         .await
         .unwrap_or_default();
 
-        // Build pubkey → display_name map from kind:0 events
-        let mut name_map = std::collections::HashMap::new();
+        // Build pubkey → profile display metadata from kind:0 events.
+        let mut profile_map = std::collections::HashMap::new();
         for ev in &profile_events {
             let pk = ev.pubkey.to_hex();
             if let Ok(profile) = nostr_convert::profile_info_from_event(ev) {
-                if let Some(name) = profile.display_name {
-                    name_map.insert(pk, name);
-                }
+                profile_map.insert(
+                    pk,
+                    (
+                        profile.display_name,
+                        nostr_convert::profile_has_valid_oa_owner(ev),
+                    ),
+                );
             }
         }
 
-        // Populate display_name on each member
+        // Populate profile-derived fields on each member.
         for member in &mut response.members {
-            if member.display_name.is_none() {
-                member.display_name = name_map.get(&member.pubkey).cloned();
+            if member.role == "bot" {
+                member.is_agent = true;
+            }
+            if let Some((display_name, is_agent)) = profile_map.get(&member.pubkey) {
+                if member.display_name.is_none() {
+                    member.display_name = display_name.clone();
+                }
+                member.is_agent = member.is_agent || *is_agent;
             }
         }
     }

desktop/src-tauri/src/commands/messages.rs

@@ -272,6 +272,7 @@ pub async fn send_channel_message(
     parent_event_id: Option<String>,
     media_tags: Option<Vec<Vec<String>>>,
     emoji_tags: Option<Vec<Vec<String>>>,
+    mention_tags: Option<Vec<Vec<String>>>,
     mention_pubkeys: Option<Vec<String>>,
     kind: Option<u32>,
     state: State<'_, AppState>,
@@ -282,14 +283,19 @@ pub async fn send_channel_message(
     let mention_refs: Vec<&str> = mentions.iter().map(|s| s.as_str()).collect();
     let media = media_tags.unwrap_or_default();
     let emoji = emoji_tags.unwrap_or_default();
+    let mention_refs_only = mention_tags.unwrap_or_default();
     let kind_num = kind.unwrap_or(sprout_core::kind::KIND_STREAM_MESSAGE);
 
     let mut resolved_root: Option<String> = None;
 
     let builder = match kind_num {
-        sprout_core::kind::KIND_FORUM_POST => {
-            events::build_forum_post(channel_uuid, content.trim(), &mention_refs, &media)?
-        }
+        sprout_core::kind::KIND_FORUM_POST => events::build_forum_post(
+            channel_uuid,
+            content.trim(),
+            &mention_refs,
+            &media,
+            &mention_refs_only,
+        )?,
         sprout_core::kind::KIND_FORUM_COMMENT => {
             let parent_id = parent_event_id
                 .as_deref()
@@ -302,6 +308,7 @@ pub async fn send_channel_message(
                 &thread_ref,
                 &mention_refs,
                 &media,
+                &mention_refs_only,
             )?
         }
         _ => {
@@ -320,6 +327,7 @@ pub async fn send_channel_message(
                 &mention_refs,
                 &media,
                 &emoji,
+                &mention_refs_only,
             )?
         }
     };

desktop/src-tauri/src/commands/profile.rs

@@ -177,10 +177,23 @@ pub async fn search_users(
     let trimmed = query.trim();
     let max = limit.unwrap_or(8).min(50) as usize;
 
-    if trimmed.is_empty() || max == 0 {
+    if max == 0 {
         return Ok(SearchUsersResponse { users: Vec::new() });
     }
 
+    if trimmed.is_empty() {
+        let events = query_relay(
+            &state,
+            &[serde_json::json!({
+                "kinds": [0],
+                "limit": max,
+            })],
+        )
+        .await?;
+
+        return Ok(nostr_convert::list_user_search_results(&events, max));
+    }
+
     // NIP-50 full-text search on kind:0 profiles. The relay's HTTP bridge
     // intercepts the `search` field on POST /query and routes to Typesense
     // (see `crates/sprout-relay/src/api/bridge.rs::handle_bridge_search`),

desktop/src-tauri/src/events.rs

@@ -76,6 +76,23 @@ fn mention_tags(mentions: &[&str]) -> Result<Vec<Tag>, String> {
     Ok(tags)
 }
 
+fn mention_reference_tags(mentions: &[Vec<String>], tags: &mut Vec<Tag>) -> Result<(), String> {
+    for mention in mentions {
+        if mention.first().map(String::as_str) != Some("mention") {
+            return Err(format!(
+                "mention reference tags must use 'mention' prefix (got {:?})",
+                mention.first()
+            ));
+        }
+        let Some(pubkey) = mention.get(1) else {
+            return Err("mention reference tag missing pubkey".into());
+        };
+        check_pubkey(pubkey)?;
+        tags.push(tag(vec!["mention", &pubkey.to_ascii_lowercase()])?);
+    }
+    Ok(())
+}
+
 /// Validate and append imeta tags. Rejects any tag whose first element is not "imeta"
 /// to prevent injection of arbitrary tags (e.g., forged "h", "e", or "p" tags).
 fn imeta_tags(media_tags: &[Vec<String>], tags: &mut Vec<Tag>) -> Result<(), String> {
@@ -277,6 +294,7 @@ pub fn build_message(
     mentions: &[&str],
     media_tags: &[Vec<String>],
     custom_emoji_tags: &[Vec<String>],
+    mention_ref_tags: &[Vec<String>],
 ) -> Result<EventBuilder, String> {
     check_content(content)?;
     let mut tags = vec![tag(vec!["h", &channel_id.to_string()])?];
@@ -286,6 +304,7 @@ pub fn build_message(
     tags.extend(mention_tags(mentions)?);
     imeta_tags(media_tags, &mut tags)?;
     emoji_tags(custom_emoji_tags, &mut tags)?;
+    mention_reference_tags(mention_ref_tags, &mut tags)?;
     Ok(EventBuilder::new(Kind::Custom(9), content).tags(tags))
 }
 
@@ -295,11 +314,13 @@ pub fn build_forum_post(
     content: &str,
     mentions: &[&str],
     media_tags: &[Vec<String>],
+    mention_ref_tags: &[Vec<String>],
 ) -> Result<EventBuilder, String> {
     check_content(content)?;
     let mut tags = vec![tag(vec!["h", &channel_id.to_string()])?];
     tags.extend(mention_tags(mentions)?);
     imeta_tags(media_tags, &mut tags)?;
+    mention_reference_tags(mention_ref_tags, &mut tags)?;
     Ok(EventBuilder::new(Kind::Custom(45001), content).tags(tags))
 }
 
@@ -310,12 +331,14 @@ pub fn build_forum_comment(
     thread_ref: &ThreadRef,
     mentions: &[&str],
     media_tags: &[Vec<String>],
+    mention_ref_tags: &[Vec<String>],
 ) -> Result<EventBuilder, String> {
     check_content(content)?;
     let mut tags = vec![tag(vec!["h", &channel_id.to_string()])?];
     tags.extend(thread_tags(thread_ref)?);
     tags.extend(mention_tags(mentions)?);
     imeta_tags(media_tags, &mut tags)?;
+    mention_reference_tags(mention_ref_tags, &mut tags)?;
     Ok(EventBuilder::new(Kind::Custom(45003), content).tags(tags))
 }
 

desktop/src-tauri/src/huddle/pipeline.rs

@@ -269,13 +269,14 @@ pub(crate) fn spawn_transcription_task(
                 .clone();
 
             let p_tags: Vec<&str> = agent_pubkeys.iter().map(|s| s.as_str()).collect();
-            let builder = match events::build_message(channel_uuid, &t, None, &p_tags, &[], &[]) {
-                Ok(b) => b,
-                Err(e) => {
-                    eprintln!("sprout-desktop: STT build_message: {e}");
-                    continue;
-                }
-            };
+            let builder =
+                match events::build_message(channel_uuid, &t, None, &p_tags, &[], &[], &[]) {
+                    Ok(b) => b,
+                    Err(e) => {
+                        eprintln!("sprout-desktop: STT build_message: {e}");
+                        continue;
+                    }
+                };
             let event = match builder.sign_with_keys(&keys) {
                 Ok(e) => e,
                 Err(e) => {