From f2ef18bee915fe939a49e5c4ee941946e1ad7842 Mon Sep 17 00:00:00 2001 From: Broc Oppler Date: Tue, 21 Jul 2026 22:19:09 -0700 Subject: [PATCH 1/3] fix(acp): follow threads the agent participates in (#2270) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A mention-gated agent (SubscribeMode::Mentions, the default) only ever received events carrying its #p tag, at two independent gates: the relay REQ filter and match_event's re-check. Once the agent replied in a thread, untagged follow-ups in that same thread were never delivered — the agent went deaf mid-conversation unless re-mentioned on every message. This records thread participation at dispatch time and opens both gates for followed threads only: - thread_follow.rs: per-channel set of participated NIP-10 thread roots with a sliding inactivity TTL (default 24h, --thread-follow-ttl-secs), a per-channel cap (64, stalest evicted), and dirty-channel tracking. Participation is recorded when a batch is dispatched — the batch's thread root, or the triggering event id when the reply starts a new thread (mirroring resolve_reply_anchor). - relay.rs: REQ construction extracted into build_req_filters(), which emits a second filter clause (#h + #e limited to followed roots, same kinds/since) when the base clause is mention-gated. The mention gate stays intact for everything else; re-issued REQs replay from last_seen minus skew, so replies landing during the update window are recovered. - filter.rs: match_event takes the followed-root set and admits an unmentioned event whose thread root is followed. Following relaxes only the mention gate — channel scope, kinds, and expression filters still apply. - Followed threads expire after the TTL and the REQ clause is narrowed again; channel unsubscribe clears its follow state. --no-follow-threads restores the previous strict-mention behavior. Closes #2270. Co-Authored-By: Claude Fable 5 Signed-off-by: Broc Oppler --- crates/buzz-acp/src/config.rs | 37 ++++ crates/buzz-acp/src/filter.rs | 133 ++++++++++- crates/buzz-acp/src/lib.rs | 94 +++++++- crates/buzz-acp/src/relay.rs | 143 ++++++++++-- crates/buzz-acp/src/setup_mode.rs | 3 + crates/buzz-acp/src/thread_follow.rs | 320 +++++++++++++++++++++++++++ 6 files changed, 694 insertions(+), 36 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 a38d6faa14..ee1f39ff0b 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -338,6 +338,22 @@ pub struct CliArgs { #[arg(long, env = "BUZZ_ACP_NO_MENTION_FILTER")] pub no_mention_filter: bool, + /// Disable thread-following. With this set, a mention-gated agent only + /// sees messages that explicitly @mention it — even untagged replies in + /// threads it is already participating in (the pre-#2270 behavior). + #[arg(long, env = "BUZZ_ACP_NO_FOLLOW_THREADS")] + pub no_follow_threads: bool, + + /// Sliding inactivity TTL, in seconds, for followed threads. A thread the + /// agent participated in stops being followed after this long without + /// activity; a fresh @mention re-joins it. + #[arg( + long, + env = "BUZZ_ACP_THREAD_FOLLOW_TTL_SECS", + default_value_t = 86_400 + )] + pub thread_follow_ttl_secs: u64, + #[arg(long, env = "BUZZ_ACP_CONFIG", default_value = "./buzz-acp.toml")] pub config: PathBuf, @@ -480,6 +496,12 @@ pub struct ChannelFilter { pub kinds: Option>, /// Whether to include `#p` tag filter for agent pubkey. pub require_mention: bool, + /// Thread roots the agent follows in this channel. When non-empty and + /// `require_mention` is set, the REQ carries a second filter clause + /// (`#h` + `#e`) so untagged replies in these threads are delivered + /// alongside explicit mentions. Ignored when `require_mention` is false + /// (the mention-less clause already delivers everything). + pub thread_roots: Vec, } #[derive(Debug)] @@ -509,6 +531,11 @@ pub struct Config { pub kinds_override: Option>, pub channels_override: Option>, pub no_mention_filter: bool, + /// Whether the agent follows threads it has participated in (#2270). + /// Only consulted in Mentions mode — All mode delivers everything anyway. + pub follow_threads: bool, + /// Sliding inactivity TTL for followed thread roots, in seconds. + pub thread_follow_ttl_secs: u64, pub config_path: PathBuf, pub context_message_limit: u32, /// Maximum turns per session before proactive rotation. 0 = disabled. @@ -985,6 +1012,8 @@ impl Config { kinds_override: args.kinds, channels_override: args.channels, no_mention_filter: args.no_mention_filter, + follow_threads: !args.no_follow_threads, + thread_follow_ttl_secs: args.thread_follow_ttl_secs, config_path: args.config, context_message_limit: args.context_message_limit, max_turns_per_session: args.max_turns_per_session, @@ -1168,6 +1197,7 @@ pub fn resolve_channel_filters( ChannelFilter { kinds: Some(kinds.clone()), require_mention, + thread_roots: Vec::new(), }, ); } @@ -1179,6 +1209,7 @@ pub fn resolve_channel_filters( ChannelFilter { kinds: config.kinds_override.clone(), require_mention: false, + thread_roots: Vec::new(), }, ); } @@ -1214,6 +1245,7 @@ pub fn resolve_channel_filters( ChannelFilter { kinds: merged_kinds, require_mention, + thread_roots: Vec::new(), }, ); } @@ -1267,10 +1299,12 @@ pub fn resolve_dynamic_channel_filter( ] })), require_mention: !config.no_mention_filter, + thread_roots: Vec::new(), }), SubscribeMode::All => Some(ChannelFilter { kinds: config.kinds_override.clone(), require_mention: false, + thread_roots: Vec::new(), }), SubscribeMode::Config => { // Same merge logic as resolve_channel_filters() Config branch: @@ -1308,6 +1342,7 @@ pub fn resolve_dynamic_channel_filter( Some(ChannelFilter { kinds: merged_kinds, require_mention, + thread_roots: Vec::new(), }) } } @@ -1354,6 +1389,8 @@ mod tests { kinds_override: None, channels_override: None, no_mention_filter: false, + follow_threads: true, + thread_follow_ttl_secs: 86_400, 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..1b2d813480 100644 --- a/crates/buzz-acp/src/filter.rs +++ b/crates/buzz-acp/src/filter.rs @@ -5,6 +5,7 @@ //! - Evaluating boolean filter expressions with a hard timeout //! - Matching events against ordered subscription rules (first match wins) +use std::collections::HashSet; use std::sync::atomic::{AtomicU32, Ordering}; use std::sync::Arc; use std::time::Duration; @@ -351,7 +352,10 @@ 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. An unmentioned event still passes this check when its NIP-10 +/// thread root is in `followed_roots` — the agent is participating in that +/// thread, so untagged replies stay audible (#2270). Following relaxes only +/// the mention gate; channel scope, kinds, and expression filters still apply. /// 4. **filter** — if `Some`, the evalexpr expression must evaluate to `true`. /// /// # Fail-closed filter error handling @@ -370,9 +374,18 @@ pub async fn match_event( channel_id: uuid::Uuid, rules: &[SubscriptionRule], agent_pubkey_hex: &str, + followed_roots: Option<&HashSet>, ) -> Option { let filter_ctx = FilterContext::from_event(event, channel_id); + // Resolved once per event: the NIP-10 thread root, iff the caller passed a + // followed set and the root is in it. `None` either way otherwise. + let followed_thread_root: Option = followed_roots.and_then(|roots| { + crate::queue::parse_thread_tags(event) + .root_event_id + .filter(|root| roots.contains(root)) + }); + for (index, rule) in rules.iter().enumerate() { // 1. Channel scope check. if !rule.channels.matches(&channel_id) { @@ -393,7 +406,7 @@ pub async fn match_event( s.first().map(|k| k.as_str()) == Some("p") && s.get(1).map(|v| v.as_str()) == Some(agent_pubkey_hex) }); - if !mentioned { + if !mentioned && followed_thread_root.is_none() { continue; } } @@ -484,6 +497,100 @@ mod tests { .unwrap() } + /// Build a test event that is an untagged reply in a NIP-10 thread + /// (marker-based `e` root tag, no `p` mention). + fn make_event_in_thread(kind: u32, content: &str, root_hex: &str) -> nostr::Event { + let keys = Keys::generate(); + let e_tag = Tag::parse(["e", root_hex, "", "root"]).expect("tag parse"); + EventBuilder::new(Kind::Custom(kind as u16), content) + .tags([e_tag]) + .sign_with_keys(&keys) + .unwrap() + } + + #[tokio::test] + async fn test_match_event_followed_thread_admits_unmentioned_reply() { + let channel_id = any_channel(); + let rules = vec![make_rule( + "mentions", + ChannelScope::All("all".into()), + vec![9], + true, + None, + None, + )]; + let root = "a".repeat(64); + let event = make_event_in_thread(9, "yes, go ahead", &root); + + // Without a followed set the reply is gated (pre-#2270 behavior). + assert!(match_event(&event, channel_id, &rules, "agent-pk", None) + .await + .is_none()); + + // With the thread followed, the same unmentioned reply is admitted. + let followed: HashSet = [root].into_iter().collect(); + let matched = match_event(&event, channel_id, &rules, "agent-pk", Some(&followed)).await; + assert!(matched.is_some()); + } + + #[tokio::test] + async fn test_match_event_followed_thread_ignores_other_threads() { + let channel_id = any_channel(); + let rules = vec![make_rule( + "mentions", + ChannelScope::All("all".into()), + vec![9], + true, + None, + None, + )]; + let followed: HashSet = [("b".repeat(64))].into_iter().collect(); + + // Reply in a different thread stays gated. + let other_thread = make_event_in_thread(9, "unrelated", &"c".repeat(64)); + assert!(match_event( + &other_thread, + channel_id, + &rules, + "agent-pk", + Some(&followed) + ) + .await + .is_none()); + + // Top-level message with no thread tags stays gated too. + let top_level = make_event(9, "no thread here"); + assert!( + match_event(&top_level, channel_id, &rules, "agent-pk", Some(&followed)) + .await + .is_none() + ); + } + + #[tokio::test] + async fn test_match_event_followed_thread_relaxes_only_the_mention_gate() { + let channel_id = any_channel(); + // Rule matches kind 9 only — a followed-thread event of kind 45001 + // must still be dropped: following bypasses the mention check, not + // the kind or channel gates. + let rules = vec![make_rule( + "mentions", + ChannelScope::All("all".into()), + vec![9], + true, + None, + None, + )]; + let root = "d".repeat(64); + let followed: HashSet = [root.clone()].into_iter().collect(); + let wrong_kind = make_event_in_thread(45001, "still gated", &root); + assert!( + match_event(&wrong_kind, channel_id, &rules, "agent-pk", Some(&followed)) + .await + .is_none() + ); + } + fn any_channel() -> Uuid { Uuid::new_v4() } @@ -601,7 +708,9 @@ mod tests { ), ]; - let matched = match_event(&event, channel_id, &rules, "").await.unwrap(); + let matched = match_event(&event, channel_id, &rules, "", None) + .await + .unwrap(); assert_eq!(matched.rule_index, 0); assert_eq!(matched.prompt_tag, "tag-first"); } @@ -630,7 +739,9 @@ mod tests { ), ]; - let matched = match_event(&event, channel_id, &rules, "").await.unwrap(); + let matched = match_event(&event, channel_id, &rules, "", None) + .await + .unwrap(); assert_eq!(matched.rule_index, 1); assert_eq!(matched.prompt_tag, "matched"); } @@ -653,11 +764,11 @@ 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, None).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, None) .await .unwrap(); assert_eq!(matched.prompt_tag, "mentioned"); @@ -677,7 +788,7 @@ mod tests { None, )]; - let result = match_event(&event, channel_id, &rules, "").await; + let result = match_event(&event, channel_id, &rules, "", None).await; assert!(result.is_none()); } @@ -725,7 +836,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, "", None) + .await + .unwrap(); assert_eq!(matched.prompt_tag, "my-rule"); } @@ -755,7 +868,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, "", None).await; assert!( result.is_none(), "filter error must fail closed, not fall through to next rule" @@ -781,7 +894,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, "", None).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 862732f478..1a328f2aeb 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; @@ -1478,6 +1479,12 @@ async fn tokio_main() -> Result<()> { ); } + // Thread-following (#2270): track the threads this agent participates in + // so untagged replies in them stay audible under mention gating. Disabled + // trackers are inert — every call below becomes a no-op. + let mut thread_follow = + thread_follow::ThreadFollowState::new(config.thread_follow_ttl_secs, config.follow_threads); + let base_prompt_content = config.base_prompt_content.take(); let ctx = Arc::new(PromptContext { mcp_servers: build_mcp_servers(&config), @@ -1693,6 +1700,37 @@ async fn tokio_main() -> Result<()> { } } + // #2270: keep followed-thread REQ clauses in sync. take_dirty() is + // empty on almost every iteration; when a dispatch records a new + // thread (or a root expires) the affected channels get their + // subscription re-issued with an updated `#e` clause. The re-REQ + // replays from last_seen minus skew, so replies that land between + // dispatch and re-subscribe are recovered by replay, not lost. + // Runs regardless of pool readiness — this is relay subscription + // state; expiry-narrowing must proceed while a lazy pool sleeps. + thread_follow.compact_if_due(); + for ch in thread_follow.take_dirty() { + if !subscribed_channel_ids.contains(&ch) { + continue; + } + let base = channel_filters + .get(&ch) + .cloned() + .or_else(|| config::resolve_dynamic_channel_filter(&config, ch, &rules)); + if let Some(mut filter) = base { + if filter.require_mention { + filter.thread_roots = thread_follow.roots_for_filter(ch); + if let Err(e) = relay.subscribe_channel(ch, filter).await { + tracing::warn!( + channel_id = %ch, + error = %e, + "failed to refresh followed-thread subscription" + ); + } + } + } + } + if pool_ready && last_maintenance.elapsed() >= maintenance_interval { last_maintenance = std::time::Instant::now(); queue.compact_expired_state(); @@ -1727,7 +1765,9 @@ async fn tokio_main() -> Result<()> { // called on relay events or pool results, neither of which // arrive when the channel is silent. if queue.has_flushable_work() { - for (channel_id, thread_tags) in dispatch_pending(&mut pool, &mut queue, &ctx) { + for (channel_id, thread_tags) in + dispatch_pending(&mut pool, &mut queue, &ctx, &mut thread_follow) + { typing_channels.insert(channel_id, thread_tags); } } @@ -1763,7 +1803,9 @@ async fn tokio_main() -> Result<()> { // this, batches requeued during crash recovery sit idle until the // next relay event arrives — which can be minutes on quiet channels. if respawn_collected { - for (channel_id, thread_tags) in dispatch_pending(&mut pool, &mut queue, &ctx) { + for (channel_id, thread_tags) in + dispatch_pending(&mut pool, &mut queue, &ctx, &mut thread_follow) + { typing_channels.insert(channel_id, thread_tags); } } @@ -1945,6 +1987,7 @@ async fn tokio_main() -> Result<()> { } else { 0 }; + thread_follow.clear_channel(ch); // Track removed channels so checked-out agents get // their sessions stripped when they return to the pool. removed_channels.insert(ch); @@ -2116,7 +2159,10 @@ async fn tokio_main() -> Result<()> { } } - let matched = filter::match_event(&buzz_event.event, buzz_event.channel_id, &rules, &pubkey_hex).await; + let followed_roots = thread_follow + .enabled() + .then(|| thread_follow.live_roots(buzz_event.channel_id)); + let matched = filter::match_event(&buzz_event.event, buzz_event.channel_id, &rules, &pubkey_hex, followed_roots.as_ref()).await; let prompt_tag = match matched { Some(m) => m.prompt_tag, None => { @@ -2124,6 +2170,15 @@ async fn tokio_main() -> Result<()> { continue; } }; + // An admitted event in a followed thread is live + // conversation — slide that thread's TTL forward. + if thread_follow.enabled() { + if let Some(root) = + queue::parse_thread_tags(&buzz_event.event).root_event_id + { + thread_follow.touch(buzz_event.channel_id, &root); + } + } // Capture author pubkey before queue.push() moves // buzz_event.event (needed for mode gate below). let author_hex = buzz_event.event.pubkey.to_hex(); @@ -2204,7 +2259,7 @@ async fn tokio_main() -> Result<()> { } if pool_ready { for (channel_id, thread_tags) in - dispatch_pending(&mut pool, &mut queue, &ctx) + dispatch_pending(&mut pool, &mut queue, &ctx, &mut thread_follow) { typing_channels.insert(channel_id, thread_tags); } @@ -2233,7 +2288,7 @@ async fn tokio_main() -> Result<()> { } else if queue.has_flushable_work() { tracing::debug!("heartbeat_skipped_events"); for (channel_id, thread_tags) in - dispatch_pending(&mut pool, &mut queue, &ctx) + dispatch_pending(&mut pool, &mut queue, &ctx, &mut thread_follow) { typing_channels.insert(channel_id, thread_tags); } @@ -2331,7 +2386,9 @@ async fn tokio_main() -> Result<()> { { break; } - for (channel_id, thread_tags) in dispatch_pending(&mut pool, &mut queue, &ctx) { + for (channel_id, thread_tags) in + dispatch_pending(&mut pool, &mut queue, &ctx, &mut thread_follow) + { typing_channels.insert(channel_id, thread_tags); } } @@ -2354,7 +2411,9 @@ async fn tokio_main() -> Result<()> { tracing::error!("all agents dead — exiting"); break; } - for (channel_id, thread_tags) in dispatch_pending(&mut pool, &mut queue, &ctx) { + for (channel_id, thread_tags) in + dispatch_pending(&mut pool, &mut queue, &ctx, &mut thread_follow) + { typing_channels.insert(channel_id, thread_tags); } } @@ -2474,7 +2533,9 @@ async fn tokio_main() -> Result<()> { // tear down the in-flight task; on its completion the // queue drains. We still try here in case the in-flight // task has already returned. - for (channel_id, thread_tags) in dispatch_pending(&mut pool, &mut queue, &ctx) { + for (channel_id, thread_tags) in + dispatch_pending(&mut pool, &mut queue, &ctx, &mut thread_follow) + { typing_channels.insert(channel_id, thread_tags); } } @@ -2835,6 +2896,7 @@ fn dispatch_pending( pool: &mut AgentPool, queue: &mut EventQueue, ctx: &Arc, + follow: &mut thread_follow::ThreadFollowState, ) -> Vec<(Uuid, ThreadTags)> { let mut dispatched_channels = Vec::new(); loop { @@ -2848,6 +2910,15 @@ fn dispatch_pending( .last() .map(|event| queue::parse_thread_tags(&event.event)) .unwrap_or_default(); + // #2270: the thread this dispatch participates in — the batch's NIP-10 + // root, or the triggering event itself when the agent's reply will + // start a new thread (mirrors resolve_reply_anchor's root resolution). + let follow_root = batch.events.last().map(|be| { + typing_scope + .root_event_id + .clone() + .unwrap_or_else(|| be.event.id.to_hex()) + }); let affinity_hit = pool.has_session_for(channel_id); let mut agent = match pool.try_claim(Some(channel_id)) { Some(a) => a, @@ -2913,6 +2984,9 @@ fn dispatch_pending( steer_tx, }, ); + if let Some(root) = follow_root { + follow.record(channel_id, &root); + } dispatched_channels.push((channel_id, typing_scope)); } tracing::debug!( @@ -4631,6 +4705,8 @@ mod build_mcp_servers_tests { kinds_override: None, channels_override: None, no_mention_filter: false, + follow_threads: true, + thread_follow_ttl_secs: 86_400, config_path: std::path::PathBuf::from("./buzz-acp.toml"), context_message_limit: 12, max_turns_per_session: 0, @@ -4797,6 +4873,8 @@ mod error_outcome_emission_tests { kinds_override: None, channels_override: None, no_mention_filter: false, + follow_threads: true, + thread_follow_ttl_secs: 86_400, 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/relay.rs b/crates/buzz-acp/src/relay.rs index ce7d565f12..6c9289631e 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -3136,40 +3136,78 @@ async fn wait_for_reconnect( } } -/// Send a NIP-01 REQ for a channel, built from a [`ChannelFilter`]. +/// Build the NIP-01 REQ filter clauses for a channel subscription. /// +/// Base clause (always present): /// - `kinds` is included only when `filter.kinds` is `Some`; `None` = wildcard. -/// - `#p` is included only when `filter.require_mention` is `true`. /// - `#h` is always included (channel-scoped subscription). -/// - On first subscribe (`since` is `None`) adds `since=now` to avoid replaying -/// history. On reconnect (`since` is `Some`) subtracts [`SINCE_SKEW_SECS`]. +/// - `#p` is included only when `filter.require_mention` is `true`. /// -/// Returns `true` if the REQ was successfully written to the WebSocket. -async fn send_subscribe( - ws: &mut WsStream, - _state: &BgState, +/// Followed-threads clause (#2270) — emitted only when `require_mention` is +/// `true` AND `filter.thread_roots` is non-empty: +/// - same `kinds` and `#h` scope, plus `#e` limited to the followed roots, so +/// the relay also delivers untagged replies in threads the agent +/// participates in. The mention gate stays intact for everything else; when +/// `require_mention` is `false` the base clause already delivers the whole +/// channel and no second clause is needed. +/// +/// Both clauses carry the same `since`. +fn build_req_filters( channel_id: Uuid, agent_pubkey_hex: &str, - since: Option, + since_ts: u64, filter: &ChannelFilter, -) -> bool { - let sub_id = channel_sub_id(channel_id); - - let mut req_filter = serde_json::Map::new(); +) -> Vec { + let mut base = serde_json::Map::new(); // kinds — omit entirely for wildcard subscriptions. if let Some(ref kinds) = filter.kinds { - req_filter.insert("kinds".into(), json!(kinds)); + base.insert("kinds".into(), json!(kinds)); } // #h — always present (channel scope). - req_filter.insert("#h".into(), json!([channel_id.to_string()])); + base.insert("#h".into(), json!([channel_id.to_string()])); // #p — only when require_mention is true. if filter.require_mention { - req_filter.insert("#p".into(), json!([agent_pubkey_hex])); + base.insert("#p".into(), json!([agent_pubkey_hex])); + } + + base.insert("since".into(), json!(since_ts)); + + let mut clauses = vec![Value::Object(base)]; + + if filter.require_mention && !filter.thread_roots.is_empty() { + let mut threads = serde_json::Map::new(); + if let Some(ref kinds) = filter.kinds { + threads.insert("kinds".into(), json!(kinds)); + } + threads.insert("#h".into(), json!([channel_id.to_string()])); + threads.insert("#e".into(), json!(filter.thread_roots)); + threads.insert("since".into(), json!(since_ts)); + clauses.push(Value::Object(threads)); } + clauses +} + +/// Send a NIP-01 REQ for a channel, built from a [`ChannelFilter`] via +/// [`build_req_filters`]. +/// +/// On first subscribe (`since` is `None`) adds `since=now` to avoid replaying +/// history. On reconnect (`since` is `Some`) subtracts [`SINCE_SKEW_SECS`]. +/// +/// Returns `true` if the REQ was successfully written to the WebSocket. +async fn send_subscribe( + ws: &mut WsStream, + _state: &BgState, + channel_id: Uuid, + agent_pubkey_hex: &str, + since: Option, + filter: &ChannelFilter, +) -> bool { + let sub_id = channel_sub_id(channel_id); + // since — on first subscribe use current time to skip history; on reconnect // subtract skew buffer to catch events missed during the disconnect window. let since_ts = match since { @@ -3179,9 +3217,15 @@ async fn send_subscribe( .unwrap_or_default() .as_secs(), }; - req_filter.insert("since".into(), json!(since_ts)); - let req = json!(["REQ", sub_id, Value::Object(req_filter)]); + let mut req = vec![json!("REQ"), json!(sub_id)]; + req.extend(build_req_filters( + channel_id, + agent_pubkey_hex, + since_ts, + filter, + )); + let req = Value::Array(req); match serde_json::to_string(&req) { Ok(text) => { @@ -4343,9 +4387,66 @@ mod tests { ChannelFilter { kinds: Some(vec![9]), require_mention: false, + thread_roots: Vec::new(), } } + #[test] + fn build_req_filters_base_clause_only() { + let channel_id = Uuid::new_v4(); + let filter = ChannelFilter { + kinds: Some(vec![9]), + require_mention: true, + thread_roots: Vec::new(), + }; + let clauses = build_req_filters(channel_id, "agent-pk", 1_000, &filter); + assert_eq!(clauses.len(), 1); + assert_eq!(clauses[0]["#h"], json!([channel_id.to_string()])); + assert_eq!(clauses[0]["#p"], json!(["agent-pk"])); + assert_eq!(clauses[0]["since"], json!(1_000)); + assert!(clauses[0].get("#e").is_none()); + } + + #[test] + fn build_req_filters_ignores_roots_without_mention_gate() { + // require_mention=false already delivers the whole channel — a second + // clause would be redundant relay load. + let filter = ChannelFilter { + kinds: None, + require_mention: false, + thread_roots: vec!["aa".repeat(32)], + }; + let clauses = build_req_filters(Uuid::new_v4(), "agent-pk", 1_000, &filter); + assert_eq!(clauses.len(), 1); + assert!(clauses[0].get("#e").is_none()); + assert!(clauses[0].get("#p").is_none()); + } + + #[test] + fn build_req_filters_adds_followed_threads_clause() { + let channel_id = Uuid::new_v4(); + let roots = vec!["aa".repeat(32), "bb".repeat(32)]; + let filter = ChannelFilter { + kinds: Some(vec![9, 45003]), + require_mention: true, + thread_roots: roots.clone(), + }; + let clauses = build_req_filters(channel_id, "agent-pk", 2_000, &filter); + assert_eq!(clauses.len(), 2); + + // Base clause: mention-gated, no #e. + assert_eq!(clauses[0]["#p"], json!(["agent-pk"])); + assert!(clauses[0].get("#e").is_none()); + + // Followed-threads clause: same channel + kinds + since scope, + // #e-limited, and NOT #p-gated. + assert_eq!(clauses[1]["#h"], json!([channel_id.to_string()])); + assert_eq!(clauses[1]["kinds"], json!([9, 45003])); + assert_eq!(clauses[1]["#e"], json!(roots)); + assert_eq!(clauses[1]["since"], json!(2_000)); + assert!(clauses[1].get("#p").is_none()); + } + fn seed_test_subscription(state: &mut BgState, channel_id: Uuid) { apply_command_to_state( state, @@ -4813,6 +4914,7 @@ mod tests { let filter = ChannelFilter { kinds: Some(vec![9]), require_mention: true, + thread_roots: Vec::new(), }; apply_command_to_state( @@ -4904,6 +5006,7 @@ mod tests { filter: ChannelFilter { kinds: Some(vec![9]), require_mention: false, + thread_roots: Vec::new(), }, replay_since: Some(1_000), }, @@ -6018,6 +6121,7 @@ mod tests { filter: ChannelFilter { kinds: Some(vec![9]), require_mention: false, + thread_roots: Vec::new(), }, replay_since: Some(1_000), }, @@ -6109,6 +6213,7 @@ mod tests { filter: ChannelFilter { kinds: Some(vec![9]), require_mention: false, + thread_roots: Vec::new(), }, replay_since: None, }, @@ -6148,6 +6253,7 @@ mod tests { filter: ChannelFilter { kinds: Some(vec![9]), require_mention: false, + thread_roots: Vec::new(), }, replay_since: None, }, @@ -6183,6 +6289,7 @@ mod tests { filter: ChannelFilter { kinds: Some(vec![9]), require_mention: false, + thread_roots: Vec::new(), }, replay_since: None, }, diff --git a/crates/buzz-acp/src/setup_mode.rs b/crates/buzz-acp/src/setup_mode.rs index dda7857260..e09df2c7b1 100644 --- a/crates/buzz-acp/src/setup_mode.rs +++ b/crates/buzz-acp/src/setup_mode.rs @@ -441,6 +441,9 @@ pub(crate) async fn run_setup_listener(config: Config, payload: SetupPayload) -> buzz_event.channel_id, &rules, &pubkey_hex, + // Setup mode has no dispatch loop, so there is no thread + // participation to follow — mention gating stays strict. + None, ) .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..feed1ff93d --- /dev/null +++ b/crates/buzz-acp/src/thread_follow.rs @@ -0,0 +1,320 @@ +//! Thread-participation tracking for mention-gated subscriptions. +//! +//! In [`SubscribeMode::Mentions`](crate::config::SubscribeMode) the harness +//! only receives events that `#p`-tag the agent. That gate is correct for +//! channel noise, but it also drops untagged replies in threads the agent is +//! actively part of — a human answering "yes, go ahead" in the agent's own +//! thread is never delivered, and the agent goes deaf mid-conversation +//! (issue #2270). +//! +//! [`ThreadFollowState`] records the NIP-10 root ids of threads the agent has +//! been dispatched into, per channel. Roots expire after a sliding TTL of +//! inactivity and each channel is capped at [`MAX_FOLLOWED_ROOTS_PER_CHANNEL`] +//! (stalest evicted first), so both the local set and the `#e` REQ filter +//! clause built from it stay bounded. +//! +//! The set feeds two consumers: +//! - [`relay`](crate::relay): a second REQ filter clause (`#h` + `#e`) so the +//! relay delivers untagged replies for followed threads only — the mention +//! gate stays intact for everything else. +//! - [`filter::match_event`](crate::filter::match_event): local admission for +//! delivered events whose thread root is followed. +//! +//! All state is in-memory, matching the rest of the harness: after a restart +//! the agent follows threads it is re-mentioned in, which is the pre-existing +//! recovery semantic for every other subscription decision. + +use std::collections::{HashMap, HashSet}; +use std::time::{Duration, Instant}; + +use tracing::debug; +use uuid::Uuid; + +/// Upper bound on followed thread roots per channel. +/// +/// Bounds both memory and the `#e` array length in the REQ filter clause +/// (relays cap filter sizes). When the cap is hit, the least-recently-active +/// root is evicted — the agent simply needs a fresh mention to rejoin that +/// thread, which is the pre-fix behavior for all threads. +pub const MAX_FOLLOWED_ROOTS_PER_CHANNEL: usize = 64; + +/// Tracks which thread roots the agent participates in, per channel. +/// +/// A root is recorded when a batch is dispatched to the agent (the agent is +/// about to speak in that thread) and refreshed whenever a followed-thread +/// event is admitted, giving active conversations a sliding TTL. +#[derive(Debug)] +pub struct ThreadFollowState { + /// When `false` every method is a no-op and [`live_roots`](Self::live_roots) + /// is always empty — call sites never need their own gating. + enabled: bool, + ttl: Duration, + /// channel → (thread root event id, hex) → last-activity instant. + followed: HashMap>, + /// Channels whose live root *set* changed since the last + /// [`take_dirty`](Self::take_dirty) — i.e. the `#e` REQ clause is stale. + dirty: HashSet, + /// Last periodic compaction, throttled to [`COMPACT_INTERVAL`]. + last_compact: Instant, +} + +/// Minimum interval between periodic compactions — the maintenance tick fires +/// far more often than expiry resolution requires. +const COMPACT_INTERVAL: Duration = Duration::from_secs(30); + +impl ThreadFollowState { + /// Create a tracker whose roots expire after `ttl_secs` of inactivity. + /// A disabled tracker never records or reports anything. + pub fn new(ttl_secs: u64, enabled: bool) -> Self { + Self { + enabled, + ttl: Duration::from_secs(ttl_secs), + followed: HashMap::new(), + dirty: HashSet::new(), + last_compact: Instant::now(), + } + } + + /// Whether thread-following is active. + pub fn enabled(&self) -> bool { + self.enabled + } + + /// Record participation in `root_id` on `channel_id`, refreshing its TTL. + /// + /// Marks the channel dirty when the root is new (the REQ clause must be + /// widened). Evicts the least-recently-active root when the per-channel + /// cap is exceeded. + pub fn record(&mut self, channel_id: Uuid, root_id: &str) { + self.record_at(channel_id, root_id, Instant::now()); + } + + fn record_at(&mut self, channel_id: Uuid, root_id: &str, now: Instant) { + if !self.enabled { + return; + } + let roots = self.followed.entry(channel_id).or_default(); + let is_new = !roots.contains_key(root_id); + roots.insert(root_id.to_string(), now); + + if roots.len() > MAX_FOLLOWED_ROOTS_PER_CHANNEL { + // Evict the stalest root. `max_by_key` on elapsed == oldest instant. + if let Some(stalest) = roots + .iter() + .min_by_key(|(_, seen)| **seen) + .map(|(root, _)| root.clone()) + { + roots.remove(&stalest); + debug!( + channel = %channel_id, + evicted_root = %stalest, + cap = MAX_FOLLOWED_ROOTS_PER_CHANNEL, + "followed-thread cap reached — evicted least-recently-active root" + ); + self.dirty.insert(channel_id); + } + } + + if is_new { + debug!(channel = %channel_id, root = %root_id, "following thread"); + self.dirty.insert(channel_id); + } + } + + /// Refresh the TTL of `root_id` if it is currently followed. + /// + /// Called when a followed-thread event is admitted, so active + /// conversations keep sliding forward instead of expiring mid-exchange. + pub fn touch(&mut self, channel_id: Uuid, root_id: &str) { + self.touch_at(channel_id, root_id, Instant::now()); + } + + fn touch_at(&mut self, channel_id: Uuid, root_id: &str, now: Instant) { + if let Some(seen) = self + .followed + .get_mut(&channel_id) + .and_then(|roots| roots.get_mut(root_id)) + { + *seen = now; + } + } + + /// Live (unexpired) root ids for `channel_id`, for local admission checks. + pub fn live_roots(&self, channel_id: Uuid) -> HashSet { + self.live_roots_at(channel_id, Instant::now()) + } + + fn live_roots_at(&self, channel_id: Uuid, now: Instant) -> HashSet { + self.followed + .get(&channel_id) + .map(|roots| { + roots + .iter() + .filter(|(_, seen)| now.duration_since(**seen) < self.ttl) + .map(|(root, _)| root.clone()) + .collect() + }) + .unwrap_or_default() + } + + /// Live root ids for `channel_id` in deterministic order, for building the + /// `#e` REQ filter clause. + pub fn roots_for_filter(&self, channel_id: Uuid) -> Vec { + let mut roots: Vec = self.live_roots(channel_id).into_iter().collect(); + roots.sort_unstable(); + roots + } + + /// Drop expired roots, throttled to [`COMPACT_INTERVAL`]. Channels whose + /// set shrank are marked dirty so their REQ clause can be narrowed. Safe + /// to call on every maintenance tick. + pub fn compact_if_due(&mut self) { + let now = Instant::now(); + if !self.enabled || now.duration_since(self.last_compact) < COMPACT_INTERVAL { + return; + } + self.last_compact = now; + self.compact_at(now); + } + + fn compact_at(&mut self, now: Instant) { + let ttl = self.ttl; + let dirty = &mut self.dirty; + self.followed.retain(|channel_id, roots| { + let before = roots.len(); + roots.retain(|_, seen| now.duration_since(*seen) < ttl); + if roots.len() != before { + dirty.insert(*channel_id); + } + !roots.is_empty() + }); + } + + /// Forget all state for a channel (e.g. after membership removal). + /// + /// Does not mark the channel dirty — the caller is tearing the + /// subscription down entirely, so there is no REQ left to update. + pub fn clear_channel(&mut self, channel_id: Uuid) { + self.followed.remove(&channel_id); + self.dirty.remove(&channel_id); + } + + /// Drain the set of channels whose followed-root set changed since the + /// last call. Each returned channel needs its subscription re-issued with + /// a fresh `#e` clause. + pub fn take_dirty(&mut self) -> Vec { + self.dirty.drain().collect() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn ch(n: u128) -> Uuid { + Uuid::from_u128(n) + } + + #[test] + fn record_marks_new_roots_dirty_once() { + let mut state = ThreadFollowState::new(60, true); + state.record(ch(1), "root-a"); + assert_eq!(state.take_dirty(), vec![ch(1)]); + + // Re-recording the same root refreshes TTL but is not a set change. + state.record(ch(1), "root-a"); + assert!(state.take_dirty().is_empty()); + + assert!(state.live_roots(ch(1)).contains("root-a")); + } + + #[test] + fn roots_expire_after_ttl_and_mark_dirty() { + let mut state = ThreadFollowState::new(60, true); + let start = Instant::now(); + state.record_at(ch(1), "root-a", start); + state.take_dirty(); + + let later = start + Duration::from_secs(61); + assert!(state.live_roots_at(ch(1), later).is_empty()); + + state.compact_at(later); + assert_eq!(state.take_dirty(), vec![ch(1)]); + assert!(state.roots_for_filter(ch(1)).is_empty()); + } + + #[test] + fn touch_slides_expiry_forward() { + let mut state = ThreadFollowState::new(60, true); + let start = Instant::now(); + state.record_at(ch(1), "root-a", start); + + // Refresh at t+45; at t+75 the root is still live (75 - 45 < 60). + state.touch_at(ch(1), "root-a", start + Duration::from_secs(45)); + let at_75 = start + Duration::from_secs(75); + assert!(state.live_roots_at(ch(1), at_75).contains("root-a")); + + // Without further activity it expires at t+45+60. + let at_106 = start + Duration::from_secs(106); + assert!(state.live_roots_at(ch(1), at_106).is_empty()); + } + + #[test] + fn touch_ignores_unknown_roots() { + let mut state = ThreadFollowState::new(60, true); + state.touch(ch(1), "never-recorded"); + assert!(state.live_roots(ch(1)).is_empty()); + assert!(state.take_dirty().is_empty()); + } + + #[test] + fn cap_evicts_least_recently_active_root() { + let mut state = ThreadFollowState::new(3600, true); + let start = Instant::now(); + for i in 0..MAX_FOLLOWED_ROOTS_PER_CHANNEL { + state.record_at( + ch(1), + &format!("root-{i:03}"), + start + Duration::from_secs(i as u64), + ); + } + state.take_dirty(); + + // One over the cap evicts root-000 (the stalest), keeps the newcomer. + state.record_at( + ch(1), + "root-new", + start + Duration::from_secs(MAX_FOLLOWED_ROOTS_PER_CHANNEL as u64), + ); + let live = state.live_roots_at(ch(1), start + Duration::from_secs(100)); + assert_eq!(live.len(), MAX_FOLLOWED_ROOTS_PER_CHANNEL); + assert!(!live.contains("root-000")); + assert!(live.contains("root-new")); + assert_eq!(state.take_dirty(), vec![ch(1)]); + } + + #[test] + fn channels_are_independent() { + let mut state = ThreadFollowState::new(60, true); + state.record(ch(1), "root-a"); + state.record(ch(2), "root-b"); + + assert!(state.live_roots(ch(1)).contains("root-a")); + assert!(!state.live_roots(ch(1)).contains("root-b")); + + state.clear_channel(ch(1)); + assert!(state.live_roots(ch(1)).is_empty()); + assert!(state.live_roots(ch(2)).contains("root-b")); + // clear_channel drops pending dirtiness for the removed channel. + assert_eq!(state.take_dirty(), vec![ch(2)]); + } + + #[test] + fn roots_for_filter_is_sorted() { + let mut state = ThreadFollowState::new(60, true); + state.record(ch(1), "bbb"); + state.record(ch(1), "aaa"); + state.record(ch(1), "ccc"); + assert_eq!(state.roots_for_filter(ch(1)), vec!["aaa", "bbb", "ccc"]); + } +} From a72b450a4fccf5da17026cb4e3f8aaf23746b090 Mon Sep 17 00:00:00 2001 From: Broc Oppler Date: Wed, 22 Jul 2026 10:30:23 -0700 Subject: [PATCH 2/3] fix(acp): restrict followed-thread admissions to human authors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review of #2375 (BradGroux) identified an agent-to-agent loop the initial thread-following change permitted: ignore_self blocks self-loops and the author gate blocks strangers, but same-owner sibling agents pass the gate by design — so two siblings participating in one thread could wake each other with unmentioned replies indefinitely, each turn refreshing the follow TTL. Followed-thread admissions are now author-aware: - match_event reports the admission basis (admitted_via_follow) so the caller can distinguish follow-bypass admissions from mention matches. - A follow-only admission from an agent author is suppressed under the new --follow-thread-authors policy (BUZZ_ACP_FOLLOW_THREAD_AUTHORS): humans (default) admits human authors only; all opts into agent authors and accepts the auto-continuation risk. Explicit mentions are never affected, and suppression happens before the TTL refresh so agent chatter cannot keep a thread alive. - Agenthood uses the existing NIP-OA heuristic (pool::profile_event_is_agent, now pub(crate)) via a cached kind:0 lookup that runs only on the rare follow-admitted path; unknown identities classify as human (fail open, matching turn_is_human_facing). The security gate remains author_allowed. Regression coverage per review: human unmentioned reply admitted; agent-authored unmentioned reply dropped (covers siblings and external agents under respond-to anyone); mentioned agent-authored replies unaffected; all-policy opt-in; and the two-agent auto-continuation scenario cannot start from either side. Co-Authored-By: Claude Fable 5 Signed-off-by: Broc Oppler --- crates/buzz-acp/src/config.rs | 26 +++++ crates/buzz-acp/src/filter.rs | 41 +++++++- crates/buzz-acp/src/lib.rs | 183 +++++++++++++++++++++++++++++++++- crates/buzz-acp/src/pool.rs | 2 +- 4 files changed, 244 insertions(+), 8 deletions(-) diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index ee1f39ff0b..77650d31db 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -54,6 +54,16 @@ pub enum SubscribeMode { Config, } +/// Author policy for followed-thread admissions (#2270 loop guard). +#[derive(Debug, Clone, Copy, PartialEq, Eq, clap::ValueEnum)] +pub enum FollowThreadAuthors { + /// Admit unmentioned followed-thread replies from humans only (default). + Humans, + /// Admit unmentioned followed-thread replies from any author, including + /// agents. Opting in accepts the risk of agent↔agent auto-continuation. + All, +} + #[derive(Debug, Clone, Copy, clap::ValueEnum)] pub enum DedupMode { Drop, @@ -344,6 +354,18 @@ pub struct CliArgs { #[arg(long, env = "BUZZ_ACP_NO_FOLLOW_THREADS")] pub no_follow_threads: bool, + /// Whose unmentioned replies a followed thread admits. `humans` (default) + /// admits human authors only — agent-authored replies still require an + /// explicit @mention, so two agents sharing a thread can never + /// auto-continue each other. `all` admits any author. + #[arg( + long, + env = "BUZZ_ACP_FOLLOW_THREAD_AUTHORS", + default_value = "humans", + value_enum + )] + pub follow_thread_authors: FollowThreadAuthors, + /// Sliding inactivity TTL, in seconds, for followed threads. A thread the /// agent participated in stops being followed after this long without /// activity; a fresh @mention re-joins it. @@ -536,6 +558,8 @@ pub struct Config { pub follow_threads: bool, /// Sliding inactivity TTL for followed thread roots, in seconds. pub thread_follow_ttl_secs: u64, + /// Author policy for followed-thread admissions (loop guard). + pub follow_thread_authors: FollowThreadAuthors, pub config_path: PathBuf, pub context_message_limit: u32, /// Maximum turns per session before proactive rotation. 0 = disabled. @@ -1014,6 +1038,7 @@ impl Config { no_mention_filter: args.no_mention_filter, follow_threads: !args.no_follow_threads, thread_follow_ttl_secs: args.thread_follow_ttl_secs, + follow_thread_authors: args.follow_thread_authors, config_path: args.config, context_message_limit: args.context_message_limit, max_turns_per_session: args.max_turns_per_session, @@ -1391,6 +1416,7 @@ mod tests { no_mention_filter: false, follow_threads: true, thread_follow_ttl_secs: 86_400, + follow_thread_authors: FollowThreadAuthors::Humans, 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 1b2d813480..1dded340ef 100644 --- a/crates/buzz-acp/src/filter.rs +++ b/crates/buzz-acp/src/filter.rs @@ -154,6 +154,11 @@ pub struct MatchedRule { pub rule_index: usize, /// Prompt tag to use (rule's `prompt_tag` or its `name`). pub prompt_tag: String, + /// True when the winning rule required a mention that was absent and the + /// event was admitted only because its thread root is followed (#2270). + /// The caller applies the followed-thread author policy to exactly these + /// admissions — mention-based matches are never subject to it. + pub admitted_via_follow: bool, } /// Maximum expression length accepted by `evaluate_filter`. @@ -387,6 +392,8 @@ pub async fn match_event( }); for (index, rule) in rules.iter().enumerate() { + let mut admitted_via_follow = false; + // 1. Channel scope check. if !rule.channels.matches(&channel_id) { continue; @@ -406,8 +413,11 @@ pub async fn match_event( s.first().map(|k| k.as_str()) == Some("p") && s.get(1).map(|v| v.as_str()) == Some(agent_pubkey_hex) }); - if !mentioned && followed_thread_root.is_none() { - continue; + if !mentioned { + if followed_thread_root.is_none() { + continue; + } + admitted_via_follow = true; } } @@ -466,6 +476,7 @@ pub async fn match_event( return Some(MatchedRule { rule_index: index, prompt_tag, + admitted_via_follow, }); } @@ -527,10 +538,32 @@ mod tests { .await .is_none()); - // With the thread followed, the same unmentioned reply is admitted. + // With the thread followed, the same unmentioned reply is admitted — + // and flagged as a follow admission so the author policy can act. + let followed: HashSet = [root].into_iter().collect(); + let matched = match_event(&event, channel_id, &rules, "agent-pk", Some(&followed)).await; + assert!(matched.as_ref().is_some_and(|m| m.admitted_via_follow)); + } + + #[tokio::test] + async fn test_match_event_mention_admission_is_not_flagged_as_follow() { + let channel_id = any_channel(); + let rules = vec![make_rule( + "mentions", + ChannelScope::All("all".into()), + vec![9], + true, + None, + None, + )]; + // Explicitly mentioned event in a followed thread: admitted via the + // mention, so the follow flag must stay false (the author policy + // never applies to explicit mentions). + let root = "e".repeat(64); let followed: HashSet = [root].into_iter().collect(); + let event = make_event_with_p_tag(9, "hey @agent", "agent-pk"); let matched = match_event(&event, channel_id, &rules, "agent-pk", Some(&followed)).await; - assert!(matched.is_some()); + assert!(matched.as_ref().is_some_and(|m| !m.admitted_via_follow)); } #[tokio::test] diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 1a328f2aeb..6bb13a99c5 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -154,6 +154,10 @@ struct OwnerCache { pubkey: Option, /// author_hex → is_sibling (true = same owner, false = not) siblings: std::sync::Mutex>, + /// author_hex → is_agent (NIP-OA auth tag present on kind:0). + /// Used by the followed-thread author policy (#2270 loop guard), not by + /// the security gate — same heuristic weight as prompt reply anchoring. + agent_flags: std::sync::Mutex>, } impl OwnerCache { @@ -161,6 +165,25 @@ impl OwnerCache { Self { pubkey: initial, siblings: std::sync::Mutex::new(HashMap::new()), + agent_flags: std::sync::Mutex::new(HashMap::new()), + } + } + + /// Check the agent-flag cache. `None` = never looked up. + fn known_agent_flag(&self, author: &str) -> Option { + self.agent_flags + .lock() + .ok() + .and_then(|map| map.get(author).copied()) + } + + /// Cache an agent-flag lookup result (capped like the sibling cache). + fn cache_agent_flag(&self, author: String, is_agent: bool) { + if let Ok(mut map) = self.agent_flags.lock() { + if map.len() >= 256 { + map.clear(); + } + map.insert(author, is_agent); } } @@ -216,6 +239,64 @@ async fn is_owner_or_sibling( is_sibling } +/// Is this author an agent (NIP-OA `auth` tag on their kind:0 profile)? +/// +/// Used by the followed-thread author policy: an *unmentioned* event admitted +/// only because its thread is followed must not fire a turn when its author +/// is another agent, or two agents sharing a thread auto-continue each other +/// indefinitely (the loop #2270 requires stay impossible). Presence/shape +/// heuristic per [`pool::profile_event_is_agent`] — same weight as prompt +/// reply anchoring; the security gate remains [`author_allowed`]. Unknown or +/// unfetchable identities classify as human (fail open for visibility, +/// matching `turn_is_human_facing`); explicit mentions are never affected. +async fn author_is_agent( + author: &str, + owner_cache: &OwnerCache, + rest_client: &relay::RestClient, +) -> bool { + if let Some(cached) = owner_cache.known_agent_flag(author) { + return cached; + } + + let author_pk = match nostr::PublicKey::from_hex(author) { + Ok(pk) => pk, + Err(_) => return false, + }; + let filter = nostr::Filter::new() + .kind(nostr::Kind::Metadata) + .author(author_pk) + .limit(1); + + let is_agent = + match tokio::time::timeout(Duration::from_millis(2000), rest_client.query(&[filter])).await + { + Ok(Ok(resp)) => resp + .as_array() + .and_then(|events| events.first()) + .is_some_and(pool::profile_event_is_agent), + // Timeout/error — treat as human (fail open) but do NOT cache, so a + // transient relay hiccup can't pin an agent as human for the session. + _ => return false, + }; + + owner_cache.cache_agent_flag(author.to_string(), is_agent); + is_agent +} + +/// Whether a followed-thread admission should be suppressed for this author. +/// +/// Pure decision core of the #2270 loop guard, kept separate for testability: +/// only events admitted *via the follow bypass* (unmentioned, followed root) +/// are ever suppressed, and only under the `humans` policy when the author is +/// an agent. Mentioned events and normally-matched events pass untouched. +fn suppress_follow_admission( + admitted_via_follow: bool, + policy: config::FollowThreadAuthors, + author_is_agent: bool, +) -> bool { + admitted_via_follow && matches!(policy, config::FollowThreadAuthors::Humans) && author_is_agent +} + /// Inbound author gate decision: does this author's event fire a turn? /// /// Coarse security policy applied before subscription rules. Both `OwnerOnly` @@ -2163,13 +2244,42 @@ async fn tokio_main() -> Result<()> { .enabled() .then(|| thread_follow.live_roots(buzz_event.channel_id)); let matched = filter::match_event(&buzz_event.event, buzz_event.channel_id, &rules, &pubkey_hex, followed_roots.as_ref()).await; - let prompt_tag = match matched { - Some(m) => m.prompt_tag, + let (prompt_tag, admitted_via_follow) = match matched { + Some(m) => (m.prompt_tag, m.admitted_via_follow), None => { tracing::debug!(channel_id = %buzz_event.channel_id, kind = buzz_event.event.kind.as_u16(), "event matched no rule — dropping"); continue; } }; + // #2270 loop guard: an admission that exists only + // because the thread is followed must not fire on + // agent authors (policy `humans`, the default) — + // otherwise two agents sharing a thread would + // auto-continue each other indefinitely. Explicit + // mentions never take this path, and the profile + // lookup runs only on follow-admitted events. + if admitted_via_follow { + let author = buzz_event.event.pubkey.to_hex(); + let is_agent_author = match config.follow_thread_authors { + config::FollowThreadAuthors::Humans => { + author_is_agent(&author, &owner_cache, &ctx.rest_client) + .await + } + config::FollowThreadAuthors::All => false, + }; + if suppress_follow_admission( + admitted_via_follow, + config.follow_thread_authors, + is_agent_author, + ) { + tracing::debug!( + channel_id = %buzz_event.channel_id, + author = %author, + "followed-thread admission suppressed for agent author" + ); + continue; + } + } // An admitted event in a followed thread is live // conversation — slide that thread's TTL forward. if thread_follow.enabled() { @@ -2562,7 +2672,7 @@ async fn tokio_main() -> Result<()> { None, ); for (channel_id, thread_tags) in - dispatch_pending(&mut pool, &mut queue, &ctx) + dispatch_pending(&mut pool, &mut queue, &ctx, &mut thread_follow) { typing_channels.insert(channel_id, thread_tags); } @@ -4341,6 +4451,71 @@ mod owner_cache_tests { } } +#[cfg(test)] +mod follow_author_policy_tests { + use super::*; + use config::FollowThreadAuthors; + + #[test] + fn human_unmentioned_reply_in_followed_root_is_admitted() { + assert!(!suppress_follow_admission( + true, + FollowThreadAuthors::Humans, + false + )); + } + + #[test] + fn agent_unmentioned_reply_in_followed_root_is_dropped() { + // Covers both the same-owner sibling case (passes the author gate) + // and the external-agent-under-RespondTo::Anyone case — the policy + // classifies by agenthood, not by gate mode. + assert!(suppress_follow_admission( + true, + FollowThreadAuthors::Humans, + true + )); + } + + #[test] + fn mentioned_admissions_are_never_suppressed() { + // A mention-based match has admitted_via_follow == false, so the + // policy cannot touch it regardless of author kind or policy value. + assert!(!suppress_follow_admission( + false, + FollowThreadAuthors::Humans, + true + )); + assert!(!suppress_follow_admission( + false, + FollowThreadAuthors::All, + true + )); + } + + #[test] + fn all_policy_opts_into_agent_authors() { + assert!(!suppress_follow_admission( + true, + FollowThreadAuthors::All, + true + )); + } + + #[test] + fn two_agents_sharing_a_followed_root_cannot_auto_continue() { + // Simulate the #2270 loop hazard at the decision layer: agents A and + // B both follow root T. B posts an unmentioned reply; A's harness + // sees a follow-only admission from an agent author — suppressed. If + // A ever replied anyway, B's harness would face the identical + // decision — also suppressed. Under the default policy the cycle + // cannot begin from either side. + let a_fires = !suppress_follow_admission(true, FollowThreadAuthors::Humans, true); + let b_fires = !suppress_follow_admission(true, FollowThreadAuthors::Humans, true); + assert!(!a_fires && !b_fires); + } +} + #[cfg(test)] mod author_gate_tests { use super::*; @@ -4707,6 +4882,7 @@ mod build_mcp_servers_tests { no_mention_filter: false, follow_threads: true, thread_follow_ttl_secs: 86_400, + follow_thread_authors: config::FollowThreadAuthors::Humans, config_path: std::path::PathBuf::from("./buzz-acp.toml"), context_message_limit: 12, max_turns_per_session: 0, @@ -4875,6 +5051,7 @@ mod error_outcome_emission_tests { no_mention_filter: false, follow_threads: true, thread_follow_ttl_secs: 86_400, + follow_thread_authors: config::FollowThreadAuthors::Humans, 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/pool.rs b/crates/buzz-acp/src/pool.rs index fb08096792..c438037a7b 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -2539,7 +2539,7 @@ fn collect_prompt_pubkeys( /// cheap routing heuristic for reply anchoring, not a verified security gate /// (the signing path in `lib.rs::check_sibling_via_profile` does full /// verification where it matters). -fn profile_event_is_agent(ev: &serde_json::Value) -> bool { +pub(crate) fn profile_event_is_agent(ev: &serde_json::Value) -> bool { ev.get("tags") .and_then(|t| t.as_array()) .is_some_and(|tags| { From b9700ca4d9b454b8c92008a95b4c750e06f2a8bd Mon Sep 17 00:00:00 2001 From: Broc Oppler Date: Wed, 22 Jul 2026 20:03:47 -0700 Subject: [PATCH 3/3] test(acp): live-relay e2e for thread-following admission and suppression Drives the real buzz-acp binary with a minimal stub ACP agent (stub-acp-agent, NDJSON JSON-RPC over stdio) against a live relay and asserts turn dispatch for the four cases from the #2375 review: explicit mention fires a turn and follows the thread; an untagged human reply in the followed thread fires a second turn; an untagged agent-authored reply (NIP-OA-tagged kind:0, respond-to=anyone so the followed-thread author policy is the only gate) is suppressed with the suppression log line asserted; an untagged top-level message stays gated. The stub logs each session/prompt to a file and never contacts the relay: participation is recorded by the harness at dispatch, so turn counts alone prove admission/suppression. The agent reply path is out-of-band by design (#2459 tracks its observability). #[ignore]-gated like the rest of the e2e suite; requires a running relay and a built buzz-acp (BUZZ_ACP_BIN override supported). Co-Authored-By: Claude Fable 5 Signed-off-by: Broc Oppler --- crates/buzz-test-client/Cargo.toml | 4 + .../tests/bin/stub_acp_agent.rs | 80 +++++ .../tests/e2e_acp_thread_follow.rs | 274 ++++++++++++++++++ 3 files changed, 358 insertions(+) create mode 100644 crates/buzz-test-client/tests/bin/stub_acp_agent.rs create mode 100644 crates/buzz-test-client/tests/e2e_acp_thread_follow.rs diff --git a/crates/buzz-test-client/Cargo.toml b/crates/buzz-test-client/Cargo.toml index 40a08f3d19..9064d46abd 100644 --- a/crates/buzz-test-client/Cargo.toml +++ b/crates/buzz-test-client/Cargo.toml @@ -41,3 +41,7 @@ buzz-sdk = { workspace = true } [[bin]] name = "buzz-test-cli" path = "src/main.rs" + +[[bin]] +name = "stub-acp-agent" +path = "tests/bin/stub_acp_agent.rs" diff --git a/crates/buzz-test-client/tests/bin/stub_acp_agent.rs b/crates/buzz-test-client/tests/bin/stub_acp_agent.rs new file mode 100644 index 0000000000..876418626d --- /dev/null +++ b/crates/buzz-test-client/tests/bin/stub_acp_agent.rs @@ -0,0 +1,80 @@ +//! Minimal ACP-speaking stub agent for harness integration tests. +//! +//! Speaks newline-delimited JSON-RPC 2.0 over stdio — just enough of the +//! Agent Client Protocol for `buzz-acp` to initialize, create a session, and +//! run prompt turns. Each `session/prompt` appends one line to the file named +//! by `STUB_TURN_LOG`, then immediately ends the turn — the test counts lines +//! to observe exactly how many turns the harness dispatched. The stub never +//! contacts the relay: thread participation is recorded by the harness at +//! dispatch time, so turn counts alone prove admission/suppression behavior. + +use std::io::{BufRead, Write}; + +fn respond( + stdout: &mut std::io::StdoutLock<'_>, + id: &serde_json::Value, + result: serde_json::Value, +) { + let msg = serde_json::json!({ "jsonrpc": "2.0", "id": id, "result": result }); + writeln!(stdout, "{msg}").expect("write response"); + stdout.flush().expect("flush stdout"); +} + +fn main() { + let stdin = std::io::stdin(); + let stdout = std::io::stdout(); + let mut out = stdout.lock(); + let turn_log = std::env::var("STUB_TURN_LOG").ok(); + let mut turn: u64 = 0; + + for line in stdin.lock().lines() { + let line = match line { + Ok(l) if !l.trim().is_empty() => l, + Ok(_) => continue, + Err(_) => break, + }; + let msg: serde_json::Value = match serde_json::from_str(&line) { + Ok(v) => v, + Err(_) => continue, + }; + let id = msg.get("id").cloned(); + let method = msg.get("method").and_then(|m| m.as_str()).unwrap_or(""); + + let Some(id) = id else { + continue; // notification — nothing to answer + }; + + match method { + "initialize" => respond( + &mut out, + &id, + serde_json::json!({ "protocolVersion": 2, "agentCapabilities": {} }), + ), + "session/new" => respond( + &mut out, + &id, + serde_json::json!({ "sessionId": "stub-session-1" }), + ), + "session/prompt" => { + turn += 1; + if let Some(path) = &turn_log { + let mut f = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(path) + .expect("open turn log"); + writeln!(f, "turn {turn}").expect("append turn log"); + } + respond( + &mut out, + &id, + serde_json::json!({ "stopReason": "end_turn" }), + ); + } + // Anything else the harness asks for (config options, custom + // extensions): empty success keeps it moving; it tolerates both + // empty results and method-not-found for optional calls. + _ => respond(&mut out, &id, serde_json::json!({})), + } + } +} diff --git a/crates/buzz-test-client/tests/e2e_acp_thread_follow.rs b/crates/buzz-test-client/tests/e2e_acp_thread_follow.rs new file mode 100644 index 0000000000..effdf04168 --- /dev/null +++ b/crates/buzz-test-client/tests/e2e_acp_thread_follow.rs @@ -0,0 +1,274 @@ +//! End-to-end test for buzz-acp thread-following (#2270) against a live relay. +//! +//! Drives the real `buzz-acp` binary with a stub ACP agent (`stub-acp-agent`, +//! built from this crate) and asserts turn dispatch across the four cases from +//! the #2375 review: +//! +//! 1. explicit @mention → one turn (thread becomes followed); +//! 2. untagged human reply in the followed thread → one more turn; +//! 3. untagged agent-authored reply in the followed thread → no turn +//! (followed-thread author policy, default `humans`); +//! 4. untagged top-level message → no turn (mention gate intact). +//! +//! Turn observation: the stub appends a line to `STUB_TURN_LOG` per +//! `session/prompt`. Participation is recorded by the harness at dispatch, so +//! turn counts alone prove admission/suppression — the stub never contacts the +//! relay (the agent's reply path is out-of-band by design; see #2459). +//! +//! Prerequisites: +//! - relay + Postgres running (`just setup`, `just relay`); `RELAY_URL` +//! defaults to ws://localhost:3000 +//! - `cargo build -p buzz-acp` (or set `BUZZ_ACP_BIN`) +//! +//! Run: `cargo test -p buzz-test-client --test e2e_acp_thread_follow -- --ignored` + +use std::path::PathBuf; +use std::time::{Duration, Instant}; + +use nostr::{EventBuilder, Keys, Kind, Tag}; + +fn relay_url() -> String { + std::env::var("RELAY_URL").unwrap_or_else(|_| "ws://localhost:3000".to_string()) +} + +fn relay_http_url() -> String { + relay_url() + .replace("wss://", "https://") + .replace("ws://", "http://") + .trim_end_matches('/') + .to_string() +} + +/// Locate the buzz-acp binary: `BUZZ_ACP_BIN` override, else the workspace +/// target dir (release preferred, debug fallback). +fn buzz_acp_bin() -> PathBuf { + if let Ok(path) = std::env::var("BUZZ_ACP_BIN") { + return PathBuf::from(path); + } + let mut workspace = PathBuf::from(env!("CARGO_MANIFEST_DIR")); + workspace.pop(); + workspace.pop(); + for profile in ["release", "debug"] { + let candidate = workspace.join("target").join(profile).join("buzz-acp"); + if candidate.exists() { + return candidate; + } + } + panic!("buzz-acp binary not found — run `cargo build -p buzz-acp` or set BUZZ_ACP_BIN"); +} + +/// POST a signed event to the relay's HTTP bridge. +async fn post_event(event: &nostr::Event) { + let client = reqwest::Client::new(); + let resp = client + .post(format!("{}/events", relay_http_url())) + .header("X-Pubkey", event.pubkey.to_hex()) + .header("Content-Type", "application/json") + .body(serde_json::to_string(event).expect("serialize event")) + .send() + .await + .expect("submit event"); + let status = resp.status(); + let body: serde_json::Value = resp.json().await.expect("parse event response"); + assert!( + status.is_success() && body["accepted"].as_bool().unwrap_or(false), + "event not accepted: {status} {body}" + ); +} + +/// Create a channel via a signed kind:9007 event (creator becomes a member). +async fn create_test_channel(keys: &Keys) -> uuid::Uuid { + let channel_uuid = uuid::Uuid::new_v4(); + let event = EventBuilder::new(Kind::Custom(9007), "") + .tags(vec![ + Tag::parse(["h", &channel_uuid.to_string()]).unwrap(), + Tag::parse(["name", &format!("acp-follow-e2e-{channel_uuid}")]).unwrap(), + Tag::parse(["channel_type", "stream"]).unwrap(), + Tag::parse(["visibility", "open"]).unwrap(), + ]) + .sign_with_keys(keys) + .unwrap(); + post_event(&event).await; + channel_uuid +} + +/// Add a member to a channel (kind:9030, signed by the channel owner). +async fn add_member(owner: &Keys, channel: uuid::Uuid, member: &Keys) { + let event = buzz_sdk::build_add_member(channel, &member.public_key().to_hex(), None) + .expect("build add-member") + .sign_with_keys(owner) + .expect("sign add-member"); + post_event(&event).await; +} + +/// Publish a kind:9 channel message. `mention` adds a `p` tag; `thread_root` +/// adds a NIP-10 marker `e` tag. +async fn send_channel_message( + keys: &Keys, + channel: uuid::Uuid, + content: &str, + mention: Option<&Keys>, + thread_root: Option<&str>, +) -> String { + let mut tags = vec![Tag::parse(["h", &channel.to_string()]).unwrap()]; + if let Some(m) = mention { + tags.push(Tag::parse(["p", &m.public_key().to_hex()]).unwrap()); + } + if let Some(root) = thread_root { + tags.push(Tag::parse(["e", root, "", "root"]).unwrap()); + } + let event = EventBuilder::new(Kind::Custom(9), content) + .tags(tags) + .sign_with_keys(keys) + .unwrap(); + let id = event.id.to_hex(); + post_event(&event).await; + id +} + +fn turn_count(log: &std::path::Path) -> usize { + std::fs::read_to_string(log) + .map(|s| s.lines().count()) + .unwrap_or(0) +} + +/// Deadline-poll until `turn_count` reaches `expected` (same shape as the +/// scheduler-push waits in e2e_event_reminder.rs). +async fn await_turns(log: &std::path::Path, expected: usize, deadline: Duration) { + let start = Instant::now(); + while start.elapsed() < deadline { + if turn_count(log) >= expected { + return; + } + tokio::time::sleep(Duration::from_millis(250)).await; + } + panic!( + "expected {expected} dispatched turn(s) within {deadline:?}, saw {}", + turn_count(log) + ); +} + +/// Assert `turn_count` stays at `expected` for the whole `window`. +async fn assert_no_new_turns(log: &std::path::Path, expected: usize, window: Duration) { + let start = Instant::now(); + while start.elapsed() < window { + let n = turn_count(log); + assert!( + n <= expected, + "unexpected turn dispatched: saw {n}, expected {expected}" + ); + tokio::time::sleep(Duration::from_millis(250)).await; + } +} + +/// Kills the harness on drop so a failing assertion can't leak the subprocess. +struct HarnessGuard(std::process::Child); +impl Drop for HarnessGuard { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +#[tokio::test] +#[ignore] +async fn thread_follow_admits_humans_and_suppresses_agents() { + let human = Keys::generate(); + let agent = Keys::generate(); + let other_agent = Keys::generate(); + + let channel = create_test_channel(&human).await; + add_member(&human, channel, &agent).await; + add_member(&human, channel, &other_agent).await; + + // Publish a kind:0 for the second agent carrying the NIP-OA `auth` tag + // shape (4 elements) — the marker `profile_event_is_agent` classifies on. + let profile = EventBuilder::new(Kind::Metadata, r#"{"name":"other-agent"}"#) + .tags(vec![Tag::parse([ + "auth", + &human.public_key().to_hex(), + "", + "e2e-attestation-placeholder", + ]) + .unwrap()]) + .sign_with_keys(&other_agent) + .unwrap(); + post_event(&profile).await; + + // Spawn the real harness with the stub ACP agent. + let scratch = std::env::temp_dir().join(format!("acp-follow-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&scratch).expect("create scratch dir"); + let turn_log = scratch.join("turns.log"); + let harness_log = scratch.join("harness.log"); + let log_file = std::fs::File::create(&harness_log).expect("create harness log"); + + let child = std::process::Command::new(buzz_acp_bin()) + .env("BUZZ_PRIVATE_KEY", agent.secret_key().to_secret_hex()) + .env("BUZZ_RELAY_URL", relay_url()) + .env( + "BUZZ_ACP_AGENT_COMMAND", + env!("CARGO_BIN_EXE_stub-acp-agent"), + ) + .env("BUZZ_ACP_RESPOND_TO", "anyone") + .env("BUZZ_ACP_HEARTBEAT_INTERVAL", "0") + .env("BUZZ_ACP_AGENTS", "1") + .env("STUB_TURN_LOG", &turn_log) + .env("RUST_LOG", "buzz_acp=debug") + .stdout(log_file.try_clone().expect("clone log handle")) + .stderr(log_file) + .spawn() + .expect("spawn buzz-acp"); + let _guard = HarnessGuard(child); + + // Readiness: the harness logs its channel subscription. + let subscribed = format!("subscribed to channel {channel}"); + let start = Instant::now(); + loop { + let log = std::fs::read_to_string(&harness_log).unwrap_or_default(); + if log.contains(&subscribed) { + break; + } + assert!( + start.elapsed() < Duration::from_secs(30), + "harness did not subscribe to {channel} within 30s; log:\n{log}" + ); + tokio::time::sleep(Duration::from_millis(250)).await; + } + + // 1. Explicit mention → one turn; the thread root (= mention id) becomes followed. + let mention_id = send_channel_message(&human, channel, "@agent ping", Some(&agent), None).await; + await_turns(&turn_log, 1, Duration::from_secs(30)).await; + + // 2. Untagged human reply in the followed thread → a second turn. + send_channel_message( + &human, + channel, + "untagged follow-up", + None, + Some(&mention_id), + ) + .await; + await_turns(&turn_log, 2, Duration::from_secs(30)).await; + + // 3. Untagged agent-authored reply in the followed thread → suppressed. + // respond-to=anyone admits the author, so the followed-thread author + // policy is the only gate exercised here. + send_channel_message( + &other_agent, + channel, + "untagged agent interjection", + None, + Some(&mention_id), + ) + .await; + assert_no_new_turns(&turn_log, 2, Duration::from_secs(10)).await; + let log = std::fs::read_to_string(&harness_log).unwrap_or_default(); + assert!( + log.contains("followed-thread admission suppressed for agent author"), + "expected suppression log line; harness log:\n{log}" + ); + + // 4. Untagged top-level message → no turn (mention gate intact). + send_channel_message(&human, channel, "no mention here", None, None).await; + assert_no_new_turns(&turn_log, 2, Duration::from_secs(10)).await; +}