fix(relay): wake mentioned agents from workflow send_message (e3661764)

The workflow send_message action emitted a kind:9 event with a fixed tag
set [p=author, h, buzz:workflow] and never parsed @Name mentions from the
message text. ACP agent-wake (event_mentions_agent) is strictly p-tag
gated, so an agent named in a workflow message was never woken.

Resolve @Name mentions against the destination channel's members and
append one p tag per match. Matching is deliberately conservative to
avoid waking the wrong agent:

- members-only candidates (no global users)
- exact display name, case-insensitive, greedy-longest non-overlapping
- next char must not extend a member name (unicode-safe boundary, no
  punctuation allowlist); matched over Vec<char>, never byte offsets
- ambiguous name (2+ members, distinct pubkeys) wakes no one
- author is not re-tagged (already attributed)

11 unit tests cover the boundary/greedy/ambiguity/dedupe cases plus a
non-ASCII member name; one Postgres-gated regression proves the emitted
event carries the mentioned member's p tag end to end.

Co-authored-by: Dawn (sprout agent) <c6237ef84fa537c78dcee78efd2d4e59f728859c7f194da42ac51ededfa0be05@sprout-oss.stage.blox.sqprod.co>
Signed-off-by: tlongwell-block <109685178+tlongwell-block@users.noreply.github.com>

crates/buzz-relay/src/workflow_sink.rs

@@ -19,6 +19,106 @@ use uuid::Uuid;
 use crate::handlers::event::dispatch_persistent_event;
 use crate::state::AppState;
 
+/// Resolves `@Name` mentions in workflow message text to the pubkeys of the
+/// channel members they name, so the emitted kind:9 carries the `p` tags that
+/// ACP agent-wake (`event_mentions_agent`) is gated on.
+///
+/// The client resolves mentions to `p` tags at compose time from an interactive
+/// autocomplete pick; the workflow path has only free text, so this reverse-parse
+/// *defines* the matching contract. It is deliberately conservative to avoid
+/// waking the wrong agent:
+///
+/// - **Members only.** Candidates are the destination channel's members; global
+///   users are never matched.
+/// - **Exact display name.** No substring, prefix, or fuzzy matching. Names may
+///   contain spaces/punctuation (`"Will Pfleger"`, `"Lep (Subagent)"`), so the
+///   match is anchored on `@` and terminated by a non-name boundary rather than
+///   whitespace.
+/// - **Greedy-longest, non-overlapping.** Longer names are matched first and
+///   consume their span, so `@Will Pfleger` binds *Pfleger* and a bare `@Will`
+///   does not match the member `"Will Pfleger"`.
+/// - **Ambiguous names wake no one.** If two or more members share the matched
+///   display name, no `p` tag is emitted for it — arbitrary selection would
+///   silently misroute and tagging all of them is a false-wake firehose.
+///
+/// Returns deduplicated pubkey hexes, in first-appearance order in `text`.
+fn resolve_mention_pubkeys(text: &str, members: &[(String, String)]) -> Vec<String> {
+    // Name → pubkey, folding case (client matches case-insensitively). A name
+    // that maps to more than one distinct pubkey is ambiguous → wake no one.
+    let mut by_name: std::collections::HashMap<String, Option<String>> =
+        std::collections::HashMap::new();
+    for (name, pubkey) in members {
+        if name.trim().is_empty() {
+            continue;
+        }
+        by_name
+            .entry(name.to_lowercase())
+            .and_modify(|slot| {
+                if slot.as_deref() != Some(pubkey.as_str()) {
+                    *slot = None; // ambiguous
+                }
+            })
+            .or_insert_with(|| Some(pubkey.clone()));
+    }
+
+    // Match longest names first so a longer name consumes its span before a
+    // shorter substring name can claim part of it.
+    let mut names: Vec<&(String, String)> = members.iter().collect();
+    names.sort_by_key(|(name, _)| std::cmp::Reverse(name.chars().count()));
+
+    let chars: Vec<char> = text.chars().collect();
+    let mut consumed = vec![false; chars.len()];
+    let lower: Vec<char> = text.to_lowercase().chars().collect();
+
+    // A mention is anchored on `@` at a left boundary (start / whitespace / `(`)
+    // and the matched name must not be followed by a name-continuation char —
+    // otherwise `@Will` would match inside `@Willow`. Combined with matching the
+    // longest member name first, this is the whole rule: no punctuation allowlist
+    // to get wrong, and it is unicode-safe (em-dash, emoji all terminate a name).
+    let is_left_boundary = |i: usize| i == 0 || chars[i - 1].is_whitespace() || chars[i - 1] == '(';
+    let extends_name = |c: char| c.is_alphanumeric() || c == '_';
+
+    let mut out: Vec<String> = Vec::new();
+    let mut seen = std::collections::HashSet::new();
+    let mut hits: Vec<(usize, String)> = Vec::new();
+
+    for (name, _) in &names {
+        let name_lower: Vec<char> = name.to_lowercase().chars().collect();
+        if name_lower.is_empty() {
+            continue;
+        }
+        // `@` + name length.
+        let span = 1 + name_lower.len();
+        let mut at = 0;
+        while at + span <= chars.len() {
+            if chars[at] == '@'
+                && is_left_boundary(at)
+                && !consumed[at]
+                && lower[at + 1..at + span] == name_lower[..]
+                && chars[at + span..].first().is_none_or(|&c| !extends_name(c))
+            {
+                if let Some(Some(pubkey)) = by_name.get(&name.to_lowercase()) {
+                    hits.push((at, pubkey.clone()));
+                }
+                for slot in consumed.iter_mut().skip(at).take(span) {
+                    *slot = true;
+                }
+                at += span;
+            } else {
+                at += 1;
+            }
+        }
+    }
+
+    hits.sort_by_key(|(at, _)| *at);
+    for (_, pubkey) in hits {
+        if seen.insert(pubkey.clone()) {
+            out.push(pubkey);
+        }
+    }
+    out
+}
+
 /// Relay-side action sink — executes workflow side-effects directly.
 ///
 /// Holds a **weak** reference to `AppState` to avoid an `Arc` reference cycle:
@@ -126,7 +226,9 @@ impl ActionSink for RelayActionSink {
             //    - `p` tag attributes the message to the workflow owner
             //    - `h` tag scopes to the channel (NIP-29, canonical UUID)
             //    - `buzz:workflow` tag prevents recursive workflow triggering
-            let tags = vec![
+            //    - one `p` tag per `@Name` that resolves to a channel member,
+            //      so mentioned agents are woken (wake is `p`-tag gated)
+            let mut tags = vec![
                 Tag::parse(["p", &author_pubkey_hex])
                     .map_err(|e| ActionSinkError::EventBuild(format!("p tag: {e}")))?,
                 Tag::parse(["h", &channel_id_canonical])
@@ -135,6 +237,38 @@ impl ActionSink for RelayActionSink {
                     .map_err(|e| ActionSinkError::EventBuild(format!("workflow tag: {e}")))?,
             ];
 
+            // Resolve `@Name` mentions to channel-member pubkeys and append a
+            // `p` tag for each (skipping the author, already tagged above). A
+            // resolution failure must not drop the message, so log and proceed
+            // with the base tags.
+            let members = state
+                .db
+                .get_members(tenant.community(), channel_uuid)
+                .await
+                .map_err(|e| ActionSinkError::Database(e.to_string()))?;
+            let member_pubkeys: Vec<Vec<u8>> = members.iter().map(|m| m.pubkey.clone()).collect();
+            let users = state
+                .db
+                .get_users_bulk(tenant.community(), &member_pubkeys)
+                .await
+                .map_err(|e| ActionSinkError::Database(e.to_string()))?;
+            let named_members: Vec<(String, String)> = users
+                .into_iter()
+                .filter_map(|u| {
+                    let name = u.display_name?;
+                    Some((name, nostr::PublicKey::from_slice(&u.pubkey).ok()?.to_hex()))
+                })
+                .collect();
+            for mentioned in resolve_mention_pubkeys(&text, &named_members) {
+                if mentioned == author_pubkey_hex {
+                    continue;
+                }
+                tags.push(
+                    Tag::parse(["p", &mentioned])
+                        .map_err(|e| ActionSinkError::EventBuild(format!("mention p tag: {e}")))?,
+                );
+            }
+
             let kind = Kind::from(KIND_STREAM_MESSAGE as u16);
             let event = EventBuilder::new(kind, &text)
                 .tags(tags)
@@ -200,3 +334,278 @@ impl ActionSink for RelayActionSink {
         })
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn m(name: &str, pubkey: &str) -> (String, String) {
+        (name.to_string(), pubkey.to_string())
+    }
+
+    // A 64-char hex pubkey built from a single repeated nibble, for readable tests.
+    fn pk(nibble: char) -> String {
+        std::iter::repeat_n(nibble, 64).collect()
+    }
+
+    #[test]
+    fn resolves_exact_member_name() {
+        let members = vec![m("Robby", &pk('a'))];
+        assert_eq!(
+            resolve_mention_pubkeys("heads up @Robby — please take a look", &members),
+            vec![pk('a')]
+        );
+    }
+
+    #[test]
+    fn matches_case_insensitively() {
+        let members = vec![m("Robby", &pk('a'))];
+        assert_eq!(
+            resolve_mention_pubkeys("ping @robby", &members),
+            vec![pk('a')]
+        );
+    }
+
+    #[test]
+    fn ignores_non_member_and_bare_at() {
+        let members = vec![m("Robby", &pk('a'))];
+        assert!(resolve_mention_pubkeys("hey @Stranger and @", &members).is_empty());
+    }
+
+    #[test]
+    fn greedy_longest_binds_full_name_not_prefix() {
+        // Both "Will" and "Will Pfleger" are members. `@Will Pfleger` must bind
+        // Pfleger's key only; a bare `@Will` binds Will.
+        let members = vec![m("Will", &pk('1')), m("Will Pfleger", &pk('2'))];
+        assert_eq!(
+            resolve_mention_pubkeys("cc @Will Pfleger on this", &members),
+            vec![pk('2')]
+        );
+        assert_eq!(
+            resolve_mention_pubkeys("cc @Will on this", &members),
+            vec![pk('1')]
+        );
+    }
+
+    #[test]
+    fn at_mid_token_does_not_match() {
+        // `@` must sit at a left boundary (start / whitespace / `(`). An email-ish
+        // or mid-token `@` (`alice@Robby`) must not wake Robby.
+        let members = vec![m("Robby", &pk('a'))];
+        assert!(resolve_mention_pubkeys("alice@Robby", &members).is_empty());
+    }
+
+    #[test]
+    fn prefix_member_does_not_match_inside_longer_word() {
+        // "Sam" is a member; `@Sami` (no "Sami" member) must not wake Sam.
+        let members = vec![m("Sam", &pk('3'))];
+        assert!(resolve_mention_pubkeys("hi @Sami", &members).is_empty());
+    }
+
+    #[test]
+    fn name_with_spaces_and_punctuation() {
+        let members = vec![m("Lep (Subagent)", &pk('4'))];
+        assert_eq!(
+            resolve_mention_pubkeys("@Lep (Subagent) take it", &members),
+            vec![pk('4')]
+        );
+    }
+
+    #[test]
+    fn em_dash_terminates_name() {
+        // Generated prose often writes `@Name—text` with no space.
+        let members = vec![m("Robby", &pk('a'))];
+        assert_eq!(
+            resolve_mention_pubkeys("@Robby—please look", &members),
+            vec![pk('a')]
+        );
+    }
+
+    #[test]
+    fn non_ascii_member_name() {
+        let members = vec![m("Zoë", &pk('5'))];
+        assert_eq!(
+            resolve_mention_pubkeys("welcome @Zoë!", &members),
+            vec![pk('5')]
+        );
+    }
+
+    #[test]
+    fn ambiguous_name_wakes_no_one() {
+        // Six "Fizz" agents (real team case) with distinct pubkeys → tag none.
+        let members = vec![
+            m("Fizz", &pk('6')),
+            m("Fizz", &pk('7')),
+            m("Fizz", &pk('8')),
+        ];
+        assert!(resolve_mention_pubkeys("@Fizz status?", &members).is_empty());
+    }
+
+    #[test]
+    fn duplicate_name_same_pubkey_is_not_ambiguous() {
+        // Same identity listed twice (e.g. two channels) is not a conflict.
+        let members = vec![m("Fizz", &pk('6')), m("Fizz", &pk('6'))];
+        assert_eq!(resolve_mention_pubkeys("@Fizz go", &members), vec![pk('6')]);
+    }
+
+    #[test]
+    fn dedupes_repeated_mentions_in_first_appearance_order() {
+        let members = vec![m("Robby", &pk('a')), m("Max", &pk('b'))];
+        assert_eq!(
+            resolve_mention_pubkeys("@Max then @Robby then @Max again", &members),
+            vec![pk('b'), pk('a')]
+        );
+    }
+}
+
+#[cfg(test)]
+mod integration_tests {
+    //! Regression test for `e3661764` / `7899c1a8`: a workflow `send_message`
+    //! that mentions a channel member by name (`@Name`) must emit a `p` tag for
+    //! that member so ACP agent wake (`event_mentions_agent`, p-tag gated) fires.
+    //!
+    //! Postgres-gated like the other DB-backed relay tests. Run with:
+    //!   `cargo test -p buzz-relay --lib workflow_sink -- --ignored`
+    use super::*;
+    use buzz_core::channel::{ChannelType, ChannelVisibility, MemberRole};
+    use buzz_db::CreateCommunityWithOwnerResult;
+    use std::sync::Arc;
+
+    /// Real-PG state mirroring `handlers::event::tests::test_state_with_redis_url`.
+    async fn test_state() -> Arc<AppState> {
+        let mut config = crate::config::Config::from_env().expect("default config loads");
+        config.require_relay_membership = false;
+        config.redis_url = "redis://127.0.0.1:1".to_string();
+        let pool = sqlx::PgPool::connect_lazy(&config.database_url).expect("lazy pg pool");
+        let db = buzz_db::Db::from_pool(pool.clone());
+        let redis_pool = deadpool_redis::Config::from_url(&config.redis_url)
+            .create_pool(Some(deadpool_redis::Runtime::Tokio1))
+            .expect("redis pool");
+        let pubsub = Arc::new(
+            buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone())
+                .await
+                .expect("pubsub manager"),
+        );
+        let audit = buzz_audit::AuditService::new(pool.clone());
+        let auth = buzz_auth::AuthService::new(config.auth.clone());
+        let search = buzz_search::SearchService::new(pool.clone());
+        let workflow_engine = Arc::new(buzz_workflow::WorkflowEngine::new(
+            db.clone(),
+            buzz_workflow::WorkflowConfig::default(),
+        ));
+        let media_storage = buzz_media::MediaStorage::new(&config.media).expect("media storage");
+        let (state, _audit_shutdown) = AppState::new(
+            config,
+            db,
+            redis_pool,
+            audit,
+            pubsub,
+            auth,
+            search,
+            workflow_engine,
+            nostr::Keys::generate(),
+            media_storage,
+        );
+        Arc::new(state)
+    }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn workflow_send_message_p_tags_mentioned_member() {
+        let state = test_state().await;
+
+        let author = nostr::Keys::generate();
+        let author_hex = author.public_key().to_hex();
+        let agent = nostr::Keys::generate();
+        let agent_hex = agent.public_key().to_hex();
+        let agent_bytes = agent.public_key().to_bytes().to_vec();
+
+        let host = format!("wf-ptag-{}.example", uuid::Uuid::new_v4().simple());
+        let community = match state
+            .db
+            .create_community_with_owner(&host, &author_hex)
+            .await
+            .expect("create community")
+        {
+            CreateCommunityWithOwnerResult::Created(rec) => rec.id,
+            other => panic!("expected fresh community, got {other:?}"),
+        };
+
+        // Open channel; the creator (author) is bootstrapped as an owner-member.
+        let channel = state
+            .db
+            .create_channel(
+                community,
+                "wf-ptag",
+                ChannelType::Stream,
+                ChannelVisibility::Open,
+                None,
+                &author.public_key().to_bytes(),
+                None,
+            )
+            .await
+            .expect("create channel");
+
+        // The mentioned agent is a real member with a resolvable display name.
+        state
+            .db
+            .ensure_user(community, &agent_bytes)
+            .await
+            .expect("ensure agent user row");
+        state
+            .db
+            .update_user_profile(community, &agent_bytes, Some("Robby"), None, None, None)
+            .await
+            .expect("set agent display name");
+        state
+            .db
+            .add_member(
+                community,
+                channel.id,
+                &agent_bytes,
+                MemberRole::Bot,
+                Some(&author.public_key().to_bytes()),
+            )
+            .await
+            .expect("add agent member");
+
+        let sink = RelayActionSink::new(&state);
+        let event_id_hex = sink
+            .send_message(
+                community,
+                &channel.id.to_string(),
+                "heads up @Robby — please take a look",
+                &author_hex,
+            )
+            .await
+            .expect("send_message");
+
+        let id_bytes = nostr::EventId::from_hex(&event_id_hex)
+            .expect("event id")
+            .as_bytes()
+            .to_vec();
+        let stored = state
+            .db
+            .get_event_by_id(community, &id_bytes)
+            .await
+            .expect("query event")
+            .expect("event persisted");
+
+        let p_tag_targets: Vec<&str> = stored
+            .event
+            .tags
+            .iter()
+            .filter(|t| t.as_slice().first().map(|s| s.as_str()) == Some("p"))
+            .filter_map(|t| t.as_slice().get(1).map(|s| s.as_str()))
+            .collect();
+
+        assert!(
+            p_tag_targets.contains(&author_hex.as_str()),
+            "author should still be attributed via p tag; got {p_tag_targets:?}"
+        );
+        assert!(
+            p_tag_targets.contains(&agent_hex.as_str()),
+            "mentioned member {agent_hex} must be p-tagged so it wakes; got {p_tag_targets:?}"
+        );
+    }
+}