From 2be524e9a3c418eb666c85fd0442deb7649b43d9 Mon Sep 17 00:00:00 2001 From: tlongwell-block <109685178+tlongwell-block@users.noreply.github.com> Date: Fri, 17 Jul 2026 12:20:02 -0400 Subject: [PATCH] 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, 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) Signed-off-by: tlongwell-block <109685178+tlongwell-block@users.noreply.github.com> --- crates/buzz-relay/src/workflow_sink.rs | 411 ++++++++++++++++++++++++- 1 file changed, 410 insertions(+), 1 deletion(-) diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 857c14626..7d872f0a0 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/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 { + // 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> = + 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 = text.chars().collect(); + let mut consumed = vec![false; chars.len()]; + let lower: Vec = 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 = 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 = 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> = 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 { + 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:?}" + ); + } +}