From 8da159f9d5af8c11c98b14090914c9758027405a Mon Sep 17 00:00:00 2001 From: Niko Moritz <126672558+nikomoritz@users.noreply.github.com> Date: Thu, 6 Aug 2026 13:54:14 +0200 Subject: [PATCH] feat(buzz-acp): follow threads the agent participates in MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds an opt-in --thread-follow / BUZZ_ACP_THREAD_FOLLOW flag (mentions mode only). When enabled, the harness tracks a bounded set of thread roots where the agent was @mentioned or has itself replied. Replies in a followed thread then trigger turns without a fresh @mention, so a human can tag the agent once at the top of a thread and keep talking. - mentions-mode subscriptions drop the server-side #p filter when the flag is set; the mention gate is re-applied client-side with the follow-set exemption, so behaviour is unchanged when the flag is off - new thread_follow module: bounded FIFO-evicting root set (cap 1024), NIP-10 root extraction (root > reply > bare e tag), p-tag check - follow roots are learned from agent mentions (after the author gate) and from self-authored replies (before the ignore-self drop) - exempted events get prompt tag "thread-follow" so prompts show why the turn triggered Known limitations of this first cut: the follow set is in-memory (a restart forgets threads until the next mention; the mention catch-up pass repopulates active ones), and the respond-to author gate applies unchanged — in owner-only mode only the owner's un-mentioned replies are admitted. Co-Authored-By: Claude Fable 5 Signed-off-by: Niko Moritz <126672558+nikomoritz@users.noreply.github.com> --- crates/buzz-acp/src/config.rs | 31 +++- crates/buzz-acp/src/filter.rs | 35 +++-- crates/buzz-acp/src/lib.rs | 41 ++++- crates/buzz-acp/src/setup_mode.rs | 1 + crates/buzz-acp/src/thread_follow.rs | 225 +++++++++++++++++++++++++++ 5 files changed, 319 insertions(+), 14 deletions(-) create mode 100644 crates/buzz-acp/src/thread_follow.rs diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index d959685846..6ed7c7cf21 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -333,6 +333,13 @@ pub struct CliArgs { #[arg(long, env = "BUZZ_ACP_NO_MENTION_FILTER")] pub no_mention_filter: bool, + /// Follow threads the agent participates in: after the agent is + /// @mentioned in a thread (or has replied there itself), subsequent + /// replies in that thread trigger turns without a fresh @mention. + /// Mentions mode only; ignored in all/config modes. + #[arg(long, env = "BUZZ_ACP_THREAD_FOLLOW")] + pub thread_follow: bool, + #[arg(long, env = "BUZZ_ACP_CONFIG", default_value = "./buzz-acp.toml")] pub config: PathBuf, @@ -514,6 +521,11 @@ pub struct Config { pub kinds_override: Option>, pub channels_override: Option>, pub no_mention_filter: bool, + /// Follow threads the agent participates in (see the `--thread-follow` + /// CLI flag). When set, mentions-mode subscriptions drop the server-side + /// `#p` filter and the event loop admits un-mentioned replies whose + /// thread root is in the agent's follow set. + pub thread_follow: bool, pub config_path: PathBuf, pub context_message_limit: u32, /// Maximum turns per session before proactive rotation. 0 = disabled. @@ -896,6 +908,12 @@ impl Config { if args.no_mention_filter { tracing::warn!("--no-mention-filter is ignored in config mode"); } + if args.thread_follow { + tracing::warn!("--thread-follow is ignored in config mode"); + } + } + if args.thread_follow && matches!(args.subscribe, SubscribeMode::All) { + tracing::warn!("--thread-follow is a no-op in all mode (every event already triggers)"); } let agent_command = args.agent_command; @@ -1081,6 +1099,7 @@ impl Config { kinds_override: args.kinds, channels_override: args.channels, no_mention_filter: args.no_mention_filter, + thread_follow: args.thread_follow, config_path: args.config, context_message_limit: args.context_message_limit, max_turns_per_session: args.max_turns_per_session, @@ -1125,7 +1144,7 @@ impl Config { format!(" allowed_respond_to=[{}]", modes.join(",")) }; format!( - "relay={} pubkey={} agent_cmd={} {} mcp_cmd={} idle_timeout={}s max_turn={}s agents={} heartbeat={}s subscribe={:?} dedup={:?} meh={:?} ignore_self={} context_limit={} max_turns_per_session={} presence={} typing={} memory={} model={} permission_mode={} {}{}", + "relay={} pubkey={} agent_cmd={} {} mcp_cmd={} idle_timeout={}s max_turn={}s agents={} heartbeat={}s subscribe={:?} dedup={:?} meh={:?} ignore_self={} thread_follow={} context_limit={} max_turns_per_session={} presence={} typing={} memory={} model={} permission_mode={} {}{}", self.relay_url, self.keys.public_key().to_hex(), self.agent_command, @@ -1139,6 +1158,7 @@ impl Config { self.dedup_mode, self.multiple_event_handling, self.ignore_self, + self.thread_follow, self.context_message_limit, self.max_turns_per_session, self.presence_enabled, @@ -1262,7 +1282,11 @@ pub fn resolve_channel_filters( KIND_STREAM_REMINDER, ] }); - let require_mention = !config.no_mention_filter; + // Thread-follow needs to see un-mentioned replies, so the + // server-side `#p` filter must come off; the event loop + // re-applies the mention gate client-side (with the + // follow-set exemption). + let require_mention = !config.no_mention_filter && !config.thread_follow; for ch in &target_channels { result.insert( *ch, @@ -1367,7 +1391,7 @@ pub fn resolve_dynamic_channel_filter( KIND_STREAM_REMINDER, ] })), - require_mention: !config.no_mention_filter, + require_mention: !config.no_mention_filter && !config.thread_follow, }), SubscribeMode::All => Some(ChannelFilter { kinds: config.kinds_override.clone(), @@ -1455,6 +1479,7 @@ mod tests { kinds_override: None, channels_override: None, no_mention_filter: false, + thread_follow: false, config_path: PathBuf::from("./buzz-acp.toml"), context_message_limit: 12, max_turns_per_session: 0, diff --git a/crates/buzz-acp/src/filter.rs b/crates/buzz-acp/src/filter.rs index 43edd969dd..512398c58b 100644 --- a/crates/buzz-acp/src/filter.rs +++ b/crates/buzz-acp/src/filter.rs @@ -351,7 +351,9 @@ const MAX_CONSECUTIVE_TIMEOUTS: u32 = 5; /// 2. **kinds** — if non-empty, the event kind must be in the list. /// 3. **require_mention** — if `true`, a `p` tag matching `agent_pubkey_hex` must /// exist. Tag kind is checked via `tag.as_slice()` for stable, library-independent -/// access. +/// access. `mention_exempt` waives this check for the whole call — the +/// thread-follow path sets it when the event's thread root is already +/// followed by the agent. /// 4. **filter** — if `Some`, the evalexpr expression must evaluate to `true`. /// /// # Fail-closed filter error handling @@ -370,6 +372,7 @@ pub async fn match_event( channel_id: uuid::Uuid, rules: &[SubscriptionRule], agent_pubkey_hex: &str, + mention_exempt: bool, ) -> Option { let filter_ctx = FilterContext::from_event(event, channel_id); @@ -387,7 +390,7 @@ pub async fn match_event( // 3. Mention check — look for a `p` tag whose first element equals // agent_pubkey_hex. Uses tag.as_slice() for stable, library-independent // access — avoids relying on the Display impl of tag kind. - if rule.require_mention { + if rule.require_mention && !mention_exempt { let mentioned = event.tags.iter().any(|tag| { let s = tag.as_slice(); s.first().map(|k| k.as_str()) == Some("p") @@ -601,7 +604,9 @@ mod tests { ), ]; - let matched = match_event(&event, channel_id, &rules, "").await.unwrap(); + let matched = match_event(&event, channel_id, &rules, "", false) + .await + .unwrap(); assert_eq!(matched.rule_index, 0); assert_eq!(matched.prompt_tag, "tag-first"); } @@ -630,7 +635,9 @@ mod tests { ), ]; - let matched = match_event(&event, channel_id, &rules, "").await.unwrap(); + let matched = match_event(&event, channel_id, &rules, "", false) + .await + .unwrap(); assert_eq!(matched.rule_index, 1); assert_eq!(matched.prompt_tag, "matched"); } @@ -653,11 +660,17 @@ mod tests { )]; // Without mention — no match. - let result = match_event(&event_no_mention, channel_id, &rules, agent_pubkey).await; + let result = match_event(&event_no_mention, channel_id, &rules, agent_pubkey, false).await; assert!(result.is_none()); // With mention — matches. - let matched = match_event(&event_with_mention, channel_id, &rules, agent_pubkey) + let matched = match_event(&event_with_mention, channel_id, &rules, agent_pubkey, false) + .await + .unwrap(); + assert_eq!(matched.prompt_tag, "mentioned"); + + // Without mention but mention-exempt (thread-follow) — matches. + let matched = match_event(&event_no_mention, channel_id, &rules, agent_pubkey, true) .await .unwrap(); assert_eq!(matched.prompt_tag, "mentioned"); @@ -677,7 +690,7 @@ mod tests { None, )]; - let result = match_event(&event, channel_id, &rules, "").await; + let result = match_event(&event, channel_id, &rules, "", false).await; assert!(result.is_none()); } @@ -725,7 +738,9 @@ mod tests { None, // no explicit tag )]; - let matched = match_event(&event, channel_id, &rules, "").await.unwrap(); + let matched = match_event(&event, channel_id, &rules, "", false) + .await + .unwrap(); assert_eq!(matched.prompt_tag, "my-rule"); } @@ -755,7 +770,7 @@ mod tests { ]; // Must return None — not "catch-all". - let result = match_event(&event, channel_id, &rules, "").await; + let result = match_event(&event, channel_id, &rules, "", false).await; assert!( result.is_none(), "filter error must fail closed, not fall through to next rule" @@ -781,7 +796,7 @@ mod tests { .store(MAX_CONSECUTIVE_TIMEOUTS, Ordering::Relaxed); let rules = vec![rule]; - let result = match_event(&event, channel_id, &rules, "").await; + let result = match_event(&event, channel_id, &rules, "", false).await; assert!(result.is_none(), "disabled rule must return None"); } } diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 65c9dd6203..d5de57d3f3 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -10,6 +10,7 @@ mod pool_lifecycle; mod queue; mod relay; mod setup_mode; +mod thread_follow; mod usage; pub use usage::TurnUsage; @@ -1755,6 +1756,10 @@ async fn tokio_main() -> Result<()> { } }; + // Thread-follow: bounded set of thread roots the agent participates in. + // Only consulted when `config.thread_follow` is set. + let mut thread_follow_set = thread_follow::ThreadFollowSet::new(thread_follow::DEFAULT_CAP); + let channel_filters = config::resolve_channel_filters(&config, &channel_ids, &rules); if channel_filters.is_empty() { tracing::warn!("no channel subscriptions resolved — agent will sit idle"); @@ -2327,6 +2332,14 @@ async fn tokio_main() -> Result<()> { } if config.ignore_self && buzz_event.event.pubkey.to_hex() == pubkey_hex { + // Learn from our own replies before dropping: + // a self-authored reply marks its thread as + // followed, so thread-follow can admit later + // un-mentioned replies there. + if config.thread_follow { + thread_follow_set + .insert(thread_follow::followable_root(&buzz_event.event)); + } tracing::debug!(channel_id = %buzz_event.channel_id, "dropping self-authored event"); continue; } @@ -2473,8 +2486,32 @@ async fn tokio_main() -> Result<()> { } } - let matched = filter::match_event(&buzz_event.event, buzz_event.channel_id, &rules, &pubkey_hex).await; + // Thread-follow: learn roots from events that + // mention the agent (the author gate has already + // passed), then exempt events in followed threads + // from the mention requirement. + let mention_exempt = if config.thread_follow { + if thread_follow::mentions_pubkey(&buzz_event.event, &pubkey_hex) { + thread_follow_set + .insert(thread_follow::followable_root(&buzz_event.event)); + false + } else { + thread_follow::thread_root_of(&buzz_event.event) + .is_some_and(|root| thread_follow_set.contains(&root)) + } + } else { + false + }; + let matched = filter::match_event(&buzz_event.event, buzz_event.channel_id, &rules, &pubkey_hex, mention_exempt).await; let prompt_tag = match matched { + Some(_) if mention_exempt => { + tracing::debug!( + channel_id = %buzz_event.channel_id, + followed_threads = thread_follow_set.len(), + "thread-follow — admitting un-mentioned thread reply" + ); + "thread-follow".to_string() + } Some(m) => m.prompt_tag, None => { tracing::debug!(channel_id = %buzz_event.channel_id, kind = buzz_event.event.kind.as_u16(), "event matched no rule — dropping"); @@ -6191,6 +6228,7 @@ mod build_mcp_servers_tests { kinds_override: None, channels_override: None, no_mention_filter: false, + thread_follow: false, config_path: std::path::PathBuf::from("./buzz-acp.toml"), context_message_limit: 12, max_turns_per_session: 0, @@ -6413,6 +6451,7 @@ mod error_outcome_emission_tests { kinds_override: None, channels_override: None, no_mention_filter: false, + thread_follow: false, config_path: std::path::PathBuf::from("./buzz-acp.toml"), context_message_limit: 12, max_turns_per_session: 0, diff --git a/crates/buzz-acp/src/setup_mode.rs b/crates/buzz-acp/src/setup_mode.rs index b1a9372ea4..c8d2b72a1c 100644 --- a/crates/buzz-acp/src/setup_mode.rs +++ b/crates/buzz-acp/src/setup_mode.rs @@ -446,6 +446,7 @@ pub(crate) async fn run_setup_listener(config: Config, payload: SetupPayload) -> buzz_event.channel_id, &rules, &pubkey_hex, + false, ) .await .is_some(); diff --git a/crates/buzz-acp/src/thread_follow.rs b/crates/buzz-acp/src/thread_follow.rs new file mode 100644 index 0000000000..9e6ee35c39 --- /dev/null +++ b/crates/buzz-acp/src/thread_follow.rs @@ -0,0 +1,225 @@ +//! Thread-follow support: a bounded set of thread roots the agent +//! participates in. +//! +//! When `--thread-follow` is enabled (mentions mode only), the harness +//! tracks the NIP-10 roots of threads where the agent was @mentioned or +//! has itself replied. Subsequent replies in a followed thread trigger +//! turns without requiring a fresh @mention. +//! +//! The set is in-memory and bounded: once `cap` roots are tracked, the +//! oldest-inserted root is evicted. A harness restart starts with an +//! empty set; the mention catch-up pass repopulates roots for threads +//! with recent agent mentions. + +use std::collections::{HashSet, VecDeque}; + +/// Default capacity for the harness's follow set. Generous for a single +/// agent (a root is ~64 bytes, so the ceiling is a few hundred KB) while +/// still bounding memory over long uptimes. +pub const DEFAULT_CAP: usize = 1024; + +/// Bounded, insertion-ordered set of thread-root event ids (hex). +/// +/// Deliberately dependency-free: `HashSet` for membership plus a +/// `VecDeque` for FIFO eviction. Re-inserting a known root is a no-op +/// (it does not refresh eviction order); with a generous capacity the +/// simpler semantics are worth it. +pub struct ThreadFollowSet { + set: HashSet, + order: VecDeque, + cap: usize, +} + +impl ThreadFollowSet { + /// Create a set that tracks at most `cap` roots (minimum 1). + pub fn new(cap: usize) -> Self { + let cap = cap.max(1); + Self { + set: HashSet::with_capacity(cap), + order: VecDeque::with_capacity(cap), + cap, + } + } + + /// Track a thread root. Evicts the oldest root when full. + pub fn insert(&mut self, root: String) { + if self.set.contains(&root) { + return; + } + while self.set.len() >= self.cap { + if let Some(oldest) = self.order.pop_front() { + self.set.remove(&oldest); + } else { + break; + } + } + self.order.push_back(root.clone()); + self.set.insert(root); + } + + /// Whether `root` is currently followed. + pub fn contains(&self, root: &str) -> bool { + self.set.contains(root) + } + + /// Number of tracked roots (for logging/tests). + pub fn len(&self) -> usize { + self.set.len() + } + + /// Whether the set is empty. + #[cfg_attr(not(test), allow(dead_code))] + pub fn is_empty(&self) -> bool { + self.set.is_empty() + } +} + +/// Extract the NIP-10 thread root of `event`, if it is a thread reply. +/// +/// Preference order: +/// 1. `["e", , .., "root"]` — explicit root marker. +/// 2. `["e", , .., "reply"]` — reply without a root marker; the +/// referenced parent is treated as the root (single-level thread). +/// 3. First bare `["e", ]` tag — deprecated positional style. +/// +/// Returns `None` for top-level messages (no `e` tags). +pub fn thread_root_of(event: &nostr::Event) -> Option { + let mut root: Option<&str> = None; + let mut reply: Option<&str> = None; + let mut bare: Option<&str> = None; + for tag in event.tags.iter() { + let s = tag.as_slice(); + if s.first().map(|k| k.as_str()) != Some("e") { + continue; + } + let Some(id) = s.get(1).map(|v| v.as_str()) else { + continue; + }; + match s.get(3).map(|m| m.as_str()) { + Some("root") => root = root.or(Some(id)), + Some("reply") => reply = reply.or(Some(id)), + _ => bare = bare.or(Some(id)), + } + } + root.or(reply).or(bare).map(str::to_owned) +} + +/// The root under which `event`'s thread should be followed: its NIP-10 +/// root when it is a reply, otherwise its own id (a top-level message +/// starts a thread rooted at itself). +pub fn followable_root(event: &nostr::Event) -> String { + thread_root_of(event).unwrap_or_else(|| event.id.to_hex()) +} + +/// Whether `event` carries a `p` tag for `pubkey_hex`. +pub fn mentions_pubkey(event: &nostr::Event, pubkey_hex: &str) -> bool { + event.tags.iter().any(|tag| { + let s = tag.as_slice(); + s.first().map(|k| k.as_str()) == Some("p") + && s.get(1).map(|v| v.as_str()) == Some(pubkey_hex) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use nostr::{EventBuilder, Keys, Kind, Tag}; + + fn event_with_tags(tags: Vec) -> nostr::Event { + let keys = Keys::generate(); + EventBuilder::new(Kind::Custom(9), "hello") + .tags(tags) + .sign_with_keys(&keys) + .unwrap() + } + + fn e_tag(id: &str, marker: Option<&str>) -> Tag { + let mut parts = vec!["e".to_string(), id.to_string()]; + if let Some(m) = marker { + parts.push(String::new()); + parts.push(m.to_string()); + } + Tag::parse(parts).unwrap() + } + + const ROOT: &str = "1111111111111111111111111111111111111111111111111111111111111111"; + const PARENT: &str = "2222222222222222222222222222222222222222222222222222222222222222"; + + #[test] + fn root_marker_wins() { + let event = event_with_tags(vec![ + e_tag(ROOT, Some("root")), + e_tag(PARENT, Some("reply")), + ]); + assert_eq!(thread_root_of(&event).as_deref(), Some(ROOT)); + } + + #[test] + fn reply_marker_without_root() { + let event = event_with_tags(vec![e_tag(PARENT, Some("reply"))]); + assert_eq!(thread_root_of(&event).as_deref(), Some(PARENT)); + } + + #[test] + fn bare_e_tag_fallback() { + let event = event_with_tags(vec![e_tag(ROOT, None)]); + assert_eq!(thread_root_of(&event).as_deref(), Some(ROOT)); + } + + #[test] + fn top_level_has_no_root_and_follows_itself() { + let event = event_with_tags(vec![]); + assert_eq!(thread_root_of(&event), None); + assert_eq!(followable_root(&event), event.id.to_hex()); + } + + #[test] + fn mentions_pubkey_matches_p_tag() { + let keys = Keys::generate(); + let pk = keys.public_key(); + let event = event_with_tags(vec![Tag::public_key(pk)]); + assert!(mentions_pubkey(&event, &pk.to_hex())); + assert!(!mentions_pubkey(&event, ROOT)); + } + + #[test] + fn follow_set_inserts_and_contains() { + let mut set = ThreadFollowSet::new(4); + assert!(set.is_empty()); + set.insert(ROOT.into()); + assert!(!set.is_empty()); + assert!(set.contains(ROOT)); + assert!(!set.contains(PARENT)); + assert_eq!(set.len(), 1); + } + + #[test] + fn follow_set_reinsert_is_noop() { + let mut set = ThreadFollowSet::new(4); + set.insert(ROOT.into()); + set.insert(ROOT.into()); + assert_eq!(set.len(), 1); + } + + #[test] + fn follow_set_evicts_oldest_at_cap() { + let mut set = ThreadFollowSet::new(2); + set.insert("a".into()); + set.insert("b".into()); + set.insert("c".into()); + assert_eq!(set.len(), 2); + assert!(!set.contains("a")); + assert!(set.contains("b")); + assert!(set.contains("c")); + } + + #[test] + fn follow_set_zero_cap_clamps_to_one() { + let mut set = ThreadFollowSet::new(0); + set.insert("a".into()); + assert!(set.contains("a")); + set.insert("b".into()); + assert!(!set.contains("a")); + assert!(set.contains("b")); + } +}