Skip to content

feat(buzz-acp): follow threads the agent participates in (--thread-follow) #5045

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
nikomoritz wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(buzz-acp): follow threads the agent participates in (--thread-follow) #5045

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 28 additions & 3 deletions crates/buzz-acp/src/config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,

Expand Down Expand Up @@ -514,6 +521,11 @@ pub struct Config {
pub kinds_override: Option<Vec<u32>>,
pub channels_override: Option<Vec<String>>,
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.
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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,

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 6, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Clear thread-follow when config mode ignores it

When --subscribe config --thread-follow is used, the warning says the flag is ignored, but this assignment keeps it enabled. If a channel has any rule that does not require a mention, its merged relay subscription omits #p; after the follow set is populated, mention_exempt can then bypass require_mention on another rule in that channel, causing unmentioned replies to trigger a rule whose configuration explicitly requires mentions. Normalize this field to false outside mentions mode rather than only warning.

Useful? React with 👍 / 👎.

config_path: args.config,
context_message_limit: args.context_message_limit,
max_turns_per_session: args.max_turns_per_session,
Expand Down Expand Up @@ -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,
Expand All @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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(),
Expand Down Expand Up @@ -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,
Expand Down
35 changes: 25 additions & 10 deletions crates/buzz-acp/src/filter.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -370,6 +372,7 @@ pub async fn match_event(
channel_id: uuid::Uuid,
rules: &[SubscriptionRule],
agent_pubkey_hex: &str,
mention_exempt: bool,
) -> Option<MatchedRule> {
let filter_ctx = FilterContext::from_event(event, channel_id);

Expand All @@ -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")
Expand Down Expand Up @@ -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");
}
Expand Down Expand Up @@ -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");
}
Expand All @@ -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");
Expand All @@ -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());
}

Expand Down Expand Up @@ -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");
}

Expand Down Expand Up @@ -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"
Expand All @@ -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");
}
}
41 changes: 40 additions & 1 deletion crates/buzz-acp/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ mod pool_lifecycle;
mod queue;
mod relay;
mod setup_mode;
mod thread_follow;
mod usage;

pub use usage::TurnUsage;
Expand Down Expand Up @@ -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");
Expand Down Expand Up @@ -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 {
Comment on lines 2334 to +2339

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 6, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Learn self-authored roots before branching on ignore-self

When --no-ignore-self is set, this entire branch is skipped, so a self-authored reply in a not-yet-followed thread that does not mention the agent is never inserted into the follow set and later human replies without mentions are dropped. The advertised "has replied there itself" behavior should not depend on whether self events are subsequently ignored; learn the root for every self-authored event first, then conditionally drop it according to ignore_self.

Useful? React with 👍 / 👎.

thread_follow_set
.insert(thread_follow::followable_root(&buzz_event.event));
}
tracing::debug!(channel_id = %buzz_event.channel_id, "dropping self-authored event");
continue;
}
Expand Down Expand Up @@ -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");
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down
1 change: 1 addition & 0 deletions crates/buzz-acp/src/setup_mode.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down
Loading